Skip to content

Commit

Permalink
NIFI-14219 Enabled EL Evaluation for Jolt Specification Path
Browse files Browse the repository at this point in the history
  • Loading branch information
@exceptionfactory
exceptionfactory committed Feb 2, 2025
1 parent 60a8a43 commit 0f58a25
Show file tree
Hide file tree
Showing 2 changed files with 65 additions and 114 deletions.
11 changes: 2 additions & 9 deletions ...-jolt-processors/src/main/java/org/apache/nifi/processors/jolt/AbstractJoltTransform.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -237,15 +237,8 @@ JoltTransform createTransform(final ProcessContext context, final FlowFile flowF
}

private String readTransform(final PropertyValue propertyValue, final FlowFile flowFile) {
final String transform;

if (propertyValue.isExpressionLanguagePresent()) {
transform = propertyValue.evaluateAttributeExpressions(flowFile).getValue();
} else {
transform = readTransform(propertyValue);
}

return transform;
final PropertyValue evaluatedPropertyValue = propertyValue.evaluateAttributeExpressions(flowFile);
return readTransform(evaluatedPropertyValue);
}

String readTransform(final PropertyValue propertyValue) {
Expand Down
168 changes: 63 additions & 105 deletions ...-jolt-processors/src/test/java/org/apache/nifi/processors/jolt/TestJoltTransformJSON.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -205,13 +205,8 @@ void testTransformInputWithChainr(Path specPath, String ignoredDescription) thro
runner.setProperty(JoltTransformJSON.JOLT_SPEC, spec);
runner.enqueue(JSON_INPUT);
runner.run();
runner.assertAllFlowFilesTransferred(JoltTransformJSON.REL_SUCCESS);
final MockFlowFile transformed = runner.getFlowFilesForRelationship(JoltTransformJSON.REL_SUCCESS).getFirst();
transformed.assertAttributeExists(CoreAttributes.MIME_TYPE.key());
transformed.assertAttributeEquals(CoreAttributes.MIME_TYPE.key(), "application/json");
Object transformedJson = JsonUtils.jsonToObject(new ByteArrayInputStream(transformed.toByteArray()));
Object compareJson = JsonUtils.jsonToObject(Files.newInputStream(Paths.get("src/test/resources/TestJoltTransformJson/chainrOutput.json")));
assertTrue(DIFFY.diff(compareJson, transformedJson).isEmpty());

assertTransformedEquals("chainrOutput.json");
}

@Test
Expand All @@ -221,13 +216,8 @@ void testTransformInputWithShiftr() throws IOException {
runner.setProperty(JoltTransformJSON.JOLT_TRANSFORM, JoltTransformStrategy.SHIFTR);
runner.enqueue(JSON_INPUT);
runner.run();
runner.assertAllFlowFilesTransferred(JoltTransformJSON.REL_SUCCESS);
final MockFlowFile transformed = runner.getFlowFilesForRelationship(JoltTransformJSON.REL_SUCCESS).getFirst();
transformed.assertAttributeExists(CoreAttributes.MIME_TYPE.key());
transformed.assertAttributeEquals(CoreAttributes.MIME_TYPE.key(), "application/json");
Object transformedJson = JsonUtils.jsonToObject(new ByteArrayInputStream(transformed.toByteArray()));
Object compareJson = JsonUtils.jsonToObject(Files.newInputStream(Paths.get("src/test/resources/TestJoltTransformJson/shiftrOutput.json")));
assertTrue(DIFFY.diff(compareJson, transformedJson).isEmpty());

assertTransformedEquals("shiftrOutput.json");
}

@Test
Expand All @@ -237,13 +227,23 @@ void testTransformInputWithShiftrFromFile() throws IOException {
runner.setProperty(JoltTransformJSON.JOLT_TRANSFORM, JoltTransformStrategy.SHIFTR);
runner.enqueue(JSON_INPUT);
runner.run();
runner.assertAllFlowFilesTransferred(JoltTransformJSON.REL_SUCCESS);
final MockFlowFile transformed = runner.getFlowFilesForRelationship(JoltTransformJSON.REL_SUCCESS).getFirst();
transformed.assertAttributeExists(CoreAttributes.MIME_TYPE.key());
transformed.assertAttributeEquals(CoreAttributes.MIME_TYPE.key(), "application/json");
Object transformedJson = JsonUtils.jsonToObject(new ByteArrayInputStream(transformed.toByteArray()));
Object compareJson = JsonUtils.jsonToObject(Files.newInputStream(Paths.get("src/test/resources/TestJoltTransformJson/shiftrOutput.json")));
assertTrue(DIFFY.diff(compareJson, transformedJson).isEmpty());

assertTransformedEquals("shiftrOutput.json");
}

@Test
void testTransformInputWithShiftrFromFileExpression() throws IOException {
final String specFilename = "shiftrSpec.json";
final String spec = "./src/test/resources/specs/${filename}";
final Map<String, String> attributes = Map.of(CoreAttributes.FILENAME.key(), specFilename);

runner.setProperty(JoltTransformJSON.JOLT_SPEC, spec);
runner.setProperty(JoltTransformJSON.JOLT_TRANSFORM, JoltTransformStrategy.SHIFTR);

runner.enqueue(JSON_INPUT, attributes);
runner.run();

assertTransformedEquals("shiftrOutput.json");
}

String addAccentedChars(String input) {
Expand All @@ -257,13 +257,8 @@ void testTransformInputWithShiftrAccentedChars() throws IOException {
runner.setProperty(JoltTransformJSON.JOLT_TRANSFORM, JoltTransformStrategy.SHIFTR);
runner.enqueue(addAccentedChars(Files.readString(JSON_INPUT)));
runner.run();
runner.assertAllFlowFilesTransferred(JoltTransformJSON.REL_SUCCESS);
final MockFlowFile transformed = runner.getFlowFilesForRelationship(JoltTransformJSON.REL_SUCCESS).getFirst();
transformed.assertAttributeExists(CoreAttributes.MIME_TYPE.key());
transformed.assertAttributeEquals(CoreAttributes.MIME_TYPE.key(), "application/json");
Object transformedJson = JsonUtils.jsonToObject(new ByteArrayInputStream(transformed.toByteArray()));
Object compareJson = JsonUtils.jsonToObject(Files.newInputStream(Paths.get("src/test/resources/TestJoltTransformJson/shiftrOutput.json")));
assertTrue(DIFFY.diff(compareJson, transformedJson).isEmpty());

assertTransformedEquals("shiftrOutput.json");
}

@Test
Expand All @@ -273,11 +268,8 @@ void testTransformInputWithDefaultr() throws IOException {
runner.setProperty(JoltTransformJSON.JOLT_TRANSFORM, JoltTransformStrategy.DEFAULTR);
runner.enqueue(JSON_INPUT);
runner.run();
runner.assertAllFlowFilesTransferred(JoltTransformJSON.REL_SUCCESS);
final MockFlowFile transformed = runner.getFlowFilesForRelationship(JoltTransformJSON.REL_SUCCESS).getFirst();
Object transformedJson = JsonUtils.jsonToObject(new ByteArrayInputStream(transformed.toByteArray()));
Object compareJson = JsonUtils.jsonToObject(Files.newInputStream(Paths.get("src/test/resources/TestJoltTransformJson/defaultrOutput.json")));
assertTrue(DIFFY.diff(compareJson, transformedJson).isEmpty());

assertTransformedEquals("defaultrOutput.json");
}

@Test
Expand All @@ -287,11 +279,8 @@ void testTransformInputWithRemovr() throws IOException {
runner.setProperty(JoltTransformJSON.JOLT_TRANSFORM, JoltTransformStrategy.REMOVR);
runner.enqueue(JSON_INPUT);
runner.run();
runner.assertAllFlowFilesTransferred(JoltTransformJSON.REL_SUCCESS);
final MockFlowFile transformed = runner.getFlowFilesForRelationship(JoltTransformJSON.REL_SUCCESS).getFirst();
Object transformedJson = JsonUtils.jsonToObject(new ByteArrayInputStream(transformed.toByteArray()));
Object compareJson = JsonUtils.jsonToObject(Files.newInputStream(Paths.get("src/test/resources/TestJoltTransformJson/removrOutput.json")));
assertTrue(DIFFY.diff(compareJson, transformedJson).isEmpty());

assertTransformedEquals("removrOutput.json");
}

@Test
Expand All @@ -301,11 +290,8 @@ void testTransformInputWithCardinality() throws IOException {
runner.setProperty(JoltTransformJSON.JOLT_TRANSFORM, JoltTransformStrategy.CARDINALITY);
runner.enqueue(JSON_INPUT);
runner.run();
runner.assertAllFlowFilesTransferred(JoltTransformJSON.REL_SUCCESS);
final MockFlowFile transformed = runner.getFlowFilesForRelationship(JoltTransformJSON.REL_SUCCESS).getFirst();
Object transformedJson = JsonUtils.jsonToObject(new ByteArrayInputStream(transformed.toByteArray()));
Object compareJson = JsonUtils.jsonToObject(Files.newInputStream(Paths.get("src/test/resources/TestJoltTransformJson/cardrOutput.json")));
assertTrue(DIFFY.diff(compareJson, transformedJson).isEmpty());

assertTransformedEquals("cardrOutput.json");
}

@Test
Expand All @@ -332,11 +318,8 @@ void testTransformInputWithDefaultrExpressionLanguage() throws IOException {
runner.setEnvironmentVariableValue("quota", "5");
runner.enqueue(JSON_INPUT);
runner.run();
runner.assertAllFlowFilesTransferred(JoltTransformJSON.REL_SUCCESS);
final MockFlowFile transformed = runner.getFlowFilesForRelationship(JoltTransformJSON.REL_SUCCESS).getFirst();
Object transformedJson = JsonUtils.jsonToObject(new ByteArrayInputStream(transformed.toByteArray()));
Object compareJson = JsonUtils.jsonToObject(Files.newInputStream(Paths.get("src/test/resources/TestJoltTransformJson/defaultrELOutput.json")));
assertTrue(DIFFY.diff(compareJson, transformedJson).isEmpty());

assertTransformedEquals("defaultrELOutput.json");
}

@Test
Expand All @@ -346,11 +329,8 @@ void testTransformInputWithModifierDefault() throws IOException {
runner.setProperty(JoltTransformJSON.JOLT_TRANSFORM, JoltTransformStrategy.MODIFIER_DEFAULTR);
runner.enqueue(JSON_INPUT);
runner.run();
runner.assertAllFlowFilesTransferred(JoltTransformJSON.REL_SUCCESS);
final MockFlowFile transformed = runner.getFlowFilesForRelationship(JoltTransformJSON.REL_SUCCESS).getFirst();
Object transformedJson = JsonUtils.jsonToObject(new ByteArrayInputStream(transformed.toByteArray()));
Object compareJson = JsonUtils.jsonToObject(Files.newInputStream(Paths.get("src/test/resources/TestJoltTransformJson/modifierDefaultOutput.json")));
assertTrue(DIFFY.diff(compareJson, transformedJson).isEmpty());

assertTransformedEquals("modifierDefaultOutput.json");
}

@Test
Expand All @@ -360,11 +340,8 @@ void testTransformInputWithModifierDefine() throws IOException {
runner.setProperty(JoltTransformJSON.JOLT_TRANSFORM, JoltTransformStrategy.MODIFIER_DEFAULTR);
runner.enqueue(JSON_INPUT);
runner.run();
runner.assertAllFlowFilesTransferred(JoltTransformJSON.REL_SUCCESS);
final MockFlowFile transformed = runner.getFlowFilesForRelationship(JoltTransformJSON.REL_SUCCESS).getFirst();
Object transformedJson = JsonUtils.jsonToObject(new ByteArrayInputStream(transformed.toByteArray()));
Object compareJson = JsonUtils.jsonToObject(Files.newInputStream(Paths.get("src/test/resources/TestJoltTransformJson/modifierDefineOutput.json")));
assertTrue(DIFFY.diff(compareJson, transformedJson).isEmpty());

assertTransformedEquals("modifierDefineOutput.json");
}

@Test
Expand All @@ -374,11 +351,8 @@ void testTransformInputWithModifierOverwrite() throws IOException {
runner.setProperty(JoltTransformJSON.JOLT_TRANSFORM, JoltTransformStrategy.MODIFIER_DEFAULTR);
runner.enqueue(JSON_INPUT);
runner.run();
runner.assertAllFlowFilesTransferred(JoltTransformJSON.REL_SUCCESS);
final MockFlowFile transformed = runner.getFlowFilesForRelationship(JoltTransformJSON.REL_SUCCESS).getFirst();
Object transformedJson = JsonUtils.jsonToObject(new ByteArrayInputStream(transformed.toByteArray()));
Object compareJson = JsonUtils.jsonToObject(Files.newInputStream(Paths.get("src/test/resources/TestJoltTransformJson/modifierOverwriteOutput.json")));
assertTrue(DIFFY.diff(compareJson, transformedJson).isEmpty());

assertTransformedEquals("modifierOverwriteOutput.json");
}

@Test
Expand Down Expand Up @@ -407,13 +381,8 @@ void testTransformInputWithCustomTransformationWithJar() throws IOException {
runner.setProperty(JoltTransformJSON.JOLT_TRANSFORM, JoltTransformStrategy.CUSTOMR);
runner.enqueue(JSON_INPUT);
runner.run();
runner.assertAllFlowFilesTransferred(JoltTransformJSON.REL_SUCCESS);
final MockFlowFile transformed = runner.getFlowFilesForRelationship(JoltTransformJSON.REL_SUCCESS).getFirst();
transformed.assertAttributeExists(CoreAttributes.MIME_TYPE.key());
transformed.assertAttributeEquals(CoreAttributes.MIME_TYPE.key(), "application/json");
Object transformedJson = JsonUtils.jsonToObject(new ByteArrayInputStream(transformed.toByteArray()));
Object compareJson = JsonUtils.jsonToObject(Files.newInputStream(Paths.get("src/test/resources/TestJoltTransformJson/chainrOutput.json")));
assertTrue(DIFFY.diff(compareJson, transformedJson).isEmpty());

assertTransformedEquals("chainrOutput.json");
}

@Test
Expand All @@ -431,13 +400,8 @@ void testExpressionLanguageJarFile() throws IOException {
runner.setEnvironmentVariableValue("CUSTOM_JAR", customJarPath);
runner.enqueue(JSON_INPUT, customSpecs);
runner.run();
runner.assertAllFlowFilesTransferred(JoltTransformJSON.REL_SUCCESS);
final MockFlowFile transformed = runner.getFlowFilesForRelationship(JoltTransformJSON.REL_SUCCESS).getFirst();
transformed.assertAttributeExists(CoreAttributes.MIME_TYPE.key());
transformed.assertAttributeEquals(CoreAttributes.MIME_TYPE.key(), "application/json");
Object transformedJson = JsonUtils.jsonToObject(new ByteArrayInputStream(transformed.toByteArray()));
Object compareJson = JsonUtils.jsonToObject(Files.newInputStream(Paths.get("src/test/resources/TestJoltTransformJson/chainrOutput.json")));
assertTrue(DIFFY.diff(compareJson, transformedJson).isEmpty());

assertTransformedEquals("chainrOutput.json");
}

@Test
Expand All @@ -449,13 +413,8 @@ void testTransformInputWithCustomTransformationWithDir() throws IOException {
runner.setProperty(JoltTransformJSON.JOLT_TRANSFORM, JoltTransformStrategy.CUSTOMR);
runner.enqueue(JSON_INPUT);
runner.run();
runner.assertAllFlowFilesTransferred(JoltTransformJSON.REL_SUCCESS);
final MockFlowFile transformed = runner.getFlowFilesForRelationship(JoltTransformJSON.REL_SUCCESS).getFirst();
transformed.assertAttributeExists(CoreAttributes.MIME_TYPE.key());
transformed.assertAttributeEquals(CoreAttributes.MIME_TYPE.key(), "application/json");
Object transformedJson = JsonUtils.jsonToObject(new ByteArrayInputStream(transformed.toByteArray()));
Object compareJson = JsonUtils.jsonToObject(Files.newInputStream(Paths.get("src/test/resources/TestJoltTransformJson/chainrOutput.json")));
assertTrue(DIFFY.diff(compareJson, transformedJson).isEmpty());

assertTransformedEquals("chainrOutput.json");
}

@Test
Expand All @@ -466,13 +425,8 @@ void testTransformInputWithChainrEmbeddedCustomTransformation() throws IOExcepti
runner.setProperty(JoltTransformJSON.MODULES, customJarPath);
runner.enqueue(JSON_INPUT);
runner.run();
runner.assertAllFlowFilesTransferred(JoltTransformJSON.REL_SUCCESS);
final MockFlowFile transformed = runner.getFlowFilesForRelationship(JoltTransformJSON.REL_SUCCESS).getFirst();
transformed.assertAttributeExists(CoreAttributes.MIME_TYPE.key());
transformed.assertAttributeEquals(CoreAttributes.MIME_TYPE.key(), "application/json");
Object transformedJson = JsonUtils.jsonToObject(new ByteArrayInputStream(transformed.toByteArray()));
Object compareJson = JsonUtils.jsonToObject(Files.newInputStream(Paths.get("src/test/resources/TestJoltTransformJson/chainrOutput.json")));
assertTrue(DIFFY.diff(compareJson, transformedJson).isEmpty());

assertTransformedEquals("chainrOutput.json");
}

@Test
Expand All @@ -485,13 +439,8 @@ void testTransformInputCustomTransformationIgnored() throws IOException {
runner.setProperty(JoltTransformJSON.JOLT_TRANSFORM, JoltTransformStrategy.DEFAULTR);
runner.enqueue(JSON_INPUT);
runner.run();
runner.assertAllFlowFilesTransferred(JoltTransformJSON.REL_SUCCESS);
final MockFlowFile transformed = runner.getFlowFilesForRelationship(JoltTransformJSON.REL_SUCCESS).getFirst();
transformed.assertAttributeExists(CoreAttributes.MIME_TYPE.key());
transformed.assertAttributeEquals(CoreAttributes.MIME_TYPE.key(), "application/json");
Object transformedJson = JsonUtils.jsonToObject(new ByteArrayInputStream(transformed.toByteArray()));
Object compareJson = JsonUtils.jsonToObject(Files.newInputStream(Paths.get("src/test/resources/TestJoltTransformJson/defaultrOutput.json")));
assertTrue(DIFFY.diff(compareJson, transformedJson).isEmpty());

assertTransformedEquals("defaultrOutput.json");
}

@Test
Expand All @@ -503,13 +452,8 @@ void testJoltSpecEL() throws IOException {
"{\"RatingRange\":5,\"rating\":{\"*\":{\"MaxLabel\":\"High\",\"MinLabel\":\"Low\",\"DisplayType\":\"NORMAL\"}}}");
runner.enqueue(JSON_INPUT, attributes);
runner.run();
runner.assertAllFlowFilesTransferred(JoltTransformJSON.REL_SUCCESS);
final MockFlowFile transformed = runner.getFlowFilesForRelationship(JoltTransformJSON.REL_SUCCESS).getFirst();
transformed.assertAttributeExists(CoreAttributes.MIME_TYPE.key());
transformed.assertAttributeEquals(CoreAttributes.MIME_TYPE.key(), "application/json");
Object transformedJson = JsonUtils.jsonToObject(new ByteArrayInputStream(transformed.toByteArray()));
Object compareJson = JsonUtils.jsonToObject(Files.newInputStream(Paths.get("src/test/resources/TestJoltTransformJson/defaultrOutput.json")));
assertTrue(DIFFY.diff(compareJson, transformedJson).isEmpty());

assertTransformedEquals("defaultrOutput.json");
}

@Test
Expand All @@ -520,6 +464,20 @@ void testJoltSpecInvalidEL() throws IOException {
runner.assertNotValid();
}

private void assertTransformedEquals(final String expectedOutputFilename) throws IOException {
runner.assertAllFlowFilesTransferred(JoltTransformJSON.REL_SUCCESS);

final MockFlowFile transformed = runner.getFlowFilesForRelationship(JoltTransformJSON.REL_SUCCESS).getFirst();
transformed.assertAttributeExists(CoreAttributes.MIME_TYPE.key());
transformed.assertAttributeEquals(CoreAttributes.MIME_TYPE.key(), "application/json");

final Object transformedJson = JsonUtils.jsonToObject(new ByteArrayInputStream(transformed.toByteArray()));

final String compareOutputPath = "src/test/resources/TestJoltTransformJson/%s".formatted(expectedOutputFilename);
final Object compareJson = JsonUtils.jsonToObject(Files.newInputStream(Paths.get(compareOutputPath)));
assertTrue(DIFFY.diff(compareJson, transformedJson).isEmpty());
}

private static Stream<Arguments> getChainrArguments() {
return Stream.of(
Arguments.of(Paths.get(CHAINR_SPEC_PATH), "has no single line comments"),
Expand Down

0 comments on commit 0f58a25

Please sign in to comment.