mcgilman commented on code in PR #11677:
URL: https://github.com/apache/nifi/pull/11677#discussion_r4020218523
##########
nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connectors-listing/connectors-listing.effects.ts:
##########
@@ -433,6 +440,82 @@ export class ConnectorsListingEffects {
)
);
+ openChangeConnectorVersionDialog$ = createEffect(
+ () =>
+ this.actions$.pipe(
+ ofType(openChangeConnectorVersionDialog),
+ map((action) => action.request),
+ switchMap((request) =>
+
from(this.extensionTypesService.getConnectorVersionsForType(request.type,
request.bundle)).pipe(
+ map(
+ (response) =>
+ ({
+ fetchRequest: request,
+ componentVersions: response.connectorTypes
+ }) as OpenChangeComponentVersionDialogRequest
+ ),
+ catchError((errorResponse: HttpErrorResponse) => {
+ this.store.dispatch(
+ ErrorActions.snackBarError({
+ error:
this.errorHelper.getErrorString(errorResponse)
+ })
+ );
+ return EMPTY;
+ })
+ )
+ ),
+ tap((request) => {
+ const dialogRequest =
this.dialog.open(ChangeComponentVersionDialog, {
+ ...LARGE_DIALOG,
+ data: request,
+ autoFocus: false
+ });
+
+
dialogRequest.componentInstance.changeVersion.pipe(take(1)).subscribe((newVersion)
=> {
+ this.store.dispatch(
+ changeConnectorVersion({
+ request: {
+ id: request.fetchRequest.id,
+ uri: request.fetchRequest.uri,
+ payload: {
+ component: {
+ bundle: newVersion.bundle,
+ id: request.fetchRequest.id
+ },
+ revision: request.fetchRequest.revision
+ }
+ }
+ })
+ );
+ dialogRequest.close();
+ });
+ })
+ ),
+ { dispatch: false }
+ );
+
+ changeConnectorVersion$ = createEffect(() =>
+ this.actions$.pipe(
+ ofType(changeConnectorVersion),
+ map((action) => action.request),
+ concatMap((request) =>
Review Comment:
**Medium — prevent queuing a stale revision**
The dialog closes immediately after dispatch, but the Change Version table
action remains available while the existing `saving` state is true. A second
invocation can capture the same revision, and `concatMap` queues it until the
first succeeds, guaranteeing a stale-revision failure. Please wire the existing
listing `saving` state specifically to this new table action and disable it
while the request is pending. This does not require changing the shared dialog
or other table actions.
##########
nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java:
##########
@@ -5409,7 +5409,19 @@ public ConnectorDTO createConnectorDto(final
ConnectorNode connector) {
dto.setType(connector.getCanonicalClassName());
dto.setExtensionMissing(connector.isExtensionMissing());
- dto.setBundle(createBundleDto(connector.getBundleCoordinate()));
+ final BundleCoordinate bundleCoordinate =
connector.getBundleCoordinate();
+ final List<Bundle> availableBundles =
extensionManager.getBundles(connector.getCanonicalClassName());
+ int compatibleBundleCount = 0;
+ for (final Bundle bundle : availableBundles) {
+ final BundleCoordinate coordinate =
bundle.getBundleDetails().getCoordinate();
+ if (bundleCoordinate.getGroup().equals(coordinate.getGroup()) &&
bundleCoordinate.getId().equals(coordinate.getId())) {
+ compatibleBundleCount++;
+ }
+ }
+
+ dto.setMultipleVersionsAvailable(connector.isExtensionMissing() ?
compatibleBundleCount > 0 : compatibleBundleCount > 1);
Review Comment:
**Medium — merge version availability across cluster nodes**
This value is calculated independently on each node, but
`ConnectorEntityMerger` retains whichever node supplied the client entity. The
UI can therefore offer Change Version when another node reports no compatible
alternative. Cluster validation prevents a partial update, but the offered
operation will fail. Please merge this conservatively, as
`ProcessorEntityMerger` does: any null/false node result should produce false.
A mixed-node merger test would cover this.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorNode.java:
##########
@@ -1310,6 +1318,126 @@ public Connector getConnector() {
return connectorDetails.getConnector();
}
+ @Override
+ public void verifyCanReload() {
+ if (!isStopped()) {
+ throw new IllegalStateException("Cannot reload " + this + "
because its state is " + getCurrentState());
+ }
+ }
+
+ @Override
+ public void verifyCanUpdateBundle(final BundleCoordinate
incomingCoordinate) {
+ final BundleCoordinate currentCoordinate = getBundleCoordinate();
+ if
(!currentCoordinate.getGroup().equals(incomingCoordinate.getGroup()) ||
!currentCoordinate.getId().equals(incomingCoordinate.getId())) {
+ throw new IllegalArgumentException("Cannot update " + this + "
from " + currentCoordinate.getCoordinate() + " to "
+ + incomingCoordinate.getCoordinate() + " because the bundle
group and artifact must be unchanged");
+ }
+ }
+
+ @Override
+ public void replaceConnector(final Connector replacement, final
BundleCoordinate replacementCoordinate, final ComponentLog replacementLog)
throws FlowUpdateException {
+ verifyCanReload();
+
+ final FrameworkConnectorInitializationContext
replacementInitializationContext = new
StandardConnectorInitializationContext.Builder()
+ .identifier(identifier)
+ .name(name)
+ .componentLog(replacementLog)
+ .secretsManager(initializationContext.getSecretsManager())
+ .assetManager(initializationContext.getAssetManager())
+
.componentBundleLookup(initializationContext.getComponentBundleLookup())
+ .build();
+
+ try (final NarCloseable ignored =
NarCloseable.withComponentNarLoader(replacement.getClass().getClassLoader())) {
+ replacement.initialize(replacementInitializationContext);
+ }
+
+ final WorkingFlowContextState workingContextState =
acquireWorkingFlowContext();
+ final FrameworkFlowContext workingContext =
workingContextState.getContext();
+ boolean workingContextReleased = false;
+ try {
+ final Map<String, StepConfiguration> originalActiveConfiguration =
toConfigurationMap(activeFlowContext.getConfigurationContext().toConnectorConfiguration());
+ final Map<String, StepConfiguration> originalWorkingConfiguration
= workingContext == null ? originalActiveConfiguration
+ :
toConfigurationMap(workingContext.getConfigurationContext().toConnectorConfiguration());
+
+ final Map<String, StepConfiguration> activeConfiguration =
migrateProperties(replacement, originalActiveConfiguration);
+ final Map<String, StepConfiguration> workingConfiguration =
migrateProperties(replacement, originalWorkingConfiguration);
+ final Bundle bundle = new Bundle(replacementCoordinate.getGroup(),
replacementCoordinate.getId(), replacementCoordinate.getVersion());
+
+ replaceConfiguration(activeFlowContext.getConfigurationContext(),
activeConfiguration);
Review Comment:
**High — incomplete rollback during connector replacement**
Rollback currently covers only failures from `notifyStepConfigured()`.
`replaceConfiguration()` can fail during secret/asset resolution after earlier
steps have been mutated, and either flow-context `reload()` can also fail after
configuration or connector metadata has changed. These paths can leave migrated
configuration under the old connector or leave the node referencing the
replacement whose candidate classloader is subsequently closed. Please stage
the replacement and publish it atomically, or restore every mutated
field/context for all failure paths. Tests should inject failures during
configuration resolution and both context reloads.
##########
nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/ui/connector-table/connector-table.component.ts:
##########
@@ -157,6 +159,20 @@ export class ConnectorTable {
return isConnectorActionAllowed(entity, 'PURGE_FLOWFILES');
}
+ canChangeVersion(entity: ConnectorEntity): boolean {
+ const versionChangeEligibleStates = [
+ ConnectorState.STOPPED,
+ ConnectorState.UPDATED,
+ ConnectorState.UPDATE_FAILED
+ ];
+ return (
+ this.canRead(entity) &&
+ this.canModify(entity) &&
+ versionChangeEligibleStates.includes(entity.component.state as
ConnectorState) &&
Review Comment:
**Medium — account for active threads in eligible states**
`UPDATED` and `UPDATE_FAILED` are considered eligible unconditionally, but
backend `verifyCanReload()` treats those states as stopped only when the active
flow has no active threads. This can expose Change Version while the request
will be rejected with 409. Please require
`entity.status.aggregateSnapshot.activeThreadCount === 0` for these two states
and add active/quiescent test cases. `STOPPED` can retain its current behavior.
--
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]
