pvillard31 commented on code in PR #11647:
URL: https://github.com/apache/nifi/pull/11647#discussion_r3968054076
##########
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:
OK I see!
--
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]
