This is an automated email from the ASF dual-hosted git repository.
bbende 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 d61be1197c4 NIFI-16378: Handle disabled Processors during Connector
drain (#11709)
d61be1197c4 is described below
commit d61be1197c4951417095486d8e0630e65379109f
Author: Mark Payne <[email protected]>
AuthorDate: Fri Sep 25 14:20:39 2026 -0400
NIFI-16378: Handle disabled Processors during Connector drain (#11709)
Co-authored-by: Cursor <[email protected]>
---
.../facades/standalone/StandaloneProcessorLifecycle.java | 6 ++++++
.../tests/system/GatedDataQueuingConnector.java | 6 ++++++
.../nifi/tests/system/connectors/ConnectorDrainIT.java | 16 ++++++++++++++++
3 files changed, 28 insertions(+)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/facades/standalone/StandaloneProcessorLifecycle.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/facades/standalone/StandaloneProcessorLifecycle.java
index bd83cbe3e70..79b870882a5 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/facades/standalone/StandaloneProcessorLifecycle.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/facades/standalone/StandaloneProcessorLifecycle.java
@@ -67,6 +67,12 @@ public class StandaloneProcessorLifecycle implements
ProcessorLifecycle {
@Override
public CompletableFuture<Void> stop() {
+ final ScheduledState scheduledState =
processorNode.getPhysicalScheduledState();
+ if (scheduledState == ScheduledState.STOPPED || scheduledState ==
ScheduledState.DISABLED) {
+ logger.debug("Not stopping Processor {} because its state is {}",
processorNode, scheduledState);
+ return CompletableFuture.completedFuture(null);
+ }
+
return processScheduler.stopProcessor(processorNode,
ProcessorStopLifecycleMethods.TRIGGER_ALL);
}
diff --git
a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/connectors/tests/system/GatedDataQueuingConnector.java
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/connectors/tests/system/GatedDataQueuingConnector.java
index 9abcbf87c59..50cb08ffeaa 100644
---
a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/connectors/tests/system/GatedDataQueuingConnector.java
+++
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/connectors/tests/system/GatedDataQueuingConnector.java
@@ -29,6 +29,7 @@ import
org.apache.nifi.components.connector.components.FlowContext;
import org.apache.nifi.components.connector.util.VersionedFlowUtils;
import org.apache.nifi.flow.Bundle;
import org.apache.nifi.flow.Position;
+import org.apache.nifi.flow.ScheduledState;
import org.apache.nifi.flow.VersionedExternalFlow;
import org.apache.nifi.flow.VersionedProcessGroup;
import org.apache.nifi.flow.VersionedProcessor;
@@ -85,6 +86,11 @@ public class GatedDataQueuingConnector extends
AbstractConnector {
final VersionedProcessor terminate =
VersionedFlowUtils.addProcessor(rootGroup,
"org.apache.nifi.processors.tests.system.TerminateFlowFile",
bundle, "TerminateFlowFile", new Position(0, 0));
+ final VersionedProcessor disabled =
VersionedFlowUtils.addProcessor(rootGroup,
+ "org.apache.nifi.processors.tests.system.GenerateFlowFile",
bundle, "Disabled GenerateFlowFile", new Position(0, 200));
+ disabled.setScheduledState(ScheduledState.DISABLED);
+ disabled.setAutoTerminatedRelationships(Set.of("success"));
+
VersionedFlowUtils.addConnection(rootGroup,
VersionedFlowUtils.createConnectableComponent(generate),
VersionedFlowUtils.createConnectableComponent(terminate),
Set.of("success"));
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ConnectorDrainIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ConnectorDrainIT.java
index 9f0321e8735..7864bf53758 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ConnectorDrainIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ConnectorDrainIT.java
@@ -18,10 +18,14 @@
package org.apache.nifi.tests.system.connectors;
import org.apache.nifi.components.connector.ConnectorState;
+import org.apache.nifi.controller.ScheduledState;
import org.apache.nifi.tests.system.NiFiSystemIT;
import org.apache.nifi.toolkit.client.NiFiClientException;
import org.apache.nifi.web.api.entity.ConnectorEntity;
+import org.apache.nifi.web.api.entity.ProcessGroupFlowEntity;
+import org.apache.nifi.web.api.entity.ProcessorEntity;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -38,6 +42,7 @@ public class ConnectorDrainIT extends NiFiSystemIT {
private static final Logger logger =
LoggerFactory.getLogger(ConnectorDrainIT.class);
@Test
+ @Timeout(60)
public void testDrainFlowFiles() throws NiFiClientException, IOException,
InterruptedException {
final File gateFile = new
File(getNiFiInstance().getInstanceDirectory(), "gate-file.txt");
gateFile.deleteOnExit();
@@ -65,6 +70,17 @@ public class ConnectorDrainIT extends NiFiSystemIT {
getClientUtil().stopConnector(connectorId);
getClientUtil().waitForConnectorStopped(connectorId);
+ final ProcessGroupFlowEntity flow =
getNifiClient().getConnectorClient().getFlow(connectorId);
+ ProcessorEntity disabledProcessor = null;
+ for (final ProcessorEntity processor :
flow.getProcessGroupFlow().getFlow().getProcessors()) {
+ if
(ScheduledState.DISABLED.name().equals(processor.getComponent().getState())) {
+ disabledProcessor = processor;
+ break;
+ }
+ }
+
+ assertNotNull(disabledProcessor);
+
final int queuedCountBeforeDrain =
getConnectorQueuedFlowFileCount(connectorId);
logger.info("Queued FlowFile count before drain: {}",
queuedCountBeforeDrain);
assertTrue(queuedCountBeforeDrain > 0);
