This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 39ca75d868 NIFI-13914 Fixed FlowFile Prioritization for Stateless
Execution (#9432)
39ca75d868 is described below
commit 39ca75d868ffe0e7df465798a77c33e74d154b7c
Author: Mark Payne <[email protected]>
AuthorDate: Mon Oct 21 16:39:12 2024 -0400
NIFI-13914 Fixed FlowFile Prioritization for Stateless Execution (#9432)
Ensure that when data is enqueued in the StatelessFlowFileQueue we order
the data properly before adding to its internal queue. Added system tests to
verify that the data is properly ordered while running within stateless and
properly ordered when coming out, if using a FIFO prioritizer.
Signed-off-by: David Handermann <[email protected]>
---
.../stateless/queue/StatelessFlowFileQueue.java | 16 +++-
.../processors/tests/system/ReOrderFlowFiles.java | 88 ++++++++++++++++++
.../processors/tests/system/VerifyEvenThenOdd.java | 100 +++++++++++++++++++++
.../services/org.apache.nifi.processor.Processor | 2 +
.../apache/nifi/tests/system/NiFiClientUtil.java | 27 ++++++
.../tests/system/stateless/StatelessBasicsIT.java | 56 ++++++++++++
6 files changed, 286 insertions(+), 3 deletions(-)
diff --git
a/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/queue/StatelessFlowFileQueue.java
b/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/queue/StatelessFlowFileQueue.java
index 03ee282ff2..caaa607b10 100644
---
a/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/queue/StatelessFlowFileQueue.java
+++
b/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/queue/StatelessFlowFileQueue.java
@@ -18,7 +18,6 @@
package org.apache.nifi.stateless.queue;
import org.apache.nifi.controller.queue.DropFlowFileStatus;
-import org.apache.nifi.controller.status.FlowFileAvailability;
import org.apache.nifi.controller.queue.ListFlowFileStatus;
import org.apache.nifi.controller.queue.LoadBalanceCompression;
import org.apache.nifi.controller.queue.LoadBalanceStrategy;
@@ -27,6 +26,8 @@ import org.apache.nifi.controller.queue.QueueDiagnostics;
import org.apache.nifi.controller.queue.QueueSize;
import org.apache.nifi.controller.repository.FlowFileRecord;
import org.apache.nifi.controller.repository.SwapSummary;
+import org.apache.nifi.controller.status.FlowFileAvailability;
+import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.flowfile.FlowFilePrioritizer;
import org.apache.nifi.processor.FlowFileFilter;
import org.apache.nifi.util.FormatUtils;
@@ -34,8 +35,10 @@ import org.apache.nifi.util.FormatUtils;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
+import java.util.Optional;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
@@ -165,8 +168,15 @@ public class StatelessFlowFileQueue implements
DrainableFlowFileQueue {
@Override
public void putAll(final Collection<FlowFileRecord> flowFiles) {
- this.flowFiles.addAll(flowFiles);
- flowFiles.forEach(ff -> totalBytes.addAndGet(ff.getSize()));
+ // Order the FlowFiles in the same order they were transferred by the
Processor. This ensures that we keep the ordering provided by the Processor.
+ // This is not important for the Standard NiFi engine because it uses
the StandardFlowFileQueue, which maintains the order of FlowFiles as configured.
+ // However, in stateless, we want to keep the order that the upstream
processor processes the data.
+ final List<FlowFileRecord> orderedFlowFiles = new
ArrayList<>(flowFiles);
+ orderedFlowFiles.sort(Comparator.comparingLong((FlowFileRecord
flowFile) -> Optional.ofNullable(flowFile.getLastQueueDate()).orElse(0L))
+ .thenComparingLong(FlowFile::getQueueDateIndex));
+
+ this.flowFiles.addAll(orderedFlowFiles);
+ orderedFlowFiles.forEach(ff -> totalBytes.addAndGet(ff.getSize()));
}
@Override
diff --git
a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/ReOrderFlowFiles.java
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/ReOrderFlowFiles.java
new file mode 100644
index 0000000000..3f671149fe
--- /dev/null
+++
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/ReOrderFlowFiles.java
@@ -0,0 +1,88 @@
+/*
+ * 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.tests.system;
+
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.PropertyValue;
+import org.apache.nifi.expression.AttributeExpression;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.FlowFileFilter;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.List;
+import java.util.Set;
+
+@CapabilityDescription("Selects FlowFiles that match the given criteria and
transfers them to the 'success' relationship. Then, selects all other FlowFiles
and transfers them " +
+ "to the success relationship. Note that this Processor
will not work properly if it is scheduled to run while its incoming queue(s)
are being populated. " +
+ "This is meant to be used only for purposes of testing
in a Stateless execution engine and makes use of FlowFileFilters.")
+public class ReOrderFlowFiles extends AbstractProcessor {
+ protected static PropertyDescriptor FIRST_SELECTION_CRITERIA = new
PropertyDescriptor.Builder()
+ .name("First Group Selection Criteria")
+ .description("An Expression Language expression that evaluates to true
or false. FlowFiles that evaluate to true will be transferred first; others " +
+ "will be transferred after.")
+ .required(true)
+
.addValidator(StandardValidators.createAttributeExpressionLanguageValidator(AttributeExpression.ResultType.BOOLEAN,
false))
+ .build();
+
+ private static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("All FlowFiles are transferred to this relationship.")
+ .build();
+
+ private static final Set<Relationship> relationships = Set.of(REL_SUCCESS);
+ private static final List<PropertyDescriptor> properties =
List.of(FIRST_SELECTION_CRITERIA);
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return properties;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession
session) throws ProcessException {
+ final PropertyValue selectionValue =
context.getProperty(FIRST_SELECTION_CRITERIA);
+
+ final List<FlowFile> matching = session.get(new FlowFileFilter() {
+ @Override
+ public FlowFileFilterResult filter(final FlowFile flowFile) {
+ final boolean selected =
selectionValue.evaluateAttributeExpressions(flowFile).asBoolean();
+ return selected ? FlowFileFilterResult.ACCEPT_AND_CONTINUE :
FlowFileFilterResult.REJECT_AND_CONTINUE;
+ }
+ });
+
+ final List<FlowFile> unmatched = session.get(new FlowFileFilter() {
+ @Override
+ public FlowFileFilterResult filter(final FlowFile flowFile) {
+ return FlowFileFilterResult.ACCEPT_AND_CONTINUE;
+ }
+ });
+
+ session.transfer(matching, REL_SUCCESS);
+ session.transfer(unmatched, REL_SUCCESS);
+ }
+}
diff --git
a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/VerifyEvenThenOdd.java
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/VerifyEvenThenOdd.java
new file mode 100644
index 0000000000..5adf559cda
--- /dev/null
+++
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/VerifyEvenThenOdd.java
@@ -0,0 +1,100 @@
+/*
+ * 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.tests.system;
+
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+@CapabilityDescription("Ensures that all FlowFiles that are in the Processor's
incoming queue are ordered such that the value of the key attribute " +
+ "is even for all FlowFiles before the first FlowFile
with an odd value. If the FlowFiles are ordered correctly, they are transferred
to " +
+ "the 'success' relationship; otherwise, they are
transferred to the 'failure' relationship. The name of the key attribute is
configurable. " +
+ "This is used to ensure that data is properly ordered
while running within a Stateless flow.")
+public class VerifyEvenThenOdd extends AbstractProcessor {
+
+ protected static final PropertyDescriptor ATTRIBUTE_NAME = new
PropertyDescriptor.Builder()
+ .name("Attribute Name")
+ .description("The name of the attribute to check for even or odd
values")
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ private static final List<PropertyDescriptor> properties =
List.of(ATTRIBUTE_NAME);
+
+ protected static final Relationship REL_SUCCESS = new
Relationship.Builder()
+ .name("success")
+ .description("All FlowFiles are transferred to this relationship in
the event that all queued FlowFiles are ordered correctly.")
+ .build();
+
+ protected static final Relationship REL_FAILURE = new
Relationship.Builder()
+ .name("failure")
+ .description("All FlowFiles are transferred to this relationship in
the event that any queued FlowFiles are not ordered correctly.")
+ .build();
+
+ private static final Set<Relationship> relationships = Set.of(REL_SUCCESS,
REL_FAILURE);
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return properties;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession
session) throws ProcessException {
+ final List<FlowFile> allFlowFiles = new ArrayList<>();
+ List<FlowFile> batch;
+ while (!(batch = session.get(1000)).isEmpty()) {
+ allFlowFiles.addAll(batch);
+ }
+
+ boolean oddSeen = false;
+ boolean ordered = true;
+ for (final FlowFile flowFile : allFlowFiles) {
+ final String value =
flowFile.getAttribute(context.getProperty(ATTRIBUTE_NAME).getValue());
+ final int intValue = Integer.parseInt(value);
+
+ final boolean even = intValue % 2 == 0;
+ if (even && oddSeen) {
+ ordered = false;
+ break;
+ } else if (!even) {
+ oddSeen = true;
+ }
+ }
+
+ if (ordered) {
+ session.transfer(allFlowFiles, REL_SUCCESS);
+ } else {
+ session.transfer(allFlowFiles, REL_FAILURE);
+ }
+ }
+}
diff --git
a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
index b86667f6a4..0b344481fc 100644
---
a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
+++
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
@@ -35,6 +35,7 @@ org.apache.nifi.processors.tests.system.PartitionText
org.apache.nifi.processors.tests.system.PassThrough
org.apache.nifi.processors.tests.system.PassThroughRequiresInstanceClassLoading
org.apache.nifi.processors.tests.system.CountPrimaryNodeChangeEvents
+org.apache.nifi.processors.tests.system.ReOrderFlowFiles
org.apache.nifi.processors.tests.system.ReplaceWithFile
org.apache.nifi.processors.tests.system.ReverseContents
org.apache.nifi.processors.tests.system.RoundRobinFlowFiles
@@ -50,6 +51,7 @@ org.apache.nifi.processors.tests.system.UpdateContent
org.apache.nifi.processors.tests.system.UnzipFlowFile
org.apache.nifi.processors.tests.system.ValidateFileExists
org.apache.nifi.processors.tests.system.VerifyContents
+org.apache.nifi.processors.tests.system.VerifyEvenThenOdd
org.apache.nifi.processors.tests.system.WriteFlowFileCountToFile
org.apache.nifi.processors.tests.system.WriteLifecycleEvents
org.apache.nifi.processors.tests.system.WriteToFile
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiClientUtil.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiClientUtil.java
index 44b4fc5013..dbdcddd053 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiClientUtil.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiClientUtil.java
@@ -1551,6 +1551,33 @@ public class NiFiClientUtil {
return getConnectionClient().updateConnection(updatedEntity);
}
+ public ConnectionEntity updateConnectionBackpressure(final
ConnectionEntity connectionEntity, final long flowFileCount, final long bytes)
throws NiFiClientException, IOException {
+ final ConnectionDTO connectionDto = new ConnectionDTO();
+ connectionDto.setBackPressureDataSizeThreshold(bytes + " B");
+ connectionDto.setBackPressureObjectThreshold(flowFileCount);
+ connectionDto.setId(connectionEntity.getId());
+
+ final ConnectionEntity updatedEntity = new ConnectionEntity();
+ updatedEntity.setComponent(connectionDto);
+ updatedEntity.setId(connectionEntity.getId());
+ updatedEntity.setRevision(connectionEntity.getRevision());
+
+ return getConnectionClient().updateConnection(updatedEntity);
+ }
+
+ public ConnectionEntity updateConnectionPrioritizer(final ConnectionEntity
connectionEntity, final String prioritizerName) throws NiFiClientException,
IOException {
+ final ConnectionDTO connectionDto = new ConnectionDTO();
+ connectionDto.setPrioritizers(List.of("org.apache.nifi.prioritizer." +
prioritizerName));
+ connectionDto.setId(connectionEntity.getId());
+
+ final ConnectionEntity updatedEntity = new ConnectionEntity();
+ updatedEntity.setComponent(connectionDto);
+ updatedEntity.setId(connectionEntity.getId());
+ updatedEntity.setRevision(connectionEntity.getRevision());
+
+ return getConnectionClient().updateConnection(updatedEntity);
+ }
+
public DropRequestEntity emptyQueue(final String connectionId) throws
NiFiClientException, IOException {
final ConnectionClient connectionClient = getConnectionClient();
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/stateless/StatelessBasicsIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/stateless/StatelessBasicsIT.java
index 6357f80561..c48874d82d 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/stateless/StatelessBasicsIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/stateless/StatelessBasicsIT.java
@@ -46,6 +46,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
@@ -82,6 +83,61 @@ public class StatelessBasicsIT extends NiFiSystemIT {
startClaimantCount = getClaimantCounts();
}
+ @Test
+ public void testOrderingIntraSession() throws NiFiClientException,
IOException, InterruptedException {
+ final int batchSize = 100;
+
+ statelessGroup = getClientUtil().createProcessGroup("Stateless",
"root");
+ getClientUtil().markStateless(statelessGroup, "1 min");
+
+ final ProcessorEntity generate =
getClientUtil().createProcessor(GENERATE_FLOWFILE, statelessGroup.getId());
+ final Map<String, String> generateProperties = new HashMap<>();
+ generateProperties.put("Text", HELLO_WORLD);
+ generateProperties.put("Batch Size", String.valueOf(batchSize));
+ generateProperties.put("Counter", "${nextInt()}");
+ getClientUtil().updateProcessorProperties(generate,
generateProperties);
+
+ final ProcessorEntity router =
getClientUtil().createProcessor("ReOrderFlowFiles", statelessGroup.getId());
+ getClientUtil().updateProcessorProperties(router, Map.of("First Group
Selection Criteria", "${Counter:mod(2):equals(0)}"));
+
+ // Verify that FlowFiles are ordered correctly within stateless flow.
+ final ProcessorEntity verifyProcessor =
getClientUtil().createProcessor("VerifyEvenThenOdd", statelessGroup.getId());
+ getClientUtil().updateProcessorProperties(verifyProcessor,
Map.of("Attribute Name", "Counter"));
+
+ final PortEntity outputPort = getClientUtil().createOutputPort("Out",
statelessGroup.getId());
+
+ getClientUtil().createConnection(generate, router, "success");
+ getClientUtil().createConnection(router, verifyProcessor, "success");
+ getClientUtil().createConnection(verifyProcessor, outputPort,
"success");
+ getClientUtil().setAutoTerminatedRelationships(verifyProcessor,
"failure");
+
+ final ProcessorEntity terminate =
getClientUtil().createProcessor(TERMINATE_FLOWFILE);
+ final ConnectionEntity outputToTerminate =
getClientUtil().createConnection(outputPort, terminate);
+ getClientUtil().updateConnectionPrioritizer(outputToTerminate,
"FirstInFirstOutPrioritizer");
+
+ getClientUtil().waitForValidProcessor(generate.getId());
+ getClientUtil().waitForValidProcessor(router.getId());
+ getClientUtil().startProcessGroupComponents(statelessGroup.getId());
+
+ waitForQueueCount(outputToTerminate.getId(), batchSize);
+ getClientUtil().stopProcessGroupComponents(statelessGroup.getId());
+
+ final List<String> actualCounterValues = new ArrayList<>();
+ for (int i = 0; i < batchSize; i++) {
+ final FlowFileEntity flowFileEntity =
getClientUtil().getQueueFlowFile(outputToTerminate.getId(), i);
+
actualCounterValues.add(flowFileEntity.getFlowFile().getAttributes().get("Counter"));
+ }
+
+ int expectedCounter = 0;
+ for (int i = 0; i < batchSize; i++) {
+ assertEquals(String.valueOf(expectedCounter),
actualCounterValues.get(i));
+ expectedCounter += 2;
+ if (expectedCounter >= batchSize) {
+ expectedCounter = 1;
+ }
+ }
+ }
+
@Test
public void testOneInOneOut() throws NiFiClientException, IOException,
InterruptedException {
createFlowShell();
