markap14 commented on code in PR #11647:
URL: https://github.com/apache/nifi/pull/11647#discussion_r3967454084
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorNode.java:
##########
@@ -395,22 +403,77 @@ public void inheritConfiguration(final
List<VersionedConfigurationStep> activeCo
// two lists actually diverge.
applyUpdate(inheritContext);
- // Tear down the working context that applyUpdate created aliased to
active, and rebuild it around an
- // independent configuration seeded from migratedWorkingProperties.
Then fire onConfigurationStepConfigured
- // for every step so renamed steps trigger the flow-builder callback
under their new name and any
- // value-derived flow state (resolved asset paths, secret values,
etc.) is populated against the fresh
- // working context.
- destroyWorkingContext();
+ // Replace the working context that applyUpdate created aliased to
active with an independent context
+ // seeded from migratedWorkingProperties. Then fire
onConfigurationStepConfigured for every step so
+ // renamed steps trigger the flow-builder callback under their new
name and any value-derived flow
+ // state (resolved asset paths, secret values, etc.) is populated
against the fresh working context.
final MutableConnectorConfigurationContext workingConfigContext =
createConfigurationContext(migratedWorkingProperties);
- workingFlowContext =
flowContextFactory.createWorkingFlowContext(identifier,
connectorDetails.getComponentLog(), workingConfigContext, flowContextBundle);
+ final WorkingFlowContextState independentWorkingContextState =
installReplacementWorkingFlowContext(workingConfigContext, flowContextBundle,
true);
+ final FrameworkFlowContext independentWorkingContext =
independentWorkingContextState.getContext();
+
getComponentLog().info("Working Flow Context has been rebuilt with
independent configuration");
- for (final String stepName : migratedWorkingProperties.keySet()) {
- notifyStepConfigured(stepName);
+
+ try {
+ for (final String stepName : migratedWorkingProperties.keySet()) {
+ notifyStepConfigured(stepName, independentWorkingContext);
+ }
+ } finally {
+ releaseWorkingFlowContext(independentWorkingContextState);
}
logger.debug("Successfully inherited configuration for {}", this);
}
+ /**
+ * Removes the current working process group before creating its
replacement. Working-context copies reuse the same
+ * connection identifiers as the active flow, and the cluster load-balance
client registry allows only one
+ * registration per connection ID, so the previous group must be gone
before the factory copies the active group.
+ * The published working context is never set to null: callers that arrive
while the previous group is being
+ * destroyed still see that context, and callers that arrive while the
replacement is created wait on the monitor.
+ */
+ private WorkingFlowContextState installReplacementWorkingFlowContext(final
MutableConnectorConfigurationContext configurationContext, final Bundle bundle,
final boolean incrementUseCount) {
+ final WorkingFlowContextState previousWorkingFlowContextState;
+ final boolean destroyPrevious;
+ synchronized (workingFlowContextLock) {
+ while (workingContextReplacementInProgress) {
+ try {
+ workingFlowContextLock.wait();
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException("Interrupted while waiting
to replace the working flow context of " + this, e);
+ }
+ }
+
+ workingContextReplacementInProgress = true;
+ previousWorkingFlowContextState = workingFlowContextState;
+ previousWorkingFlowContextState.retire();
+ destroyPrevious =
previousWorkingFlowContextState.claimDestruction();
Review Comment:
[Cursor Grok 4.6] @pvillard31 No. Working-context copies reuse the same
connection IDs as the active flow, and the cluster load-balance client registry
allows only one registration per ID. If we wait for the last lease to drop
before destroying, `createWorkingFlowContext` tries to copy the active group
while the previous working group is still registered and fails with
`IllegalStateException: Connection with ID ... is already registered` (that is
what broke `ClusteredConnectorTroubleshootingIT` on the first CI run).
`applyUpdate` also still holds a lease on the previous holder when it
recreates, so waiting for `useCount == 0` would deadlock that path.
`claimDestruction()` here is what lets recreation tear down the previous
process group immediately; `decrementUseCount()` still uses it so a later
release does not purge twice.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorNode.java:
##########
@@ -640,29 +611,144 @@ public void
testReplaceWorkingConfigurationDoesNotFireWhenUnchanged() throws Flo
connectorNode.replaceWorkingConfiguration("step1",
createStepConfiguration(Map.of("propA", "valueA")));
-
assertFalse(trackingConnector.wasOnPropertyGroupConfiguredCalled("step1"));
+
assertFalse(trackingConnector.wasOnConfigurationStepConfiguredCalled("step1"));
}
@Test
- public void
testDiscardWorkingConfigurationFiresOnConfiguredForEveryWorkingStep() throws
FlowUpdateException {
- final TrackingConnector trackingConnector = new TrackingConnector();
- final StandardConnectorNode connectorNode =
createConnectorNode(trackingConnector);
+ public void
testReplaceWorkingConfigurationWaitsForWorkingContextRecreation() throws
Exception {
+ final BlockingWorkingFlowContextFactory blockingFlowContextFactory =
new BlockingWorkingFlowContextFactory(flowContextFactory);
+ flowContextFactory = blockingFlowContextFactory;
+ final StandardConnectorNode connectorNode = createConnectorNode(new
TrackingConnector());
connectorNode.transitionStateForUpdating();
connectorNode.prepareForUpdate();
- connectorNode.setConfiguration("step1",
createStepConfiguration(Map.of("propA", "valueA")));
- connectorNode.setConfiguration("step2",
createStepConfiguration(Map.of("propB", "valueB")));
+ connectorNode.setConfiguration("step1",
createStepConfiguration(Map.of("propA", "oldA")));
connectorNode.applyUpdate();
+ blockingFlowContextFactory.blockNextWorkingContextCreation();
- trackingConnector.reset();
+ final ExecutorService executor = Executors.newFixedThreadPool(2);
+ try {
+ final Future<?> recreationFuture =
executor.submit(connectorNode::recreateWorkingFlowContext);
+
assertTrue(blockingFlowContextFactory.awaitWorkingContextCreation(5,
TimeUnit.SECONDS));
+
+ final CountDownLatch replaceStarted = new CountDownLatch(1);
+ final Future<?> replacementFuture = executor.submit(() -> {
+ replaceStarted.countDown();
+ connectorNode.replaceWorkingConfiguration("step1",
createStepConfiguration(Map.of("propA", "newA")));
+ return null;
+ });
+ assertTrue(replaceStarted.await(5, TimeUnit.SECONDS));
- // Recreating the working flow context from the active flow must fire
onConfigurationStepConfigured
- // for every working configuration step so that flow parameters
derived from the configuration
- // (resolved asset paths, secrets, etc.) are refreshed.
- connectorNode.discardWorkingConfiguration();
+ try {
+ assertThrows(TimeoutException.class, () ->
replacementFuture.get(STOP_NOT_EXPECTED_MILLIS, TimeUnit.MILLISECONDS));
+ } finally {
+ blockingFlowContextFactory.releaseWorkingContextCreation();
+ }
+
+ recreationFuture.get(5, TimeUnit.SECONDS);
+ replacementFuture.get(5, TimeUnit.SECONDS);
+ } finally {
+ executor.shutdownNow();
+ assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
+ }
+
+ final ConnectorConfiguration workingConfiguration =
connectorNode.getWorkingFlowContext().getConfigurationContext().toConnectorConfiguration();
+ final NamedStepConfiguration namedStep =
workingConfiguration.getNamedStepConfigurations().iterator().next();
+ assertEquals(Map.of("propA", new StringLiteralValue("newA")),
namedStep.configuration().getPropertyValues());
+ }
+
+ @Test
+ @Timeout(10)
+ public void testRecreationRefreshDoesNotOverwriteConcurrentReplace()
throws Exception {
+ final CountDownLatch refreshStarted = new CountDownLatch(1);
+ final CountDownLatch permitRefresh = new CountDownLatch(1);
+ final AtomicBoolean blockNextRefresh = new AtomicBoolean();
+ final AtomicReference<String> refreshingStepName = new
AtomicReference<>();
+ final TrackingConnector trackingConnector = new TrackingConnector() {
+ @Override
+ protected void onStepConfigured(final String stepName, final
FlowContext workingContext) throws FlowUpdateException {
+ if (!blockNextRefresh.compareAndSet(true, false)) {
+ return;
+ }
+
+ refreshingStepName.set(stepName);
+ refreshStarted.countDown();
+ try {
+ permitRefresh.await();
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new FlowUpdateException("Interrupted while waiting
to refresh the working flow context", e);
+ }
+ }
+ };
+
+ final StandardConnectorNode connectorNode =
createConnectorNode(trackingConnector);
+ connectorNode.transitionStateForUpdating();
+ connectorNode.prepareForUpdate();
+ connectorNode.setConfiguration("step1",
createStepConfiguration(Map.of("propA", "oldA")));
+ connectorNode.setConfiguration("step2",
createStepConfiguration(Map.of("propA", "oldB")));
+ connectorNode.applyUpdate();
+ blockNextRefresh.set(true);
+
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ final Future<?> recreationFuture =
executor.submit(connectorNode::recreateWorkingFlowContext);
+ final String replacedStepName;
+ try {
+ assertTrue(refreshStarted.await(5, TimeUnit.SECONDS));
+ replacedStepName = "step1".equals(refreshingStepName.get()) ?
"step2" : "step1";
+ connectorNode.replaceWorkingConfiguration(replacedStepName,
createStepConfiguration(Map.of("propA", "newA")));
+ } finally {
+ permitRefresh.countDown();
+ }
-
assertTrue(trackingConnector.wasOnPropertyGroupConfiguredCalled("step1"));
-
assertTrue(trackingConnector.wasOnPropertyGroupConfiguredCalled("step2"));
+ recreationFuture.get(5, TimeUnit.SECONDS);
+
+ final ConnectorConfiguration workingConfiguration =
connectorNode.getWorkingFlowContext().getConfigurationContext().toConnectorConfiguration();
+ final NamedStepConfiguration namedStep =
workingConfiguration.getNamedStepConfiguration(replacedStepName);
+ assertEquals(Map.of("propA", new StringLiteralValue("newA")),
namedStep.configuration().getPropertyValues());
+ } finally {
+ executor.shutdownNow();
+ assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
+ }
+ }
+
+ @Test
+ @Timeout(10)
+ public void
testOnConfigurationStepConfiguredCanWaitForWorkingContextRecreation() throws
Exception {
Review Comment:
[Cursor Grok 4.6] @pvillard31 That is not the contract this test is
checking. It only verifies that `onConfigurationStepConfigured` can wait for
`recreateWorkingFlowContext` on another thread without deadlocking. Recreation
must destroy the previous working process group before creating the replacement
(same connection-ID constraint as above), so the old group can already be gone
while the callback is still on the stack. The callback is handed the old
`FlowContext` as an argument; it must not assume the managed process group
still exists after it triggers recreation.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorNode.java:
##########
@@ -640,29 +611,144 @@ public void
testReplaceWorkingConfigurationDoesNotFireWhenUnchanged() throws Flo
connectorNode.replaceWorkingConfiguration("step1",
createStepConfiguration(Map.of("propA", "valueA")));
-
assertFalse(trackingConnector.wasOnPropertyGroupConfiguredCalled("step1"));
+
assertFalse(trackingConnector.wasOnConfigurationStepConfiguredCalled("step1"));
}
@Test
Review Comment:
[Cursor Grok 4.6] @pvillard31 Yes. Added `@Timeout(10)` on this test to
match the other concurrency tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
