This is an automated email from the ASF dual-hosted git repository.
pvillard31 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 d512716ed9e NIFI-16055: Avoid starting processors and controller
services in Troubleshooting Connectors when autoResumeState=false (#11375)
d512716ed9e is described below
commit d512716ed9e2b0ac766259bcf84498fa777a675e
Author: Mark Payne <[email protected]>
AuthorDate: Tue Jun 30 02:57:33 2026 -0400
NIFI-16055: Avoid starting processors and controller services in
Troubleshooting Connectors when autoResumeState=false (#11375)
---
.../nifi/groups/NonStartingComponentScheduler.java | 76 ++++++++++++++
.../apache/nifi/groups/StandardProcessGroup.java | 11 +-
.../connectors/ClusteredConnectorAutoResumeIT.java | 33 ++++++
.../system/connectors/ConnectorAutoResumeIT.java | 112 +++++++++++++++++++--
4 files changed, 223 insertions(+), 9 deletions(-)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/NonStartingComponentScheduler.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/NonStartingComponentScheduler.java
new file mode 100644
index 00000000000..673e59b3948
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/NonStartingComponentScheduler.java
@@ -0,0 +1,76 @@
+/*
+ * 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.groups;
+
+import org.apache.nifi.connectable.Connectable;
+import org.apache.nifi.controller.ReportingTaskNode;
+import org.apache.nifi.controller.service.ControllerServiceNode;
+import org.apache.nifi.controller.service.ControllerServiceProvider;
+import org.apache.nifi.registry.flow.mapping.VersionedComponentStateLookup;
+
+import java.util.Collection;
+
+/**
+ * A {@link ComponentScheduler} that synchronizes a flow's structure while
leaving every component stopped or disabled
+ * rather than resuming the running or enabled state recorded in the proposed
flow. Components that the proposed flow marks
+ * as DISABLED remain DISABLED; components marked ENABLED or RUNNING are left
in a stopped (but enabled) state. No
+ * Processor, Port, Reporting Task, stateless group, or Controller Service is
started or enabled.
+ *
+ * <p>This differs from {@link ComponentScheduler#NOP_SCHEDULER}, which
performs no scheduling action at all: because a
+ * newly created component defaults to a stopped state, the no-op scheduler
silently turns a component that the proposed
+ * flow marks DISABLED into a stopped (enabled) component. This scheduler
still applies the enabled/disabled transition so
+ * that a DISABLED component is faithfully restored as DISABLED, while
suppressing only the start or enable that would
+ * resume a component to a running or active state.</p>
+ *
+ * <p>It is used wherever a flow's components must be synchronized without
resuming their running state, such as restoring
+ * a Connector's managed flow during NiFi startup when {@code
nifi.flowcontroller.autoResumeState} is {@code false}.</p>
+ */
+public class NonStartingComponentScheduler extends AbstractComponentScheduler {
+
+ public NonStartingComponentScheduler(final ControllerServiceProvider
controllerServiceProvider, final VersionedComponentStateLookup stateLookup) {
+ super(controllerServiceProvider, stateLookup);
+ }
+
+ @Override
+ public void startComponent(final Connectable component) {
+ }
+
+ @Override
+ public void startReportingTask(final ReportingTaskNode reportingTask) {
+ }
+
+ @Override
+ public void startStatelessGroup(final ProcessGroup group) {
+ }
+
+ @Override
+ public void enableControllerServicesAsync(final
Collection<ControllerServiceNode> controllerServices) {
+ }
+
+ @Override
+ protected void startNow(final Connectable component) {
+ }
+
+ @Override
+ protected void startNow(final ReportingTaskNode reportingTask) {
+ }
+
+ @Override
+ protected void startNow(final ProcessGroup statelessGroup) {
+ }
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
index c4da90a3537..dc46a099703 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
@@ -218,6 +218,7 @@ public final class StandardProcessGroup implements
ProcessGroup {
private final DataValve dataValve;
private final Long nifiPropertiesBackpressureCount;
private final String nifiPropertiesBackpressureSize;
+ private final boolean autoResumeState;
private final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
private final Lock readLock = rwLock.readLock();
@@ -272,6 +273,8 @@ public final class StandardProcessGroup implements
ProcessGroup {
this.logFileSuffix = null;
// save only the nifi properties needed, and account for the
possibility those properties are missing
+ this.autoResumeState = nifiProperties == null ?
NiFiProperties.DEFAULT_AUTO_RESUME_STATE : nifiProperties.getAutoResumeState();
+
if (nifiProperties == null) {
nifiPropertiesBackpressureCount = DEFAULT_BACKPRESSURE_OBJECT;
nifiPropertiesBackpressureSize = DEFAULT_BACKPRESSURE_DATA_SIZE;
@@ -3989,7 +3992,13 @@ public final class StandardProcessGroup implements
ProcessGroup {
// their Connections upon restore.
final ComponentIdGenerator idGenerator = (proposedId, instanceId,
destinationGroupId) -> instanceId;
final VersionedComponentStateLookup stateLookup =
VersionedComponentStateLookup.IDENTITY_LOOKUP;
- final ComponentScheduler componentScheduler = new
DefaultComponentScheduler(controllerServiceProvider, stateLookup);
+
+ // When nifi.flowcontroller.autoResumeState is false, restore the flow
structure (so queued FlowFiles can be
+ // re-associated with their Connections) without resuming any
component to its previously running or enabled
+ // state. Otherwise, restore each component to the running/enabled
state captured in the persisted flow.
+ final ComponentScheduler componentScheduler = autoResumeState
+ ? new DefaultComponentScheduler(controllerServiceProvider,
stateLookup)
+ : new NonStartingComponentScheduler(controllerServiceProvider,
stateLookup);
final FlowSynchronizationOptions synchronizationOptions = new
FlowSynchronizationOptions.Builder()
.componentIdGenerator(idGenerator)
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ClusteredConnectorAutoResumeIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ClusteredConnectorAutoResumeIT.java
new file mode 100644
index 00000000000..338b2f1953d
--- /dev/null
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ClusteredConnectorAutoResumeIT.java
@@ -0,0 +1,33 @@
+/*
+ * 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.tests.system.connectors;
+
+import org.apache.nifi.tests.system.NiFiInstanceFactory;
+
+/**
+ * Clustered variant of {@link ConnectorAutoResumeIT}. Re-executes every
auto-resume test defined on the parent class
+ * against a two-node NiFi cluster so that the cluster-replication and
node-restart paths are exercised in addition to
+ * the standalone paths.
+ */
+public class ClusteredConnectorAutoResumeIT extends ConnectorAutoResumeIT {
+
+ @Override
+ public NiFiInstanceFactory getInstanceFactory() {
+ return createTwoNodeInstanceFactory();
+ }
+}
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ConnectorAutoResumeIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ConnectorAutoResumeIT.java
index 2f85b4f7b20..a921f445cee 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ConnectorAutoResumeIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ConnectorAutoResumeIT.java
@@ -18,21 +18,32 @@
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.util.NiFiProperties;
+import org.apache.nifi.web.api.dto.flow.FlowDTO;
import org.apache.nifi.web.api.entity.ConnectorEntity;
+import org.apache.nifi.web.api.entity.ControllerServiceEntity;
+import org.apache.nifi.web.api.entity.ProcessGroupEntity;
+import org.apache.nifi.web.api.entity.ProcessGroupFlowEntity;
+import org.apache.nifi.web.api.entity.ProcessorEntity;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
/**
- * System test verifying that a running Connector is not auto-resumed after a
restart
- * when the {@code nifi.flowcontroller.autoResumeState} property is set to
{@code false}.
+ * System tests verifying that Connectors and the components inside their
managed flows are not
+ * auto-resumed after a restart when the {@code
nifi.flowcontroller.autoResumeState} property is
+ * set to {@code false}.
*/
public class ConnectorAutoResumeIT extends NiFiSystemIT {
@@ -58,19 +69,104 @@ public class ConnectorAutoResumeIT extends NiFiSystemIT {
getClientUtil().startConnector(connectorId);
getClientUtil().waitForConnectorState(connectorId,
ConnectorState.RUNNING);
- logger.info("Connector {} is RUNNING", connectorId);
- getNiFiInstance().stop();
- logger.info("NiFi stopped");
+ restartWithAutoResumeStateDisabled();
+
+ final ConnectorEntity connectorAfterRestart =
getNifiClient().getConnectorClient().getConnector(connectorId);
+ final String stateAfterRestart =
connectorAfterRestart.getComponent().getState();
+
+ assertEquals(ConnectorState.STOPPED.name(), stateAfterRestart);
+ }
+
+ /**
+ * Verifies that when a Connector is in Troubleshooting mode with
processors running and controller services
+ * enabled, restarting NiFi with {@code
nifi.flowcontroller.autoResumeState=false} leaves the Connector in
+ * Troubleshooting mode but does not resume any of the processors or
controller services inside the managed flow.
+ */
+ @Test
+ public void testTroubleshootingConnectorNotResumedWhenAutoResumeDisabled()
throws NiFiClientException, IOException, InterruptedException {
+ final ConnectorEntity connector =
getClientUtil().createConnector("ComponentLifecycleConnector");
+ final String connectorId = connector.getId();
+
+ getClientUtil().applyConnectorUpdate(connector);
+ getClientUtil().waitForValidConnector(connectorId);
+
+ getClientUtil().startConnector(connectorId);
+ getClientUtil().waitForConnectorState(connectorId,
ConnectorState.RUNNING);
+
+ // Entering Troubleshooting from RUNNING leaves the managed flow's
processors running and its controller services
+ // enabled, so the post-restart assertions have running components to
verify against.
+ getClientUtil().enterTroubleshooting(connectorId);
+ getClientUtil().waitForConnectorState(connectorId,
ConnectorState.TROUBLESHOOTING);
+
+ restartWithAutoResumeStateDisabled();
+
+ // The Troubleshooting state itself is preserved across restarts;
autoResumeState does not transition the
+ // Connector out of Troubleshooting mode.
+ final ConnectorEntity connectorAfterRestart =
getNifiClient().getConnectorClient().getConnector(connectorId);
+ final String connectorStateAfterRestart =
connectorAfterRestart.getComponent().getState();
+ assertEquals(ConnectorState.TROUBLESHOOTING.name(),
connectorStateAfterRestart);
+
+ final List<ProcessorEntity> processorsAfterRestart = new ArrayList<>();
+ collectProcessors(connectorId, null, processorsAfterRestart);
+ assertFalse(processorsAfterRestart.isEmpty());
+
+ for (final ProcessorEntity processor : processorsAfterRestart) {
+ final String processorState = processor.getComponent().getState();
+ assertNotEquals(ScheduledState.RUNNING.name(), processorState);
+ }
+
+ final String managedGroupId =
connectorAfterRestart.getComponent().getManagedProcessGroupId();
+ final List<String> controllerServiceIds =
collectAllControllerServiceIds(managedGroupId);
+ assertFalse(controllerServiceIds.isEmpty());
+
+ for (final String serviceId : controllerServiceIds) {
+ final ControllerServiceEntity serviceEntity =
getNifiClient().getControllerServicesClient().getControllerService(serviceId);
+ final String serviceState =
serviceEntity.getComponent().getState();
+ assertNotEquals("ENABLED", serviceState);
+ }
+ }
+ private void restartWithAutoResumeStateDisabled() throws IOException {
+ getNiFiInstance().stop();
getNiFiInstance().setProperty(NiFiProperties.AUTO_RESUME_STATE,
"false");
getNiFiInstance().start();
+
setupClient();
+
+ if (getNiFiInstance().isClustered()) {
+ waitForAllNodesConnected();
+ }
+
logger.info("NiFi restarted with autoResumeState=false");
+ }
- final ConnectorEntity connectorAfterRestart =
getNifiClient().getConnectorClient().getConnector(connectorId);
- final String stateAfterRestart =
connectorAfterRestart.getComponent().getState();
+ private void collectProcessors(final String connectorId, final String
groupId, final List<ProcessorEntity> collected) throws NiFiClientException,
IOException {
+ final ProcessGroupFlowEntity entity = (groupId == null)
+ ? getNifiClient().getConnectorClient().getFlow(connectorId)
+ : getNifiClient().getConnectorClient().getFlow(connectorId,
groupId);
- assertEquals(ConnectorState.STOPPED.name(), stateAfterRestart);
+ final FlowDTO flow = entity.getProcessGroupFlow().getFlow();
+ collected.addAll(flow.getProcessors());
+
+ for (final ProcessGroupEntity child : flow.getProcessGroups()) {
+ collectProcessors(connectorId, child.getId(), collected);
+ }
+ }
+
+ private List<String> collectAllControllerServiceIds(final String groupId)
throws NiFiClientException, IOException {
+ final List<String> serviceIds = new ArrayList<>();
+ for (final ControllerServiceEntity entity :
getNifiClient().getFlowClient().getControllerServices(groupId).getControllerServices())
{
+ if (entity.getComponent() != null) {
+ serviceIds.add(entity.getId());
+ }
+ }
+
+ final ProcessGroupFlowEntity groupFlow =
getNifiClient().getFlowClient().getProcessGroup(groupId);
+ for (final ProcessGroupEntity child :
groupFlow.getProcessGroupFlow().getFlow().getProcessGroups()) {
+ serviceIds.addAll(collectAllControllerServiceIds(child.getId()));
+ }
+
+ return serviceIds;
}
}
