Apache/nifi
1
0
Fork 0
mirror of https://github.com/apache/nifi.git synced 2025-02-07 18:48:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

[NIFI-11511] Added Provenance route events for all the relationships.

Browse Source 
This closes #7221

Signed-off-by: Mike Thomsen <mthomsen@apache.org>
This commit is contained in:
dan-s1 2023-05-01 21:34:14 +00:00 committed by Mike Thomsen
parent f4850bec90
commit dfff80d3c5
2 changed files with 17 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ValidateJson.java
View File

@ -187,15 +187,18 @@ public class ValidateJson extends AbstractProcessor {
if (errors.isEmpty()) { if (errors.isEmpty()) {
getLogger().debug("JSON {} valid", flowFile); getLogger().debug("JSON {} valid", flowFile);
session.getProvenanceReporter().route(flowFile, REL_VALID);
session.transfer(flowFile, REL_VALID); session.transfer(flowFile, REL_VALID);
} else { } else {
final String validationMessages = errors.toString(); final String validationMessages = errors.toString();
flowFile = session.putAttribute(flowFile, ERROR_ATTRIBUTE_KEY, validationMessages); flowFile = session.putAttribute(flowFile, ERROR_ATTRIBUTE_KEY, validationMessages);
getLogger().warn("JSON {} invalid: Validation Errors {}", flowFile, validationMessages); getLogger().warn("JSON {} invalid: Validation Errors {}", flowFile, validationMessages);
session.getProvenanceReporter().route(flowFile, REL_INVALID);
session.transfer(flowFile, REL_INVALID); session.transfer(flowFile, REL_INVALID);
} }
} catch (final Exception e) { } catch (final Exception e) {
getLogger().error("JSON processing failed {}", flowFile, e); getLogger().error("JSON processing failed {}", flowFile, e);
session.getProvenanceReporter().route(flowFile, REL_FAILURE);
session.transfer(flowFile, REL_FAILURE); session.transfer(flowFile, REL_FAILURE);
} }
} }

14
nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestValidateJson.java
View File

@ -18,6 +18,7 @@ package org.apache.nifi.processors.standard;
import org.apache.commons.io.IOUtils; import org.apache.commons.io.IOUtils;
import org.apache.nifi.processor.Relationship; import org.apache.nifi.processor.Relationship;
import org.apache.nifi.provenance.ProvenanceEventType;
import org.apache.nifi.util.TestRunner; import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners; import org.apache.nifi.util.TestRunners;
import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.BeforeEach;
@ -32,6 +33,7 @@ import java.nio.charset.StandardCharsets;
import java.util.Map; import java.util.Map;
import java.util.Objects; import java.util.Objects;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertThrows;
@ -62,6 +64,8 @@ class TestValidateJson {
runner.assertTransferCount(ValidateJson.REL_VALID, 1); runner.assertTransferCount(ValidateJson.REL_VALID, 1);
assertValidationErrors(ValidateJson.REL_VALID, false); assertValidationErrors(ValidateJson.REL_VALID, false);
assertEquals(1, runner.getProvenanceEvents().size());
assertEquals(ProvenanceEventType.ROUTE, runner.getProvenanceEvents().get(0).getEventType());
} }
@Test @Test
@ -77,6 +81,8 @@ class TestValidateJson {
runner.assertTransferCount(ValidateJson.REL_VALID, 1); runner.assertTransferCount(ValidateJson.REL_VALID, 1);
assertValidationErrors(ValidateJson.REL_VALID, false); assertValidationErrors(ValidateJson.REL_VALID, false);
assertEquals(1, runner.getProvenanceEvents().size());
assertEquals(ProvenanceEventType.ROUTE, runner.getProvenanceEvents().get(0).getEventType());
} }
@Test @Test
@ -93,6 +99,8 @@ class TestValidateJson {
runner.assertTransferCount(ValidateJson.REL_VALID, 1); runner.assertTransferCount(ValidateJson.REL_VALID, 1);
assertValidationErrors(ValidateJson.REL_VALID, false); assertValidationErrors(ValidateJson.REL_VALID, false);
assertEquals(1, runner.getProvenanceEvents().size());
assertEquals(ProvenanceEventType.ROUTE, runner.getProvenanceEvents().get(0).getEventType());
} }
@Test @Test
@ -109,6 +117,8 @@ class TestValidateJson {
runner.assertTransferCount(ValidateJson.REL_VALID, 0); runner.assertTransferCount(ValidateJson.REL_VALID, 0);
assertValidationErrors(ValidateJson.REL_INVALID, true); assertValidationErrors(ValidateJson.REL_INVALID, true);
assertEquals(1, runner.getProvenanceEvents().size());
assertEquals(ProvenanceEventType.ROUTE, runner.getProvenanceEvents().get(0).getEventType());
} }
@Test @Test
@ -125,6 +135,8 @@ class TestValidateJson {
runner.assertTransferCount(ValidateJson.REL_VALID, 0); runner.assertTransferCount(ValidateJson.REL_VALID, 0);
assertValidationErrors(ValidateJson.REL_INVALID, true); assertValidationErrors(ValidateJson.REL_INVALID, true);
assertEquals(1, runner.getProvenanceEvents().size());
assertEquals(ProvenanceEventType.ROUTE, runner.getProvenanceEvents().get(0).getEventType());
} }
@Test @Test
@ -141,6 +153,8 @@ class TestValidateJson {
runner.assertTransferCount(ValidateJson.REL_VALID, 0); runner.assertTransferCount(ValidateJson.REL_VALID, 0);
assertValidationErrors(ValidateJson.REL_FAILURE, false); assertValidationErrors(ValidateJson.REL_FAILURE, false);
assertEquals(1, runner.getProvenanceEvents().size());
assertEquals(ProvenanceEventType.ROUTE, runner.getProvenanceEvents().get(0).getEventType());
} }
@Test @Test