pvillard31 commented on code in PR #11647:
URL: https://github.com/apache/nifi/pull/11647#discussion_r3968059982
##########
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:
OK fair enough, it clarifies that the lease protects against duplicate
cleanup, not continued availability of the managed process group.
--
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]
