pvillard31 commented on code in PR #11647:
URL: https://github.com/apache/nifi/pull/11647#discussion_r3967434955
##########
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:
Can this test also verify that the old working context is not destroyed
until onConfigurationStepConfigured() returns?
##########
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:
Should this concurrency test also use @Timeout(10) like the other new
concurrency tests?
##########
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:
Should destruction only be claimed when retire() reports that no callers are
using the context, so the last releaseWorkingFlowContext() destroys a leased
context?
--
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]
