You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by bb...@apache.org on 2023/05/19 12:24:25 UTC
[nifi] branch main updated: NIFI-11563: Allowing source connectables to be restarted on new connections in the StandardVersionedComponentSynchronizer (#7261)
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 28f003d886 NIFI-11563: Allowing source connectables to be restarted on new connections in the StandardVersionedComponentSynchronizer (#7261)
28f003d886 is described below
commit 28f003d8869d390b132204b04639c8db6015ba95
Author: Joe Gresock <jg...@gmail.com>
AuthorDate: Fri May 19 08:24:13 2023 -0400
NIFI-11563: Allowing source connectables to be restarted on new connections in the StandardVersionedComponentSynchronizer (#7261)
---
.../StandardVersionedComponentSynchronizer.java | 26 +++++++++++++++++++-
...StandardVersionedComponentSynchronizerTest.java | 28 +++++++++++++++++++++-
2 files changed, 52 insertions(+), 2 deletions(-)
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
index f634ef538d..62be6a1467 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
@@ -50,6 +50,7 @@ import org.apache.nifi.flow.BatchSize;
import org.apache.nifi.flow.Bundle;
import org.apache.nifi.flow.ComponentType;
import org.apache.nifi.flow.ConnectableComponent;
+import org.apache.nifi.flow.ConnectableComponentType;
import org.apache.nifi.flow.ParameterProviderReference;
import org.apache.nifi.flow.VersionedComponent;
import org.apache.nifi.flow.VersionedConnection;
@@ -3194,6 +3195,26 @@ public class StandardVersionedComponentSynchronizer implements VersionedComponen
}
}
+ private Set<Connectable> getUpstreamComponents(final VersionedConnection connection) {
+ if (connection == null) {
+ return Collections.emptySet();
+ }
+
+ final Set<Connectable> components = new HashSet<>();
+ findUpstreamComponents(connection, components);
+ return components;
+ }
+
+ private void findUpstreamComponents(final VersionedConnection connection, final Set<Connectable> components) {
+ final ConnectableComponent sourceConnectable = connection.getSource();
+ final Connectable source = context.getFlowManager().findConnectable(sourceConnectable.getId());
+ if (sourceConnectable.getType() == ConnectableComponentType.FUNNEL) {
+ source.getIncomingConnections().forEach(incoming -> findUpstreamComponents(incoming, components));
+ } else {
+ components.add(source);
+ }
+ }
+
@Override
public void synchronize(final Connection connection, final VersionedConnection proposedConnection, final ProcessGroup group, final FlowSynchronizationOptions synchronizationOptions)
throws FlowSynchronizationException, TimeoutException {
@@ -3205,7 +3226,10 @@ public class StandardVersionedComponentSynchronizer implements VersionedComponen
final long timeout = System.currentTimeMillis() + synchronizationOptions.getComponentStopTimeout().toMillis();
// Stop any upstream components so that we can update the connection
- final Set<Connectable> upstream = getUpstreamComponents(connection);
+ final Set<Connectable> upstream = new HashSet<>(getUpstreamComponents(connection));
+ if (connection == null) {
+ upstream.addAll(getUpstreamComponents(proposedConnection));
+ }
Set<Connectable> stoppedComponents;
try {
stoppedComponents = stopOrTerminate(upstream, timeout, synchronizationOptions);
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizerTest.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizerTest.java
index 042d8b68e5..2e356828c1 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizerTest.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizerTest.java
@@ -140,6 +140,7 @@ public class StandardVersionedComponentSynchronizerTest {
private CapturingScheduledStateChangeListener scheduledStateChangeListener;
private ControllerServiceNode controllerServiceNode;
private BundleCoordinate bundleCoordinate;
+ private FlowManager flowManager;
private final Set<String> queuesWithData = Collections.synchronizedSet(new HashSet<>());
private final Bundle bundle = new Bundle("group", "artifact", "version 1.0");
@@ -147,7 +148,7 @@ public class StandardVersionedComponentSynchronizerTest {
@BeforeEach
public void setup() {
final ExtensionManager extensionManager = Mockito.mock(ExtensionManager.class);
- final FlowManager flowManager = Mockito.mock(FlowManager.class);
+ flowManager = Mockito.mock(FlowManager.class);
controllerServiceProvider = Mockito.mock(ControllerServiceProvider.class);
final Function<ProcessorNode, ProcessContext> processContextFactory = proc -> Mockito.mock(ProcessContext.class);
final ReloadComponent reloadComponent = Mockito.mock(ReloadComponent.class);
@@ -459,6 +460,31 @@ public class StandardVersionedComponentSynchronizerTest {
verifyCallbackIndicatedRestart(processorA);
}
+ @Test
+ public void testSourceStoppedForNewConnection() throws FlowSynchronizationException, TimeoutException {
+
+ when(flowManager.findConnectable(processorA.getIdentifier())).thenReturn(processorA);
+
+ startProcessor(processorA, true);
+
+ final VersionedConnection versionedConnection = createMinimalVersionedConnection(processorA, processorB);
+ versionedConnection.setName("Hello");
+
+ synchronizer.synchronize(connectionAB, versionedConnection, group, synchronizationOptions);
+
+ // Ensure that we terminate the source
+ verify(group, times(1)).stopProcessor(processorA);
+
+ // Ensure that the update occurred
+ verify(connectionAB, times(1)).setName("Hello");
+
+ // Ensure that the source was stopped and restarted
+ verifyStopped(processorA);
+ verifyRestarted(processorA);
+
+ verifyCallbackIndicatedRestart(processorA);
+ }
+
@Test
public void testTimeoutStoppingSource() {
startProcessor(processorA, false);