You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by mc...@apache.org on 2015/03/21 00:40:41 UTC
[2/2] incubator-nifi git commit: Squashed commit of the following:
Squashed commit of the following:
commit e88ed13d8d7221f5a91588f553d039d3917494be
Merge: 93b361e 3f36236
Author: Mark Payne <ma...@hotmail.com>
Date: Fri Mar 20 09:11:39 2015 -0400
Merge branch 'inputstream-callback-protection' of https://github.com/rowolabi/incubator-nifi into develop
commit 93b361e69bd046d6b2e0f561d8b2231a4505b6a9
Merge: c9eb237 a6740a6
Author: Mark Payne <ma...@hotmail.com>
Date: Thu Mar 19 11:49:11 2015 -0400
Merge branch 'develop' of http://git-wip-us.apache.org/repos/asf/incubator-nifi into develop
commit c9eb237895a94a27fc6f760b82b16ac3e2cdab95
Author: Mark Payne <ma...@hotmail.com>
Date: Thu Mar 19 11:40:01 2015 -0400
NIFI-443: Removed stopFunnel from ProcessGroup because we don't want to allow it
commit 8b911c5aab2a4b8283510a3423e3c8962a533b96
Author: Mark Payne <ma...@hotmail.com>
Date: Thu Mar 19 11:04:46 2015 -0400
NIFI-443: Always start funnels when added to process group, even when autoResumeState is false
commit a6740a6e2c87f4c994d305db55c0777dc4f99976
Author: joewitt <jo...@apache.org>
Date: Thu Mar 19 01:21:32 2015 -0400
NIFI-399 addressed items in the ticket
commit ad18853b589d80331e2f4574bce35d79bce09c28
Author: joewitt <jo...@apache.org>
Date: Wed Mar 18 10:59:13 2015 -0400
NIFI-399 initial port
commit 3f3623647367421cd8eb318668144aca1afb9bf4
Author: Bobby Owolabi <bo...@gmail.com>
Date: Thu Mar 19 01:43:17 2015 -0400
NIFI-396 reverting accidentially modified whitespace in TestStandardProcessSession
commit cd183be4410bdc88de7d3a0026452ab62eb10621
Author: Bobby Owolabi <bo...@gmail.com>
Date: Thu Mar 19 01:18:22 2015 -0400
NIFI-396 updated the javadocs of ProcessSession to reflect that a FlowFileAccessExcpetion will be thrown if an Input/Output Stream is attempted to be accessed after the callback is executed.
commit e2760f8c980583d285137134e05c435c930fb4d2
Author: Bobby Owolabi <bo...@gmail.com>
Date: Thu Mar 19 00:54:24 2015 -0400
NIFI-396 added a DisableOnCloseInputStream class; modified StandardProcessSession to prevent access of the Input/OutputStreams after callbacks have been executed; updated tests
commit 7272d0df58c23d099809bf96993b55d73c617476
Author: Bobby Owolabi <bo...@gmail.com>
Date: Wed Mar 18 23:30:57 2015 -0400
NIFI-396 created tests to demonstrate the situations where the ProcessSession throws an Exception and where it doesn't after it returns from the callback
commit eb5ec703ba0d5c188822a37f6d7eed14af56a594
Author: Oscar de la Pena <od...@exist.com>
Date: Thu Mar 19 10:10:09 2015 +0800
Fixes incorrect messages count in Provenance reporter. Adds Unit test to verify fix
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/3410197b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/3410197b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/3410197b
Branch: refs/heads/NIFI-353
Commit: 3410197b7d88444de5c6f74622d67a1b0cc39e6a
Parents: b1873d8
Author: Matt Gilman <ma...@gmail.com>
Authored: Fri Mar 20 19:40:02 2015 -0400
Committer: Matt Gilman <ma...@gmail.com>
Committed: Fri Mar 20 19:40:02 2015 -0400
----------------------------------------------------------------------
.../apache/nifi/processor/ProcessSession.java | 40 ++-
.../nifi/processor/util/StandardValidators.java | 22 ++
.../processor/util/TestStandardValidators.java | 31 ++
.../org/apache/nifi/groups/ProcessGroup.java | 44 ++-
.../apache/nifi/controller/FlowController.java | 71 +++--
.../nifi/controller/StandardFlowService.java | 21 +-
.../controller/StandardFlowSynchronizer.java | 6 +-
.../repository/StandardProcessSession.java | 11 +-
.../io/DisableOnCloseInputStream.java | 93 ++++++
.../nifi/groups/StandardProcessGroup.java | 58 +---
.../repository/TestStandardProcessSession.java | 149 +++++++++
.../ApplicationStartupContextListener.java | 6 +-
.../apache/nifi/processors/kafka/PutKafka.java | 90 +++---
.../nifi/processors/kafka/TestPutKafka.java | 76 ++++-
.../standard/EvaluateRegularExpression.java | 7 +-
.../nifi/processors/standard/ExtractText.java | 313 ++++++++++++++++++
.../org.apache.nifi.processor.Processor | 1 +
.../standard/TestEvaluateRegularExpression.java | 319 -------------------
.../processors/standard/TestExtractText.java | 314 ++++++++++++++++++
19 files changed, 1169 insertions(+), 503 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/3410197b/nifi/nifi-api/src/main/java/org/apache/nifi/processor/ProcessSession.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-api/src/main/java/org/apache/nifi/processor/ProcessSession.java b/nifi/nifi-api/src/main/java/org/apache/nifi/processor/ProcessSession.java
index d3de916..7b855f2 100644
--- a/nifi/nifi-api/src/main/java/org/apache/nifi/processor/ProcessSession.java
+++ b/nifi/nifi-api/src/main/java/org/apache/nifi/processor/ProcessSession.java
@@ -484,6 +484,9 @@ public interface ProcessSession {
/**
* Executes the given callback against the contents corresponding to the
* given FlowFile.
+ *
+ * <i>Note</i>: The OutputStream provided to the given OutputStreamCallback
+ * will not be accessible once this method has completed its execution.
*
* @param source
* @param reader
@@ -498,9 +501,11 @@ public interface ProcessSession {
* destroyed, and the session is automatically rolled back and what is left
* of the FlowFile is destroyed.
* @throws FlowFileAccessException if some IO problem occurs accessing
- * FlowFile content
+ * FlowFile content; if an attempt is made to access the InputStream
+ * provided to the given InputStreamCallback after this method completed its
+ * execution
*/
- void read(FlowFile source, InputStreamCallback reader);
+ void read(FlowFile source, InputStreamCallback reader) throws FlowFileAccessException;
/**
* Combines the content of all given source FlowFiles into a single given
@@ -560,7 +565,10 @@ public interface ProcessSession {
/**
* Executes the given callback against the content corresponding to the
- * given FlowFile
+ * given FlowFile.
+ *
+ * <i>Note</i>: The OutputStream provided to the given OutputStreamCallback
+ * will not be accessible once this method has completed its execution.
*
* @param source
* @param writer
@@ -576,13 +584,19 @@ public interface ProcessSession {
* destroyed, and the session is automatically rolled back and what is left
* of the FlowFile is destroyed.
* @throws FlowFileAccessException if some IO problem occurs accessing
- * FlowFile content
+ * FlowFile content; if an attempt is made to access the OutputStream
+ * provided to the given OutputStreamCallaback after this method completed
+ * its execution
*/
- FlowFile write(FlowFile source, OutputStreamCallback writer);
+ FlowFile write(FlowFile source, OutputStreamCallback writer) throws FlowFileAccessException;
/**
* Executes the given callback against the content corresponding to the
- * given flow file
+ * given flow file.
+ *
+ * <i>Note</i>: The InputStream & OutputStream provided to the given
+ * StreamCallback will not be accessible once this method has completed its
+ * execution.
*
* @param source
* @param writer
@@ -598,20 +612,28 @@ public interface ProcessSession {
* destroyed, and the session is automatically rolled back and what is left
* of the FlowFile is destroyed.
* @throws FlowFileAccessException if some IO problem occurs accessing
- * FlowFile content
+ * FlowFile content; if an attempt is made to access the InputStream or
+ * OutputStream provided to the given StreamCallback after this method
+ * completed its execution
*/
- FlowFile write(FlowFile source, StreamCallback writer);
+ FlowFile write(FlowFile source, StreamCallback writer) throws FlowFileAccessException;
/**
* Executes the given callback against the content corresponding to the
* given FlowFile, such that any data written to the OutputStream of the
* content will be appended to the end of FlowFile.
+ *
+ * <i>Note</i>: The OutputStream provided to the given OutputStreamCallback
+ * will not be accessible once this method has completed its execution.
*
* @param source
* @param writer
* @return
+ * @throws FlowFileAccessException if an attempt is made to access the
+ * OutputStream provided to the given OutputStreamCallaback after this method
+ * completed its execution
*/
- FlowFile append(FlowFile source, OutputStreamCallback writer);
+ FlowFile append(FlowFile source, OutputStreamCallback writer) throws FlowFileAccessException;
/**
* Writes to the given FlowFile all content from the given content path.
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/3410197b/nifi/nifi-commons/nifi-processor-utilities/src/main/java/org/apache/nifi/processor/util/StandardValidators.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-commons/nifi-processor-utilities/src/main/java/org/apache/nifi/processor/util/StandardValidators.java b/nifi/nifi-commons/nifi-processor-utilities/src/main/java/org/apache/nifi/processor/util/StandardValidators.java
index 10748fe..c9ae609 100644
--- a/nifi/nifi-commons/nifi-processor-utilities/src/main/java/org/apache/nifi/processor/util/StandardValidators.java
+++ b/nifi/nifi-commons/nifi-processor-utilities/src/main/java/org/apache/nifi/processor/util/StandardValidators.java
@@ -287,6 +287,28 @@ public class StandardValidators {
return createAttributeExpressionLanguageValidator(expectedResultType, true);
}
+ public static Validator createDataSizeBoundsValidator(final long minBytesInclusive, final long maxBytesInclusive) {
+ return new Validator() {
+
+ @Override
+ public ValidationResult validate(final String subject, final String input, final ValidationContext context) {
+ final ValidationResult vr = DATA_SIZE_VALIDATOR.validate(subject, input, context);
+ if(!vr.isValid()){
+ return vr;
+ }
+ final long dataSizeBytes = DataUnit.parseDataSize(input, DataUnit.B).longValue();
+ if(dataSizeBytes < minBytesInclusive){
+ return new ValidationResult.Builder().subject(subject).input(input).valid(false).explanation("Cannot be smaller than " + minBytesInclusive + " bytes").build();
+ }
+ if(dataSizeBytes > maxBytesInclusive){
+ return new ValidationResult.Builder().subject(subject).input(input).valid(false).explanation("Cannot be larger than " + maxBytesInclusive + " bytes").build();
+ }
+ return new ValidationResult.Builder().subject(subject).input(input).valid(true).build();
+ }
+ };
+
+ }
+
public static Validator createRegexMatchingValidator(final Pattern pattern) {
return new Validator() {
@Override
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/3410197b/nifi/nifi-commons/nifi-processor-utilities/src/test/java/org/apache/nifi/processor/util/TestStandardValidators.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-commons/nifi-processor-utilities/src/test/java/org/apache/nifi/processor/util/TestStandardValidators.java b/nifi/nifi-commons/nifi-processor-utilities/src/test/java/org/apache/nifi/processor/util/TestStandardValidators.java
index 2ae50c9..70b8d21 100644
--- a/nifi/nifi-commons/nifi-processor-utilities/src/test/java/org/apache/nifi/processor/util/TestStandardValidators.java
+++ b/nifi/nifi-commons/nifi-processor-utilities/src/test/java/org/apache/nifi/processor/util/TestStandardValidators.java
@@ -51,4 +51,35 @@ public class TestStandardValidators {
vr = val.validate("TimePeriodTest", "1 sec", null);
assertTrue(vr.isValid());
}
+
+ @Test
+ public void testDataSizeBoundsValidator() {
+ Validator val = StandardValidators.createDataSizeBoundsValidator(100, 1000);
+ ValidationResult vr;
+
+ vr = val.validate("DataSizeBounds", "5 GB", null);
+ assertFalse(vr.isValid());
+
+ vr = val.validate("DataSizeBounds", "0 B", null);
+ assertFalse(vr.isValid());
+
+ vr = val.validate("DataSizeBounds", "99 B", null);
+ assertFalse(vr.isValid());
+
+ vr = val.validate("DataSizeBounds", "100 B", null);
+ assertTrue(vr.isValid());
+
+ vr = val.validate("DataSizeBounds", "999 B", null);
+ assertTrue(vr.isValid());
+
+ vr = val.validate("DataSizeBounds", "1000 B", null);
+ assertTrue(vr.isValid());
+
+ vr = val.validate("DataSizeBounds", "1001 B", null);
+ assertFalse(vr.isValid());
+
+ vr = val.validate("DataSizeBounds", "water", null);
+ assertFalse(vr.isValid());
+
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/3410197b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/ProcessGroup.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/ProcessGroup.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/ProcessGroup.java
index 61be59c..b898638 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/ProcessGroup.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/ProcessGroup.java
@@ -118,7 +118,7 @@ public interface ProcessGroup {
void stopProcessing();
/**
- * Starts the given Processor
+ * Enables the given Processor
*
* @param processor the processor to start
* @throws IllegalStateException if the processor is not valid, or is
@@ -127,25 +127,19 @@ public interface ProcessGroup {
void enableProcessor(ProcessorNode processor);
/**
- * Starts the given Input Port
+ * Enables the given Input Port
*
* @param port
*/
void enableInputPort(Port port);
/**
- * Starts the given Output Port
+ * Enables the given Output Port
*
* @param port
*/
void enableOutputPort(Port port);
- /**
- * Starts the given Funnel
- *
- * @param funnel
- */
- void enableFunnel(Funnel funnel);
/**
* Starts the given Processor
@@ -198,15 +192,9 @@ public interface ProcessGroup {
*/
void stopOutputPort(Port port);
- /**
- * Stops the given Funnel
- *
- * @param processor
- */
- void stopFunnel(Funnel funnel);
/**
- * Starts the given Processor
+ * Disables the given Processor
*
* @param processor the processor to start
* @throws IllegalStateException if the processor is not valid, or is
@@ -215,25 +203,19 @@ public interface ProcessGroup {
void disableProcessor(ProcessorNode processor);
/**
- * Starts the given Input Port
+ * Disables the given Input Port
*
* @param port
*/
void disableInputPort(Port port);
/**
- * Starts the given Output Port
+ * Disables the given Output Port
*
* @param port
*/
void disableOutputPort(Port port);
- /**
- * Starts the given Funnel
- *
- * @param funnel
- */
- void disableFunnel(Funnel funnel);
/**
* Indicates that the Flow is being shutdown; allows cleanup of resources
@@ -618,11 +600,23 @@ public interface ProcessGroup {
Port getOutputPortByName(String name);
/**
- * Adds the given funnel to this ProcessGroup
+ * Adds the given funnel to this ProcessGroup and starts it. While other components
+ * do not automatically start, the funnel does by default because it is intended to be
+ * more of a notional component that users are unable to explicitly start and stop.
+ * However, there is an override available in {@link #addFunnel(Funnel, boolean)} because
+ * we may need to avoid starting the funnel on restart until the flow is completely
+ * initialized.
*
* @param funnel
*/
void addFunnel(Funnel funnel);
+
+ /**
+ * Adds the given funnel to this ProcessGroup and optionally starts the funnel.
+ * @param funnel
+ * @param autoStart
+ */
+ void addFunnel(Funnel funnel, boolean autoStart);
/**
* Returns a Set of all Funnels that belong to this ProcessGroup
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/3410197b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
index 54f0807..06ef203 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
@@ -576,40 +576,57 @@ public class FlowController implements EventAccess, ControllerServiceProvider, H
* flag of true to now start
* </p>
*/
- public void startDelayed() {
+ public void onFlowInitialized(final boolean startDelayedComponents) {
writeLock.lock();
try {
- LOG.info("Starting {} processors/ports/funnels", (startConnectablesAfterInitialization.size() + startRemoteGroupPortsAfterInitialization.size()));
- for (final Connectable connectable : startConnectablesAfterInitialization) {
- if (connectable.getScheduledState() == ScheduledState.DISABLED) {
- continue;
+ if ( startDelayedComponents ) {
+ LOG.info("Starting {} processors/ports/funnels", (startConnectablesAfterInitialization.size() + startRemoteGroupPortsAfterInitialization.size()));
+ for (final Connectable connectable : startConnectablesAfterInitialization) {
+ if (connectable.getScheduledState() == ScheduledState.DISABLED) {
+ continue;
+ }
+
+ try {
+ if (connectable instanceof ProcessorNode) {
+ connectable.getProcessGroup().startProcessor((ProcessorNode) connectable);
+ } else {
+ startConnectable(connectable);
+ }
+ } catch (final Throwable t) {
+ LOG.error("Unable to start {} due to {}", new Object[]{connectable, t});
+ }
}
-
- try {
- if (connectable instanceof ProcessorNode) {
- connectable.getProcessGroup().startProcessor((ProcessorNode) connectable);
- } else {
- startConnectable(connectable);
+
+ startConnectablesAfterInitialization.clear();
+
+ int startedTransmitting = 0;
+ for (final RemoteGroupPort remoteGroupPort : startRemoteGroupPortsAfterInitialization) {
+ try {
+ remoteGroupPort.getRemoteProcessGroup().startTransmitting(remoteGroupPort);
+ startedTransmitting++;
+ } catch (final Throwable t) {
+ LOG.error("Unable to start transmitting with {} due to {}", new Object[]{remoteGroupPort, t});
}
- } catch (final Throwable t) {
- LOG.error("Unable to start {} due to {}", new Object[]{connectable, t});
}
- }
-
- startConnectablesAfterInitialization.clear();
-
- int startedTransmitting = 0;
- for (final RemoteGroupPort remoteGroupPort : startRemoteGroupPortsAfterInitialization) {
- try {
- remoteGroupPort.getRemoteProcessGroup().startTransmitting(remoteGroupPort);
- startedTransmitting++;
- } catch (final Throwable t) {
- LOG.error("Unable to start transmitting with {} due to {}", new Object[]{remoteGroupPort, t});
+
+ LOG.info("Started {} Remote Group Ports transmitting", startedTransmitting);
+ startRemoteGroupPortsAfterInitialization.clear();
+ } else {
+ // We don't want to start all of the delayed components. However, funnels need to be started anyway
+ // because we don't provide users the ability to start or stop them - they are just notional.
+ for (final Connectable connectable : startConnectablesAfterInitialization) {
+ try {
+ if (connectable instanceof Funnel) {
+ startConnectable(connectable);
+ }
+ } catch (final Throwable t) {
+ LOG.error("Unable to start {} due to {}", new Object[]{connectable, t});
+ }
}
+
+ startConnectablesAfterInitialization.clear();
+ startRemoteGroupPortsAfterInitialization.clear();
}
-
- LOG.info("Started {} Remote Group Ports transmitting", startedTransmitting);
- startRemoteGroupPortsAfterInitialization.clear();
} finally {
writeLock.unlock();
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/3410197b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowService.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowService.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowService.java
index d459b00..64ce5c4 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowService.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowService.java
@@ -423,18 +423,15 @@ public class StandardFlowService implements FlowService, ProtocolHandler {
*/
controller.startHeartbeating();
- // if configured, start all components
- if (autoResumeState) {
- try {
- controller.startDelayed();
- } catch (final Exception ex) {
- logger.warn("Unable to start all processors due to invalid flow configuration.");
- if (logger.isDebugEnabled()) {
- logger.warn(StringUtils.EMPTY, ex);
- }
+ // notify controller that flow is initialized
+ try {
+ controller.onFlowInitialized(autoResumeState);
+ } catch (final Exception ex) {
+ logger.warn("Unable to start all processors due to invalid flow configuration.");
+ if (logger.isDebugEnabled()) {
+ logger.warn(StringUtils.EMPTY, ex);
}
}
-
} else {
try {
loadFromConnectionResponse(response);
@@ -732,9 +729,7 @@ public class StandardFlowService implements FlowService, ProtocolHandler {
controller.setPrimary(response.isPrimary());
// start the processors as indicated by the dataflow
- if (dataFlow.isAutoStartProcessors()) {
- controller.startDelayed();
- }
+ controller.onFlowInitialized(dataFlow.isAutoStartProcessors());
loadTemplates(dataFlow.getTemplates());
loadSnippets(dataFlow.getSnippets());
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/3410197b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java
index b60d187..05a8f01 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java
@@ -692,7 +692,11 @@ public class StandardFlowSynchronizer implements FlowSynchronizer {
final FunnelDTO funnelDTO = FlowFromDOMFactory.getFunnel(funnelElement);
final Funnel funnel = controller.createFunnel(funnelDTO.getId());
funnel.setPosition(toPosition(funnelDTO.getPosition()));
- processGroup.addFunnel(funnel);
+
+ // Since this is called during startup, we want to add the funnel without enabling it
+ // and then tell the controller to enable it. This way, if the controller is not fully
+ // initialized, the starting of the funnel is delayed until the controller is ready.
+ processGroup.addFunnel(funnel, false);
controller.startConnectable(funnel);
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/3410197b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
index 8d2e456..e5cd03e 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
@@ -46,6 +46,7 @@ import org.apache.nifi.controller.ProcessorNode;
import org.apache.nifi.controller.repository.claim.ContentClaim;
import org.apache.nifi.controller.repository.io.ByteCountingInputStream;
import org.apache.nifi.controller.repository.io.ByteCountingOutputStream;
+import org.apache.nifi.controller.repository.io.DisableOnCloseInputStream;
import org.apache.nifi.controller.repository.io.DisableOnCloseOutputStream;
import org.apache.nifi.controller.repository.io.FlowFileAccessInputStream;
import org.apache.nifi.controller.repository.io.FlowFileAccessOutputStream;
@@ -1735,7 +1736,8 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
try (final InputStream rawIn = getInputStream(source, record.getCurrentClaim(), record.getCurrentClaimOffset());
final InputStream limitedIn = new LimitedInputStream(rawIn, source.getSize());
- final ByteCountingInputStream countingStream = new ByteCountingInputStream(limitedIn, this.bytesRead)) {
+ final InputStream disableOnCloseIn = new DisableOnCloseInputStream(limitedIn);
+ final ByteCountingInputStream countingStream = new ByteCountingInputStream(disableOnCloseIn, this.bytesRead)) {
// We want to differentiate between IOExceptions thrown by the repository and IOExceptions thrown from
// Processor code. As a result, as have the FlowFileAccessInputStream that catches IOException from the repository
@@ -2180,9 +2182,10 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
try (final InputStream rawIn = getInputStream(source, currClaim, record.getCurrentClaimOffset());
final InputStream limitedIn = new LimitedInputStream(rawIn, source.getSize());
- final InputStream countingIn = new ByteCountingInputStream(limitedIn, bytesRead);
- final OutputStream disableOnClose = new DisableOnCloseOutputStream(currentWriteClaimStream);
- final OutputStream countingOut = new ByteCountingOutputStream(disableOnClose, writtenHolder)) {
+ final InputStream disableOnCloseIn = new DisableOnCloseInputStream(limitedIn);
+ final InputStream countingIn = new ByteCountingInputStream(disableOnCloseIn, bytesRead);
+ final OutputStream disableOnCloseOut = new DisableOnCloseOutputStream(currentWriteClaimStream);
+ final OutputStream countingOut = new ByteCountingOutputStream(disableOnCloseOut, writtenHolder)) {
recursionSet.add(source);
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/3410197b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/io/DisableOnCloseInputStream.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/io/DisableOnCloseInputStream.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/io/DisableOnCloseInputStream.java
new file mode 100644
index 0000000..ddcf6c9
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/io/DisableOnCloseInputStream.java
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.controller.repository.io;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * Wraps an existing InputStream, so that when {@link InputStream#close()} is
+ * called, the underlying InputStream is NOT closed but this InputStream can no
+ * longer be written to
+ */
+public class DisableOnCloseInputStream extends InputStream {
+
+ private final InputStream wrapped;
+ private boolean closed = false;
+
+ public DisableOnCloseInputStream(final InputStream wrapped) {
+ this.wrapped = wrapped;
+ }
+
+ @Override
+ public int read() throws IOException {
+ checkClosed();
+ return wrapped.read();
+ }
+
+ @Override
+ public int read(byte[] b) throws IOException {
+ checkClosed();
+ return wrapped.read(b);
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ checkClosed();
+ return wrapped.read(b, off, len);
+ }
+
+ @Override
+ public long skip(long n) throws IOException {
+ checkClosed();
+ return wrapped.skip(n);
+ }
+
+ @Override
+ public int available() throws IOException {
+ return wrapped.available();
+ }
+
+ private void checkClosed() throws IOException {
+ if (closed) {
+ throw new IOException("Stream is closed");
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ closed = true;
+ }
+
+ @Override
+ public void mark(int readlimit) {
+ if (closed == false) {
+ wrapped.mark(readlimit);
+ }
+ }
+
+ @Override
+ public synchronized void reset() throws IOException {
+ checkClosed();
+ wrapped.reset();
+ }
+
+ @Override
+ public boolean markSupported() {
+ return wrapped.markSupported();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/3410197b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
index 856ccc1..3cd5853 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
@@ -1036,9 +1036,7 @@ public final class StandardProcessGroup implements ProcessGroup {
}
final ScheduledState state = funnel.getScheduledState();
- if (state == ScheduledState.DISABLED) {
- throw new IllegalStateException("Funnel is disabled");
- } else if (state == ScheduledState.RUNNING) {
+ if (state == ScheduledState.RUNNING) {
return;
}
scheduler.startFunnel(funnel);
@@ -1110,8 +1108,7 @@ public final class StandardProcessGroup implements ProcessGroup {
}
}
- @Override
- public void stopFunnel(final Funnel funnel) {
+ private void stopFunnel(final Funnel funnel) {
readLock.lock();
try {
if (!funnels.containsKey(funnel.getIdentifier())) {
@@ -1132,27 +1129,6 @@ public final class StandardProcessGroup implements ProcessGroup {
}
@Override
- public void enableFunnel(final Funnel funnel) {
- readLock.lock();
- try {
- if (!funnels.containsKey(funnel.getIdentifier())) {
- throw new IllegalStateException("No Funnel with ID " + funnel.getIdentifier() + " belongs to this Process Group");
- }
-
- final ScheduledState state = funnel.getScheduledState();
- if (state == ScheduledState.STOPPED) {
- return;
- } else if (state == ScheduledState.RUNNING) {
- throw new IllegalStateException("Funnel is currently running");
- }
-
- scheduler.enableFunnel(funnel);
- } finally {
- readLock.unlock();
- }
- }
-
- @Override
public void enableInputPort(final Port port) {
readLock.lock();
try {
@@ -1215,26 +1191,6 @@ public final class StandardProcessGroup implements ProcessGroup {
}
}
- @Override
- public void disableFunnel(final Funnel funnel) {
- readLock.lock();
- try {
- if (!funnels.containsKey(funnel.getIdentifier())) {
- throw new IllegalStateException("No Funnel with ID " + funnel.getIdentifier() + " belongs to this Process Group");
- }
-
- final ScheduledState state = funnel.getScheduledState();
- if (state == ScheduledState.DISABLED) {
- return;
- } else if (state == ScheduledState.RUNNING) {
- throw new IllegalStateException("Funnel is currently running");
- }
-
- scheduler.disableFunnel(funnel);
- } finally {
- readLock.unlock();
- }
- }
@Override
public void disableInputPort(final Port port) {
@@ -1546,8 +1502,14 @@ public final class StandardProcessGroup implements ProcessGroup {
return null;
}
+
@Override
public void addFunnel(final Funnel funnel) {
+ addFunnel(funnel, true);
+ }
+
+ @Override
+ public void addFunnel(final Funnel funnel, final boolean autoStart) {
writeLock.lock();
try {
final Funnel existing = funnels.get(requireNonNull(funnel).getIdentifier());
@@ -1557,6 +1519,10 @@ public final class StandardProcessGroup implements ProcessGroup {
funnel.setProcessGroup(this);
funnels.put(funnel.getIdentifier(), funnel);
+
+ if ( autoStart ) {
+ startFunnel(funnel);
+ }
} finally {
writeLock.unlock();
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/3410197b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestStandardProcessSession.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestStandardProcessSession.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestStandardProcessSession.java
index 1ff63c5..ca68725 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestStandardProcessSession.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestStandardProcessSession.java
@@ -56,6 +56,7 @@ import org.apache.nifi.controller.repository.claim.StandardContentClaimManager;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.FlowFileAccessException;
import org.apache.nifi.processor.exception.MissingFlowFileException;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.InputStreamCallback;
@@ -65,6 +66,7 @@ import org.apache.nifi.provenance.MockProvenanceEventRepository;
import org.apache.nifi.provenance.ProvenanceEventRecord;
import org.apache.nifi.provenance.ProvenanceEventRepository;
import org.apache.nifi.provenance.ProvenanceEventType;
+import org.apache.nifi.util.ObjectHolder;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@@ -235,6 +237,147 @@ public class TestStandardProcessSession {
assertEquals(0, contentRepo.getExistingClaims().size());
}
+ private void assertDisabled(final OutputStream outputStream) {
+ try {
+ outputStream.write(new byte[0]);
+ Assert.fail("Expected OutputStream to be disabled; was able to call write(byte[])");
+ } catch (final Exception ex) {
+ Assert.assertEquals(FlowFileAccessException.class, ex.getClass());
+ }
+ try {
+ outputStream.write(0);
+ Assert.fail("Expected OutputStream to be disabled; was able to call write(int)");
+ } catch (final Exception ex) {
+ Assert.assertEquals(FlowFileAccessException.class, ex.getClass());
+ }
+ try {
+ outputStream.write(new byte[0], 0, 0);
+ Assert.fail("Expected OutputStream to be disabled; was able to call write(byte[], int, int)");
+ } catch (final Exception ex) {
+ Assert.assertEquals(FlowFileAccessException.class, ex.getClass());
+ }
+ }
+
+ private void assertDisabled(final InputStream inputStream) {
+ try {
+ inputStream.read();
+ Assert.fail("Expected InputStream to be disabled; was able to call read()");
+ } catch (final Exception ex) {
+ Assert.assertEquals(FlowFileAccessException.class, ex.getClass());
+ }
+ try {
+ inputStream.read(new byte[0]);
+ Assert.fail("Expected InputStream to be disabled; was able to call read(byte[])");
+ } catch (final Exception ex) {
+ Assert.assertEquals(FlowFileAccessException.class, ex.getClass());
+ }
+ try {
+ inputStream.read(new byte[0], 0, 0);
+ Assert.fail("Expected InputStream to be disabled; was able to call read(byte[], int, int)");
+ } catch (final Exception ex) {
+ Assert.assertEquals(FlowFileAccessException.class, ex.getClass());
+ }
+ try {
+ inputStream.reset();
+ Assert.fail("Expected InputStream to be disabled; was able to call reset()");
+ } catch (final Exception ex) {
+ Assert.assertEquals(FlowFileAccessException.class, ex.getClass());
+ }
+ try {
+ inputStream.skip(1L);
+ Assert.fail("Expected InputStream to be disabled; was able to call skip(long)");
+ } catch (final Exception ex) {
+ Assert.assertEquals(FlowFileAccessException.class, ex.getClass());
+ }
+ }
+
+ @Test
+ public void testAppendAfterSessionClosesStream() throws IOException {
+ final ContentClaim claim = contentRepo.create(false);
+ final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder()
+ .contentClaim(claim)
+ .addAttribute("uuid", "12345678-1234-1234-1234-123456789012")
+ .entryDate(System.currentTimeMillis())
+ .build();
+ flowFileQueue.put(flowFileRecord);
+ FlowFile flowFile = session.get();
+ assertNotNull(flowFile);
+ final ObjectHolder<OutputStream> outputStreamHolder = new ObjectHolder<>(null);
+ flowFile = session.append(flowFile, new OutputStreamCallback() {
+ @Override
+ public void process(final OutputStream outputStream) throws IOException {
+ outputStreamHolder.set(outputStream);
+ }
+ });
+ assertDisabled(outputStreamHolder.get());
+ }
+
+ @Test
+ public void testReadAfterSessionClosesStream() throws IOException {
+ final ContentClaim claim = contentRepo.create(false);
+ final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder()
+ .contentClaim(claim)
+ .addAttribute("uuid", "12345678-1234-1234-1234-123456789012")
+ .entryDate(System.currentTimeMillis())
+ .build();
+ flowFileQueue.put(flowFileRecord);
+ FlowFile flowFile = session.get();
+ assertNotNull(flowFile);
+ final ObjectHolder<InputStream> inputStreamHolder = new ObjectHolder<>(null);
+ session.read(flowFile, new InputStreamCallback() {
+ @Override
+ public void process(final InputStream inputStream) throws IOException {
+ inputStreamHolder.set(inputStream);
+ }
+ });
+ assertDisabled(inputStreamHolder.get());
+ }
+
+ @Test
+ public void testStreamAfterSessionClosesStream() throws IOException {
+ final ContentClaim claim = contentRepo.create(false);
+ final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder()
+ .contentClaim(claim)
+ .addAttribute("uuid", "12345678-1234-1234-1234-123456789012")
+ .entryDate(System.currentTimeMillis())
+ .build();
+ flowFileQueue.put(flowFileRecord);
+ FlowFile flowFile = session.get();
+ assertNotNull(flowFile);
+ final ObjectHolder<InputStream> inputStreamHolder = new ObjectHolder<>(null);
+ final ObjectHolder<OutputStream> outputStreamHolder = new ObjectHolder<>(null);
+ flowFile = session.write(flowFile, new StreamCallback() {
+ @Override
+ public void process(final InputStream input, final OutputStream output) throws IOException {
+ inputStreamHolder.set(input);
+ outputStreamHolder.set(output);
+ }
+ });
+ assertDisabled(inputStreamHolder.get());
+ assertDisabled(outputStreamHolder.get());
+ }
+
+ @Test
+ public void testWriteAfterSessionClosesStream() throws IOException {
+ final ContentClaim claim = contentRepo.create(false);
+ final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder()
+ .contentClaim(claim)
+ .addAttribute("uuid", "12345678-1234-1234-1234-123456789012")
+ .entryDate(System.currentTimeMillis())
+ .build();
+ flowFileQueue.put(flowFileRecord);
+ FlowFile flowFile = session.get();
+ assertNotNull(flowFile);
+ final ObjectHolder<OutputStream> outputStreamHolder = new ObjectHolder<>(null);
+ flowFile = session.write(flowFile, new OutputStreamCallback() {
+ @Override
+ public void process(final OutputStream out) throws IOException {
+ outputStreamHolder.set(out);
+ }
+ });
+ assertDisabled(outputStreamHolder.get());
+ }
+
@Test
public void testCreateThenRollbackRemovesContent() throws IOException {
@@ -998,6 +1141,12 @@ public class TestStandardProcessSession {
public ContentClaim create(boolean lossTolerant) throws IOException {
final ContentClaim claim = claimManager.newContentClaim("container", "section", String.valueOf(idGenerator.getAndIncrement()), false);
claimantCounts.put(claim, new AtomicInteger(1));
+ final Path path = getPath(claim);
+ final Path parent = path.getParent();
+ if (Files.exists(parent) == false) {
+ Files.createDirectories(parent);
+ }
+ Files.createFile(getPath(claim));
return claim;
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/3410197b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/contextlistener/ApplicationStartupContextListener.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/contextlistener/ApplicationStartupContextListener.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/contextlistener/ApplicationStartupContextListener.java
index a631670..8b48abf 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/contextlistener/ApplicationStartupContextListener.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/contextlistener/ApplicationStartupContextListener.java
@@ -91,10 +91,8 @@ public class ApplicationStartupContextListener implements ServletContextListener
* reloading actions, the node will start the necessary
* processors.
*/
- if (properties.getAutoResumeState()) {
- final FlowController flowController = flowService.getController();
- flowController.startDelayed();
- }
+ final FlowController flowController = flowService.getController();
+ flowController.onFlowInitialized(properties.getAutoResumeState());
logger.info("Flow Controller started successfully.");
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/3410197b/nifi/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/processors/kafka/PutKafka.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/processors/kafka/PutKafka.java b/nifi/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/processors/kafka/PutKafka.java
index 4df4e08..e0b7588 100644
--- a/nifi/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/processors/kafka/PutKafka.java
+++ b/nifi/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/processors/kafka/PutKafka.java
@@ -309,64 +309,61 @@ public class PutKafka extends AbstractProcessor {
data = Arrays.copyOfRange(baos.getUnderlyingBuffer(), 0, baos.size() - delimiterBytes.length);
}
- createMessage: if ( data != null ) {
+ if ( data != null ) {
// If the message has no data, ignore it.
- if ( data.length == 0 ) {
- data = null;
- baos.reset();
- break createMessage;
- }
-
- // either we ran out of data or we reached the end of the message.
- // Either way, create the message because it's ready to send.
- final KeyedMessage<byte[], byte[]> message;
- if ( key == null ) {
- message = new KeyedMessage<>(topic, data);
- } else {
- message = new KeyedMessage<>(topic, keyBytes, data);
- }
-
- // Add the message to the list of messages ready to send. If we've reached our
- // threshold of how many we're willing to send (or if we're out of data), go ahead
- // and send the whole List.
- messages.add(message);
- messageBytes += data.length;
- if ( messageBytes >= maxBufferSize || streamFinished ) {
- // send the messages, then reset our state.
- try {
- producer.send(messages);
- } catch (final Exception e) {
- // we wrap the general exception in ProcessException because we want to separate
- // failures in sending messages from general Exceptions that would indicate bugs
- // in the Processor. Failure to send a message should be handled appropriately, but
- // we don't want to catch the general Exception or RuntimeException in order to catch
- // failures from Kafka's Producer.
- throw new ProcessException("Failed to send messages to Kafka", e);
+ if ( data.length != 0 ) {
+ // either we ran out of data or we reached the end of the message.
+ // Either way, create the message because it's ready to send.
+ final KeyedMessage<byte[], byte[]> message;
+ if (key == null) {
+ message = new KeyedMessage<>(topic, data);
+ } else {
+ message = new KeyedMessage<>(topic, keyBytes, data);
+ }
+
+ // Add the message to the list of messages ready to send. If we've reached our
+ // threshold of how many we're willing to send (or if we're out of data), go ahead
+ // and send the whole List.
+ messages.add(message);
+ messageBytes += data.length;
+ if (messageBytes >= maxBufferSize || streamFinished) {
+ // send the messages, then reset our state.
+ try {
+ producer.send(messages);
+ } catch (final Exception e) {
+ // we wrap the general exception in ProcessException because we want to separate
+ // failures in sending messages from general Exceptions that would indicate bugs
+ // in the Processor. Failure to send a message should be handled appropriately, but
+ // we don't want to catch the general Exception or RuntimeException in order to catch
+ // failures from Kafka's Producer.
+ throw new ProcessException("Failed to send messages to Kafka", e);
+ }
+
+ messagesSent.addAndGet(messages.size()); // count number of messages sent
+
+ // reset state
+ messages.clear();
+ messageBytes = 0;
+
+ // We've successfully sent a batch of messages. Keep track of the byte offset in the
+ // FlowFile of the last successfully sent message. This way, if the messages cannot
+ // all be successfully sent, we know where to split off the data. This allows us to then
+ // split off the first X number of bytes and send to 'success' and then split off the rest
+ // and send them to 'failure'.
+ lastMessageOffset.set(in.getBytesConsumed());
}
-
- messagesSent.addAndGet(messages.size()); // count number of messages sent
-
- // reset state
- messages.clear();
- messageBytes = 0;
-
- // We've successfully sent a batch of messages. Keep track of the byte offset in the
- // FlowFile of the last successfully sent message. This way, if the messages cannot
- // all be successfully sent, we know where to split off the data. This allows us to then
- // split off the first X number of bytes and send to 'success' and then split off the rest
- // and send them to 'failure'.
- lastMessageOffset.set(in.getBytesConsumed());
}
-
// reset BAOS so that we can start a new message.
baos.reset();
data = null;
+
}
}
// If there are messages left, send them
if ( !messages.isEmpty() ) {
try {
+ messagesSent.addAndGet(messages.size()); // add count of messages
producer.send(messages);
} catch (final Exception e) {
throw new ProcessException("Failed to send messages to Kafka", e);
@@ -377,7 +374,6 @@ public class PutKafka extends AbstractProcessor {
});
final long nanos = System.nanoTime() - start;
-
session.getProvenanceReporter().send(flowFile, "kafka://" + topic, "Sent " + messagesSent.get() + " messages");
session.transfer(flowFile, REL_SUCCESS);
getLogger().info("Successfully sent {} messages to Kafka for {} in {} millis", new Object[] {messagesSent.get(), flowFile, TimeUnit.NANOSECONDS.toMillis(nanos)});
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/3410197b/nifi/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/processors/kafka/TestPutKafka.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/processors/kafka/TestPutKafka.java b/nifi/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/processors/kafka/TestPutKafka.java
index b0f2394..56a5c4b 100644
--- a/nifi/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/processors/kafka/TestPutKafka.java
+++ b/nifi/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/test/java/org/apache/nifi/processors/kafka/TestPutKafka.java
@@ -25,6 +25,7 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
import kafka.common.FailedToSendMessageException;
import kafka.javaapi.producer.Producer;
@@ -32,12 +33,14 @@ import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
+
import org.apache.nifi.processor.ProcessContext;
-import org.apache.nifi.util.MockFlowFile;
-import org.apache.nifi.util.TestRunner;
-import org.apache.nifi.util.TestRunners;
+import org.apache.nifi.provenance.ProvenanceReporter;
+import org.apache.nifi.util.*;
import org.junit.Ignore;
import org.junit.Test;
+import org.mockito.Mockito;
+import org.mockito.internal.util.reflection.Whitebox;
public class TestPutKafka {
@@ -139,8 +142,70 @@ public class TestPutKafka {
assertTrue(Arrays.equals("3".getBytes(), msgs.get(2)));
assertTrue(Arrays.equals("4".getBytes(), msgs.get(3)));
}
-
-
+
+ @Test
+ public void testProvenanceReporterMessagesCount(){
+ final TestableProcessor processor = new TestableProcessor();
+
+ ProvenanceReporter spyProvenanceReporter = Mockito.spy(new MockProvenanceReporter());
+
+ AtomicLong idGenerator = new AtomicLong(0L);
+ SharedSessionState sharedState = new SharedSessionState(processor, idGenerator);
+ Whitebox.setInternalState(sharedState, "provenanceReporter", spyProvenanceReporter);
+ MockFlowFileQueue flowFileQueue = sharedState.getFlowFileQueue();
+ MockSessionFactory sessionFactory = Mockito.mock(MockSessionFactory.class);
+ MockProcessSession mockProcessSession = new MockProcessSession(sharedState);
+ Mockito.when(sessionFactory.createSession()).thenReturn(mockProcessSession);
+
+
+ final TestRunner runner = TestRunners.newTestRunner(processor);
+ Whitebox.setInternalState(runner, "flowFileQueue", flowFileQueue);
+ Whitebox.setInternalState(runner, "sessionFactory", sessionFactory);
+
+ runner.setProperty(PutKafka.TOPIC, "topic1");
+ runner.setProperty(PutKafka.KEY, "key1");
+ runner.setProperty(PutKafka.SEED_BROKERS, "localhost:1234");
+ runner.setProperty(PutKafka.MESSAGE_DELIMITER, "\\n");
+
+ final byte[] bytes = "\n\n\n1\n2\n\n\n\n3\n4\n\n\n".getBytes();
+ runner.enqueue(bytes);
+ runner.run();
+
+ MockFlowFile mockFlowFile = mockProcessSession.getFlowFilesForRelationship(PutKafka.REL_SUCCESS).get(0);
+ Mockito.verify(spyProvenanceReporter, Mockito.atLeastOnce()).send(mockFlowFile, "kafka://topic1", "Sent 4 messages");
+ }
+
+ @Test
+ public void testProvenanceReporterWithoutDelimiterMessagesCount(){
+ final TestableProcessor processor = new TestableProcessor();
+
+ ProvenanceReporter spyProvenanceReporter = Mockito.spy(new MockProvenanceReporter());
+
+ AtomicLong idGenerator = new AtomicLong(0L);
+ SharedSessionState sharedState = new SharedSessionState(processor, idGenerator);
+ Whitebox.setInternalState(sharedState, "provenanceReporter", spyProvenanceReporter);
+ MockFlowFileQueue flowFileQueue = sharedState.getFlowFileQueue();
+ MockSessionFactory sessionFactory = Mockito.mock(MockSessionFactory.class);
+ MockProcessSession mockProcessSession = new MockProcessSession(sharedState);
+ Mockito.when(sessionFactory.createSession()).thenReturn(mockProcessSession);
+
+
+ final TestRunner runner = TestRunners.newTestRunner(processor);
+ Whitebox.setInternalState(runner, "flowFileQueue", flowFileQueue);
+ Whitebox.setInternalState(runner, "sessionFactory", sessionFactory);
+
+ runner.setProperty(PutKafka.TOPIC, "topic1");
+ runner.setProperty(PutKafka.KEY, "key1");
+ runner.setProperty(PutKafka.SEED_BROKERS, "localhost:1234");
+
+ final byte[] bytes = "\n\n\n1\n2\n\n\n\n3\n4\n\n\n".getBytes();
+ runner.enqueue(bytes);
+ runner.run();
+
+ MockFlowFile mockFlowFile = mockProcessSession.getFlowFilesForRelationship(PutKafka.REL_SUCCESS).get(0);
+ Mockito.verify(spyProvenanceReporter, Mockito.atLeastOnce()).send(mockFlowFile, "kafka://topic1");
+ }
+
@Test
@Ignore("Intended only for local testing; requires an actual running instance of Kafka & ZooKeeper...")
public void testKeyValuePut() {
@@ -233,4 +298,5 @@ public class TestPutKafka {
failAfter = successCount;
}
}
+
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/3410197b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/EvaluateRegularExpression.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/EvaluateRegularExpression.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/EvaluateRegularExpression.java
index 4140943..bb2e31a 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/EvaluateRegularExpression.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/EvaluateRegularExpression.java
@@ -52,9 +52,10 @@ import org.apache.commons.lang3.StringUtils;
@EventDriven
@SideEffectFree
@SupportsBatching
-@Tags({"evaluate", "Text", "Regular Expression", "regex", "experimental"})
+@Tags({"deprecated"})
@CapabilityDescription(
- "Evaluates one or more Regular Expressions against the content of a FlowFile. "
+ "WARNING: This has been deprecated and will be removed in 0.2.0. Use ExtractText instead.\n"
+ + "Evaluates one or more Regular Expressions against the content of a FlowFile. "
+ "The results of those Regular Expressions are assigned to FlowFile Attributes. "
+ "Regular Expressions are entered by adding user-defined properties; "
+ "the name of the property maps to the Attribute Name into which the result will be placed. "
@@ -62,7 +63,7 @@ import org.apache.commons.lang3.StringUtils;
+ "If the Regular Expression matches more than once, only the first match will be used. "
+ "If any provided Regular Expression matches, the FlowFile(s) will be routed to 'matched'. "
+ "If no provided Regular Expression matches, the FlowFile will be routed to 'unmatched' and no attributes will be applied to the FlowFile.")
-
+@Deprecated
public class EvaluateRegularExpression extends AbstractProcessor {
public static final PropertyDescriptor CHARACTER_SET = new PropertyDescriptor.Builder()
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/3410197b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExtractText.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExtractText.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExtractText.java
new file mode 100644
index 0000000..aa1b7a9
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExtractText.java
@@ -0,0 +1,313 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.standard;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.stream.io.StreamUtils;
+import org.apache.nifi.logging.ProcessorLog;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.SupportsBatching;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+
+@EventDriven
+@SideEffectFree
+@SupportsBatching
+@Tags({"evaluate", "extract", "Text", "Regular Expression", "regex"})
+@CapabilityDescription(
+ "Evaluates one or more Regular Expressions against the content of a FlowFile. "
+ + "The results of those Regular Expressions are assigned to FlowFile Attributes. "
+ + "Regular Expressions are entered by adding user-defined properties; "
+ + "the name of the property maps to the Attribute Name into which the result will be placed. "
+ + "The first capture group, if any found, will be placed into that attribute name."
+ + "But all catpure groups, including the matching string sequence itself will also be "
+ + "provided at that attribute name with an index value provided."
+ + "The value of the property must be a valid Regular Expressions with one or more capturing groups. "
+ + "If the Regular Expression matches more than once, only the first match will be used. "
+ + "If any provided Regular Expression matches, the FlowFile(s) will be routed to 'matched'. "
+ + "If no provided Regular Expression matches, the FlowFile will be routed to 'unmatched' "
+ + "and no attributes will be applied to the FlowFile.")
+
+public class ExtractText extends AbstractProcessor {
+
+ public static final PropertyDescriptor CHARACTER_SET = new PropertyDescriptor.Builder()
+ .name("Character Set")
+ .description("The Character Set in which the file is encoded")
+ .required(true)
+ .addValidator(StandardValidators.CHARACTER_SET_VALIDATOR)
+ .defaultValue("UTF-8")
+ .build();
+
+ public static final PropertyDescriptor MAX_BUFFER_SIZE = new PropertyDescriptor.Builder()
+ .name("Maximum Buffer Size")
+ .description("Specifies the maximum amount of data to buffer (per file) in order to apply the regular expressions. Files larger than the specified maximum will not be fully evaluated.")
+ .required(true)
+ .addValidator(StandardValidators.DATA_SIZE_VALIDATOR)
+ .addValidator(StandardValidators.createDataSizeBoundsValidator(0, Integer.MAX_VALUE))
+ .defaultValue("1 MB")
+ .build();
+
+ public static final PropertyDescriptor MAX_CAPTURE_GROUP_LENGTH = new PropertyDescriptor.Builder()
+ .name("Maximum Capture Group Length")
+ .description("Specifies the maximum number of characters a given capture group value can have. Any characters beyond the max will be truncated.")
+ .required(false)
+ .defaultValue("1024")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor CANON_EQ = new PropertyDescriptor.Builder()
+ .name("Enable Canonical Equivalence")
+ .description("Indicates that two characters match only when their full canonical decompositions match.")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .build();
+
+ public static final PropertyDescriptor CASE_INSENSITIVE = new PropertyDescriptor.Builder()
+ .name("Enable Case-insensitive Matching")
+ .description("Indicates that two characters match even if they are in a different case. Can also be specified via the embeded flag (?i).")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .build();
+
+ public static final PropertyDescriptor COMMENTS = new PropertyDescriptor.Builder()
+ .name("Permit Whitespace and Comments in Pattern")
+ .description("In this mode, whitespace is ignored, and embedded comments starting with # are ignored until the end of a line. Can also be specified via the embeded flag (?x).")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .build();
+
+ public static final PropertyDescriptor DOTALL = new PropertyDescriptor.Builder()
+ .name("Enable DOTALL Mode")
+ .description("Indicates that the expression '.' should match any character, including a line terminator. Can also be specified via the embeded flag (?s).")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .build();
+
+ public static final PropertyDescriptor LITERAL = new PropertyDescriptor.Builder()
+ .name("Enable Literal Parsing of the Pattern")
+ .description("Indicates that Metacharacters and escape characters should be given no special meaning.")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .build();
+
+ public static final PropertyDescriptor MULTILINE = new PropertyDescriptor.Builder()
+ .name("Enable Multiline Mode")
+ .description("Indicates that '^' and '$' should match just after and just before a line terminator or end of sequence, instead of only the begining or end of the entire input. Can also be specified via the embeded flag (?m).")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .build();
+
+ public static final PropertyDescriptor UNICODE_CASE = new PropertyDescriptor.Builder()
+ .name("Enable Unicode-aware Case Folding")
+ .description("When used with 'Enable Case-insensitive Matching', matches in a manner consistent with the Unicode Standard. Can also be specified via the embeded flag (?u).")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .build();
+
+ public static final PropertyDescriptor UNICODE_CHARACTER_CLASS = new PropertyDescriptor.Builder()
+ .name("Enable Unicode Predefined Character Classes")
+ .description("Specifies conformance with the Unicode Technical Standard #18: Unicode Regular Expression Annex C: Compatibility Properties. Can also be specified via the embeded flag (?U).")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .build();
+
+ public static final PropertyDescriptor UNIX_LINES = new PropertyDescriptor.Builder()
+ .name("Enable Unix Lines Mode")
+ .description("Indicates that only the '\n' line terminator is recognized int the behavior of '.', '^', and '$'. Can also be specified via the embeded flag (?d).")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .build();
+
+ public static final Relationship REL_MATCH = new Relationship.Builder()
+ .name("matched")
+ .description(
+ "FlowFiles are routed to this relationship when the Regular Expression is successfully evaluated and the FlowFile "
+ + "is modified as a result")
+ .build();
+
+ public static final Relationship REL_NO_MATCH = new Relationship.Builder()
+ .name("unmatched")
+ .description(
+ "FlowFiles are routed to this relationship when no provided Regular Expression matches the content of the FlowFile")
+ .build();
+
+ private Set<Relationship> relationships;
+ private List<PropertyDescriptor> properties;
+ private final AtomicReference<Map<String, Pattern>> compiledPattersMapRef = new AtomicReference<>();
+
+ @Override
+ protected void init(final ProcessorInitializationContext context) {
+ final Set<Relationship> rels = new HashSet<>();
+ rels.add(REL_MATCH);
+ rels.add(REL_NO_MATCH);
+ this.relationships = Collections.unmodifiableSet(rels);
+
+ final List<PropertyDescriptor> props = new ArrayList<>();
+ props.add(CHARACTER_SET);
+ props.add(MAX_BUFFER_SIZE);
+ props.add(MAX_CAPTURE_GROUP_LENGTH);
+ props.add(CANON_EQ);
+ props.add(CASE_INSENSITIVE);
+ props.add(COMMENTS);
+ props.add(DOTALL);
+ props.add(LITERAL);
+ props.add(MULTILINE);
+ props.add(UNICODE_CASE);
+ props.add(UNICODE_CHARACTER_CLASS);
+ props.add(UNIX_LINES);
+ this.properties = Collections.unmodifiableList(props);
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return properties;
+ }
+
+ @Override
+ protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(final String propertyDescriptorName) {
+ return new PropertyDescriptor.Builder()
+ .name(propertyDescriptorName)
+ .expressionLanguageSupported(false)
+ .addValidator(StandardValidators.createRegexValidator(1, 40, true))
+ .required(false)
+ .dynamic(true)
+ .build();
+ }
+
+ @OnScheduled
+ public final void onScheduled(final ProcessContext context) throws IOException {
+ final Map<String, Pattern> compiledPatternsMap = new HashMap<>();
+
+ for (final Map.Entry<PropertyDescriptor, String> entry : context.getProperties().entrySet()) {
+ if (!entry.getKey().isDynamic()) {
+ continue;
+ }
+ final int flags = getCompileFlags(context);
+ final Pattern pattern = Pattern.compile(entry.getValue(), flags);
+ compiledPatternsMap.put(entry.getKey().getName(), pattern);
+ }
+ compiledPattersMapRef.set(compiledPatternsMap);
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+ final ProcessorLog logger = getLogger();
+ final Charset charset = Charset.forName(context.getProperty(CHARACTER_SET).getValue());
+ final int maxCaptureGroupLength = context.getProperty(MAX_CAPTURE_GROUP_LENGTH).asInteger();
+ final long maxBufferSizeL = Math.min(flowFile.getSize(), context.getProperty(MAX_BUFFER_SIZE).asDataSize(DataUnit.B).longValue());
+ final byte[] buffer = new byte[(int) maxBufferSizeL];
+
+ session.read(flowFile, new InputStreamCallback() {
+ @Override
+ public void process(InputStream in) throws IOException {
+ StreamUtils.fillBuffer(in, buffer, false);
+ }
+ });
+
+ final String contentString = new String(buffer, 0, (int) maxBufferSizeL, charset);
+ final Map<String, String> regexResults = new HashMap<>();
+
+ final Map<String, Pattern> patternMap = compiledPattersMapRef.get();
+ for (final Map.Entry<String, Pattern> entry : patternMap.entrySet()) {
+
+ final Matcher matcher = entry.getValue().matcher(contentString);
+
+ if (matcher.find()) {
+ final String baseKey = entry.getKey();
+ for (int i = 0; i <= matcher.groupCount(); i++) {
+ final String key = new StringBuilder(baseKey).append(".").append(i).toString();
+ String value = matcher.group(i);
+ if (value.length() > maxCaptureGroupLength) {
+ value = value.substring(0, maxCaptureGroupLength);
+ }
+ regexResults.put(key, value);
+ if (i == 1) {
+ regexResults.put(baseKey, value);
+ }
+ }
+ }
+ }
+
+ if (!regexResults.isEmpty()) {
+ flowFile = session.putAllAttributes(flowFile, regexResults);
+ session.getProvenanceReporter().modifyAttributes(flowFile);
+ session.transfer(flowFile, REL_MATCH);
+ logger.info("Matched {} Regular Expressions and added attributes to FlowFile {}", new Object[]{regexResults.size(), flowFile});
+ } else {
+ session.transfer(flowFile, REL_NO_MATCH);
+ logger.info("Did not match any Regular Expressions for FlowFile {}", new Object[]{flowFile});
+ }
+
+ }
+
+ int getCompileFlags(ProcessContext context) {
+ int flags = (context.getProperty(UNIX_LINES).asBoolean() ? Pattern.UNIX_LINES : 0)
+ | (context.getProperty(CASE_INSENSITIVE).asBoolean() ? Pattern.CASE_INSENSITIVE : 0)
+ | (context.getProperty(COMMENTS).asBoolean() ? Pattern.COMMENTS : 0)
+ | (context.getProperty(MULTILINE).asBoolean() ? Pattern.MULTILINE : 0)
+ | (context.getProperty(LITERAL).asBoolean() ? Pattern.LITERAL : 0)
+ | (context.getProperty(DOTALL).asBoolean() ? Pattern.DOTALL : 0)
+ | (context.getProperty(UNICODE_CASE).asBoolean() ? Pattern.UNICODE_CASE : 0)
+ | (context.getProperty(CANON_EQ).asBoolean() ? Pattern.CANON_EQ : 0)
+ | (context.getProperty(UNICODE_CHARACTER_CLASS).asBoolean() ? Pattern.UNICODE_CHARACTER_CLASS : 0);
+ return flags;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/3410197b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
index f81ccec..7fbd781 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
@@ -25,6 +25,7 @@ org.apache.nifi.processors.standard.EvaluateXPath
org.apache.nifi.processors.standard.EvaluateXQuery
org.apache.nifi.processors.standard.ExecuteStreamCommand
org.apache.nifi.processors.standard.ExecuteProcess
+org.apache.nifi.processors.standard.ExtractText
org.apache.nifi.processors.standard.GenerateFlowFile
org.apache.nifi.processors.standard.GetFile
org.apache.nifi.processors.standard.GetFTP