Repository: nifi Updated Branches: refs/heads/develop e5fa76345 -> d328ca0a4
NIFI-829: ensure that the proper provenance events are emitted by InvokeHTTP Project: http://git-wip-us.apache.org/repos/asf/nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/6a7a4d07 Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/6a7a4d07 Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/6a7a4d07 Branch: refs/heads/develop Commit: 6a7a4d0762f7f5fd306f8def019ee904180043d7 Parents: 33279fd Author: Mark Payne <[email protected]> Authored: Mon Aug 10 17:31:44 2015 -0400 Committer: Mark Payne <[email protected]> Committed: Mon Aug 10 17:31:44 2015 -0400 ---------------------------------------------------------------------- .../nifi/processors/standard/InvokeHTTP.java | 10 ++++--- .../processors/standard/TestInvokeHTTP.java | 29 ++++++++++++++++---- 2 files changed, 29 insertions(+), 10 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi/blob/6a7a4d07/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/InvokeHTTP.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/InvokeHTTP.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/InvokeHTTP.java index 82e0573..77412cd 100644 --- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/InvokeHTTP.java +++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/InvokeHTTP.java @@ -331,6 +331,7 @@ public final class InvokeHTTP extends AbstractProcessor { private final ProcessSession session; private final long txId = txIdGenerator.incrementAndGet(); + private final long startNanos = System.nanoTime(); private FlowFile request; private FlowFile response; @@ -482,7 +483,7 @@ public final class InvokeHTTP extends AbstractProcessor { // and the status codes. if (isSuccess()) { // clone the flowfile to capture the response - response = session.clone(request); + response = session.create(request); // write the status attributes response = writeStatusAttributes(response); @@ -495,10 +496,11 @@ public final class InvokeHTTP extends AbstractProcessor { // can potentially be null in edge cases if (is != null) { response = session.importFrom(is, response); - } - // invoke provenance events - session.getProvenanceReporter().receive(response, conn.getURL().toExternalForm()); + // emit provenance event + final long millis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos); + session.getProvenanceReporter().modifyContent(response, "Updated content with data received from " + conn.getURL().toExternalForm(), millis); + } } http://git-wip-us.apache.org/repos/asf/nifi/blob/6a7a4d07/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestInvokeHTTP.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestInvokeHTTP.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestInvokeHTTP.java index acff14b..46cacca 100644 --- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestInvokeHTTP.java +++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestInvokeHTTP.java @@ -16,6 +16,11 @@ */ package org.apache.nifi.processors.standard; +import static org.apache.commons.codec.binary.Base64.encodeBase64; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + import java.io.IOException; import java.io.PrintWriter; import java.io.UnsupportedEncodingException; @@ -23,6 +28,7 @@ import java.nio.charset.StandardCharsets; import java.text.SimpleDateFormat; import java.util.Date; import java.util.HashMap; +import java.util.List; import java.util.Locale; import java.util.Map; @@ -32,6 +38,8 @@ import javax.servlet.http.HttpServletResponse; import org.apache.nifi.flowfile.attributes.CoreAttributes; import org.apache.nifi.processors.standard.InvokeHTTP.Config; +import org.apache.nifi.provenance.ProvenanceEventRecord; +import org.apache.nifi.provenance.ProvenanceEventType; import org.apache.nifi.ssl.StandardSSLContextService; import org.apache.nifi.util.MockFlowFile; import org.apache.nifi.util.TestRunner; @@ -42,12 +50,6 @@ import org.eclipse.jetty.server.handler.AbstractHandler; import org.junit.After; import org.junit.AfterClass; import org.junit.Assert; - -import static org.apache.commons.codec.binary.Base64.encodeBase64; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertTrue; -import static org.junit.Assert.fail; - import org.junit.Before; import org.junit.BeforeClass; import org.junit.Test; @@ -217,6 +219,21 @@ public class TestInvokeHTTP { bundle1.assertAttributeEquals(Config.STATUS_MESSAGE, "OK"); bundle1.assertAttributeEquals("Foo", "Bar"); bundle1.assertAttributeEquals("Content-Type", "text/plain; charset=ISO-8859-1"); + + final List<ProvenanceEventRecord> provEvents = runner.getProvenanceEvents(); + assertEquals(2, provEvents.size()); + boolean forkEvent = false; + boolean contentModEvent = false; + for (final ProvenanceEventRecord event : provEvents) { + if (event.getEventType() == ProvenanceEventType.FORK) { + forkEvent = true; + } else if (event.getEventType() == ProvenanceEventType.CONTENT_MODIFIED) { + contentModEvent = true; + } + } + + assertTrue(forkEvent); + assertTrue(contentModEvent); } @Test
