bobpaulin commented on code in PR #11872:
URL: https://github.com/apache/nifi/pull/11872#discussion_r4483079746
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorNode.java:
##########
@@ -1619,6 +1783,623 @@ public boolean isExtensionMissing() {
return extensionMissing;
}
+ @Override
+ public boolean isPendingFlowEvolution() {
+ synchronized (flowEvolutionLock) {
+ if (extensionMissing) {
+ return false;
+ }
+
+ final ConnectorFlowSnapshot snapshot = authoritativeFlowSnapshot;
+ final Bundle snapshotBundle = snapshot == null ? null :
snapshot.bundle();
+ return (snapshotBundle != null && !sameBundle(snapshotBundle,
bundleCoordinate)) || isFlowEvolutionPhaseOneComplete();
+ }
+ }
+
+ @Override
+ public FlowEvolutionStatus getFlowEvolutionStatus() {
+ return hasFlowEvolutionFailed() ? FlowEvolutionStatus.FAILED :
flowEvolutionStatus.get();
+ }
+
+ boolean isFlowEvolutionPhaseOneComplete() {
+ synchronized (flowEvolutionLock) {
+ return hasCompletedFlowEvolutionHookProgress()
+ && (hasFlowEvolutionFailed() ||
isFlowEvolutionInProgress());
+ }
+ }
+
+ private boolean hasCompletedFlowEvolutionHookProgress() {
+ final FlowEvolutionStatus status = flowEvolutionStatus.get();
+ return status == FlowEvolutionStatus.PHASE_ONE_COMPLETE || status ==
FlowEvolutionStatus.PHASE_TWO_COMPLETE;
+ }
+
+ boolean hasFlowEvolutionFailed() {
+ return flowEvolutionFailure.get() != null;
+ }
+
+ private void verifyFlowEvolutionActive() throws FlowUpdateException {
+ if (hasFlowEvolutionFailed() || getCurrentState() !=
ConnectorState.DRAINING) {
+ throw new FlowUpdateException("Flow evolution was cancelled for "
+ this);
+ }
+ }
+
+ private VersionedProcessGroup getAuthoritativeManagedProcessGroup() {
+ final ConnectorFlowSnapshot snapshot = authoritativeFlowSnapshot;
+ return snapshot == null ? null : snapshot.flow();
+ }
+
+ @Override
+ public ConnectorFlowSnapshot getAuthoritativeFlowSnapshot() {
+ return authoritativeFlowSnapshot;
+ }
+
+ @Override
+ public boolean restorePendingFlowEvolution(final VersionedConnector
persisted, final List<VersionedConfigurationStep> activeConfiguration,
+ final List<VersionedConfigurationStep> workingConfiguration)
throws FlowUpdateException {
+ final Bundle managedFlowBundle = resolveManagedFlowBundle(persisted);
+ final VersionedProcessGroup persistedManagedGroup =
persisted.getManagedProcessGroup();
+ final Map<String, SecretReference> parameterSecretReferences =
ParameterSecretReferences.convertManagedParameterSecretReferences(persisted);
+ // Components in the persisted group may name NAR versions that are no
longer installed. A missing coordinate
+ // is rewritten only when exactly one compatible bundle is installed.
A ghost has no initialization context,
+ // and its coordinates have to stay as persisted.
+ if (!extensionMissing) {
+ resolvePersistedBundles(persistedManagedGroup);
+ }
+
+ if (extensionMissing && persistedManagedGroup != null) {
+ // Adopt before inheritConfiguration. A flow save with no snapshot
omits the managed group, and inherit
+ // rebuilds the live group before the persisted one is installed
below.
+ adoptAuthoritativeSnapshot(persistedManagedGroup,
persisted.getManagedParameterContext(), parameterSecretReferences,
managedFlowBundle);
+ return false;
+ }
+
+ if (!extensionMissing && !sameManagedBundle(managedFlowBundle) &&
persistedManagedGroup != null) {
+ // Skip inheritConfiguration. Its applyUpdate would replace the
pre-upgrade group with the new bundle's
+ // flow. The restored group keeps the previous bundle, so the next
start evolves it.
+ restorePersistedFlow(persistedManagedGroup,
persisted.getManagedParameterContext(), parameterSecretReferences, true);
+ loadConfigurationsWithoutRebuildingFlow(activeConfiguration,
workingConfiguration);
+ adoptAuthoritativeSnapshot(persistedManagedGroup,
persisted.getManagedParameterContext(), parameterSecretReferences,
managedFlowBundle);
+ return true;
+ }
+
+ return false;
+ }
+
+ @Override
+ public void restorePersistedManagedFlow(final VersionedConnector
persisted) {
+ final Bundle managedFlowBundle = resolveManagedFlowBundle(persisted);
+ final VersionedProcessGroup persistedManagedGroup =
persisted.getManagedProcessGroup();
+ final Map<String, SecretReference> parameterSecretReferences =
ParameterSecretReferences.convertManagedParameterSecretReferences(persisted);
+
+ if (extensionMissing && persistedManagedGroup != null) {
+ // Install the persisted flow while ghosted so queues exist before
the FlowFile repository recovers.
+ // Components stay stopped: a ghost cannot start, and resuming
would run them under a missing extension.
+ try {
+ restorePersistedFlow(persistedManagedGroup,
persisted.getManagedParameterContext(), parameterSecretReferences, false);
+ } catch (final Exception e) {
+ logger.error("{} failed to restore the persisted managed flow
while ghosted; "
+ + "queued FlowFiles may not rejoin until NiFi restarts
after the Connector becomes loadable again", this, e);
+ markManagedFlowMissingAtStartup();
+ }
+ adoptAuthoritativeSnapshot(persistedManagedGroup,
persisted.getManagedParameterContext(), parameterSecretReferences,
managedFlowBundle);
+ return;
+ }
+
+ if (sameManagedBundle(managedFlowBundle) && persistedManagedGroup !=
null) {
+ // The bundles match, so this is a restart rather than an upgrade.
inheritConfiguration has already
+ // replaced the group; install the persisted one so connection
identifiers still match the flow file.
+ restorePersistedFlow(persistedManagedGroup,
persisted.getManagedParameterContext(), parameterSecretReferences, true);
+ adoptAuthoritativeSnapshot(persistedManagedGroup,
persisted.getManagedParameterContext(), parameterSecretReferences,
managedFlowBundle);
+ return;
Review Comment:
Agree will fix.
--
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]
