This is an automated email from the ASF dual-hosted git repository.
rfellows pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new f3a773b2136 NIFI-16489: Reconcile connector canvas after cluster
connection changes (#11842)
f3a773b2136 is described below
commit f3a773b2136e885be0230153137190b577481f1f
Author: Matt Gilman <[email protected]>
AuthorDate: Tue Oct 27 16:20:44 2026 -0400
NIFI-16489: Reconcile connector canvas after cluster connection changes
(#11842)
Preserve authoritative snapshot reconciliation across failed or cancelled
loads while retaining optimistic in-flight mutations.
This closes #11842
---
.../connector-canvas/connector-canvas.actions.ts | 9 +-
.../connector-canvas.effects.spec.ts | 120 +++++++++++-
.../connector-canvas/connector-canvas.effects.ts | 145 ++++++++------
.../connector-canvas.reducer.spec.ts | 209 +++++++++++++++++++--
.../connector-canvas/connector-canvas.reducer.ts | 95 +++++++---
.../connector-canvas/connector-canvas.selectors.ts | 10 +
.../connectors/state/connector-canvas/index.ts | 2 +
.../connector-canvas.component.spec.ts | 4 +-
8 files changed, 489 insertions(+), 105 deletions(-)
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.actions.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.actions.ts
index 20bb8331aac..9d99bc65eb5 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.actions.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.actions.ts
@@ -69,6 +69,8 @@ export const loadConnectorFlow = createAction(
props<{ connectorId: string; processGroupId: string }>()
);
+export const markAuthoritativeSnapshotPending = createAction('[Connector
Canvas] Mark Authoritative Snapshot Pending');
+
export const loadConnectorFlowSuccess = createAction(
'[Connector Canvas] Load Connector Flow Success',
props<{
@@ -84,12 +86,17 @@ export const loadConnectorFlowSuccess = createAction(
processGroups: ProcessGroupEntity[];
processors: ProcessorEntity[];
connections: ConnectionEntity[];
+ reconciliationSnapshot: {
+ addedCache: string[];
+ removedCache: string[];
+ };
+ connectedStateChanged: boolean;
}>()
);
export const loadConnectorFlowFailure = createAction(
'[Connector Canvas] Load Connector Flow Failure',
- props<{ errorContext: ErrorContext; isRefresh: boolean }>()
+ props<{ errorContext: ErrorContext; isRefresh: boolean;
connectedStateChanged: boolean }>()
);
export const loadConnectorFlowComplete = createAction('[Connector Canvas] Load
Connector Flow Complete');
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.spec.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.spec.ts
index 64561fb26b7..d3faa13eeaf 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.spec.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.spec.ts
@@ -66,6 +66,7 @@ import {
loadConnectorFlowComplete,
loadConnectorFlowFailure,
loadConnectorFlowSuccess,
+ markAuthoritativeSnapshotPending,
loadConnectorParameterContext,
loadConnectorParameterContextFailure,
loadConnectorParameterContextSuccess,
@@ -107,7 +108,9 @@ import {
selectHasLoadedFlow,
selectLoadingStatus,
selectMaxConnectionZIndex,
+ selectOptimisticCacheSnapshot,
selectParentProcessGroupId,
+ selectPendingAuthoritativeSnapshot,
selectProcessGroupId,
selectProcessGroupIdFromRoute,
selectSaving
@@ -117,6 +120,8 @@ import { DocumentVisibility } from
'../../../../state/document-visibility';
import type { Mock } from 'vitest';
import { selectPrioritizerTypes } from
'../../../../state/extension-types/extension-types.selectors';
import { EditConnectionComponent } from
'../../../../ui/common/component-dialogs/edit-connection/edit-connection.component';
+import { resetConnectedStateChanged } from
'../../../../state/cluster-summary/cluster-summary.actions';
+import { selectConnectedStateChanged } from
'../../../../state/cluster-summary/cluster-summary.selectors';
describe('ConnectorCanvasEffects', () => {
// Setup function following SIFERS pattern
@@ -129,6 +134,10 @@ describe('ConnectorCanvasEffects', () => {
documentVisibility?: DocumentVisibility;
loadingStatus?: 'pending' | 'loading' | 'success' |
'refresh-error' | 'error';
hasLoadedFlow?: boolean;
+ connectedStateChanged?: boolean;
+ pendingAuthoritativeSnapshot?: boolean;
+ addedCache?: string[];
+ removedCache?: string[];
parameterContext?: ParameterContextEntity | null;
parameterContextResponse?: ParameterContextEntity | null;
parameterContextError?: unknown;
@@ -217,6 +226,18 @@ describe('ConnectorCanvasEffects', () => {
{ selector: selectMaxConnectionZIndex, value: 4 },
{ selector: selectLoadingStatus, value: loadingStatus
},
{ selector: selectHasLoadedFlow, value:
options.hasLoadedFlow ?? false },
+ { selector: selectConnectedStateChanged, value:
options.connectedStateChanged ?? false },
+ {
+ selector: selectPendingAuthoritativeSnapshot,
+ value: options.pendingAuthoritativeSnapshot ??
false
+ },
+ {
+ selector: selectOptimisticCacheSnapshot,
+ value: {
+ addedCache: options.addedCache ?? [],
+ removedCache: options.removedCache ?? []
+ }
+ },
{
selector: selectDocumentVisibilityState,
value: { documentVisibility, changedTimestamp: 0 }
@@ -549,7 +570,10 @@ describe('ConnectorCanvasEffects', () => {
describe('loadConnectorFlow$', () => {
it('should call getConnectorFlow and dispatch
loadConnectorFlowSuccess', async () => {
- const { effects, actions$, mockConnectorService } = await setup();
+ const { effects, actions$, mockConnectorService, store } = await
setup({
+ connectedStateChanged: true
+ });
+ const dispatch = vi.spyOn(store, 'dispatch');
const label = makeLabel('l1', { position: { x: 10, y: 20 } });
const flowResponse = {
processGroupFlow: {
@@ -593,9 +617,14 @@ describe('ConnectorCanvasEffects', () => {
remoteProcessGroups: [],
processGroups: [],
processors: [],
- connections: []
+ connections: [],
+ reconciliationSnapshot: { addedCache: [], removedCache: []
},
+ connectedStateChanged: true
})
);
+
expect(dispatch).toHaveBeenCalledWith(markAuthoritativeSnapshotPending());
+
expect(dispatch).toHaveBeenCalledWith(resetConnectedStateChanged());
+ expect(dispatch).toHaveBeenCalledTimes(2);
});
it('should default flow collections when processGroupFlow.flow is
missing', async () => {
@@ -631,7 +660,73 @@ describe('ConnectorCanvasEffects', () => {
remoteProcessGroups: [],
processGroups: [],
processors: [],
- connections: []
+ connections: [],
+ reconciliationSnapshot: { addedCache: [], removedCache: []
},
+ connectedStateChanged: false
+ })
+ );
+ });
+
+ it('should preserve a pending authoritative snapshot after the cluster
signal has been reset', async () => {
+ const { effects, actions$, mockConnectorService } = await setup({
+ pendingAuthoritativeSnapshot: true
+ });
+ (mockConnectorService.getConnectorFlow as Mock).mockReturnValue(
+ of({
+ processGroupFlow: {
+ id: 'pg-123',
+ parentGroupId: null,
+ breadcrumb: null
+ }
+ })
+ );
+ actions$(of(loadConnectorFlow({ connectorId: 'conn-a',
processGroupId: 'pg-in' })));
+
+ const action = await firstValueFrom(effects.loadConnectorFlow$);
+
+ expect(action).toEqual(expect.objectContaining({
connectedStateChanged: true }));
+ });
+
+ it('should mark the authoritative snapshot pending before the request
completes', async () => {
+ const { effects, actions$, mockConnectorService, store } = await
setup({
+ connectedStateChanged: true
+ });
+ const dispatch = vi.spyOn(store, 'dispatch');
+ const response$ = new Subject<ProcessGroupFlowEntity>();
+ (mockConnectorService.getConnectorFlow as
Mock).mockReturnValue(response$);
+ actions$(of(loadConnectorFlow({ connectorId: 'conn-a',
processGroupId: 'pg-in' })));
+
+ const subscription = effects.loadConnectorFlow$.subscribe();
+
+
expect(dispatch).toHaveBeenCalledWith(markAuthoritativeSnapshotPending());
+
expect(dispatch).toHaveBeenCalledWith(resetConnectedStateChanged());
+ subscription.unsubscribe();
+ });
+
+ it('should capture optimistic caches when the request begins', async
() => {
+ const { effects, actions$, mockConnectorService } = await setup({
+ addedCache: ['proc-created'],
+ removedCache: ['proc-deleted']
+ });
+ (mockConnectorService.getConnectorFlow as Mock).mockReturnValue(
+ of({
+ processGroupFlow: {
+ id: 'pg-123',
+ parentGroupId: null,
+ breadcrumb: null
+ }
+ })
+ );
+ actions$(of(loadConnectorFlow({ connectorId: 'conn-a',
processGroupId: 'pg-in' })));
+
+ const action = await firstValueFrom(effects.loadConnectorFlow$);
+
+ expect(action).toEqual(
+ expect.objectContaining({
+ reconciliationSnapshot: {
+ addedCache: ['proc-created'],
+ removedCache: ['proc-deleted']
+ }
})
);
});
@@ -657,13 +752,17 @@ describe('ConnectorCanvasEffects', () => {
errors: ['Error message'],
context: ErrorContextKey.CONNECTORS
},
- isRefresh: false
+ isRefresh: false,
+ connectedStateChanged: false
})
);
});
it('should dispatch a recoverable failure and banner when a refresh
fails', async () => {
- const { effects, actions$, mockConnectorService } = await setup({
hasLoadedFlow: true });
+ const { effects, actions$, mockConnectorService } = await setup({
+ hasLoadedFlow: true,
+ connectedStateChanged: true
+ });
const errorResponse = new HttpErrorResponse({ error: 'Failed',
status: 503, statusText: 'Unavailable' });
(mockConnectorService.getConnectorFlow as
Mock).mockReturnValue(throwError(() => errorResponse));
actions$(
@@ -680,7 +779,8 @@ describe('ConnectorCanvasEffects', () => {
expect(emitted[0]).toEqual(
expect.objectContaining({
type: loadConnectorFlowFailure.type,
- isRefresh: true
+ isRefresh: true,
+ connectedStateChanged: true
})
);
expect(emitted[1]).toEqual(
@@ -709,7 +809,9 @@ describe('ConnectorCanvasEffects', () => {
remoteProcessGroups: [],
processGroups: [],
processors: [],
- connections: []
+ connections: [],
+ reconciliationSnapshot: { addedCache: [],
removedCache: [] },
+ connectedStateChanged: false
})
)
);
@@ -1667,7 +1769,9 @@ describe('ConnectorCanvasEffects', () => {
remoteProcessGroups: [],
processGroups: [],
processors: [],
- connections: []
+ connections: [],
+ reconciliationSnapshot: { addedCache: [], removedCache: [] },
+ connectedStateChanged: false
});
}
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.ts
index af00b696be2..1d94424a93a 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.ts
@@ -91,8 +91,10 @@ import {
selectInputPort,
selectLoadingStatus,
selectMaxConnectionZIndex,
+ selectOptimisticCacheSnapshot,
selectOutputPort,
selectParentProcessGroupId,
+ selectPendingAuthoritativeSnapshot,
selectProcessGroup,
selectProcessGroupId,
selectProcessGroupIdFromRoute,
@@ -106,6 +108,8 @@ import type { ConnectorCanvasComponentEntity } from
'./index';
import { ClusterConnectionService } from
'../../../../service/cluster-connection.service';
import { ConnectorEditService } from '../../service/connector-edit.service';
import { HttpErrorResponse } from '@angular/common/http';
+import { resetConnectedStateChanged } from
'../../../../state/cluster-summary/cluster-summary.actions';
+import { selectConnectedStateChanged } from
'../../../../state/cluster-summary/cluster-summary.selectors';
/**
* Shallow-copies `entity` with `permissions.canWrite` forced to false.
@@ -173,68 +177,91 @@ export class ConnectorCanvasEffects {
loadConnectorFlow$ = createEffect(() =>
this.actions$.pipe(
ofType(ConnectorCanvasActions.loadConnectorFlow),
- concatLatestFrom(() => this.store.select(selectHasLoadedFlow)),
- switchMap(([action, hasLoadedFlow]) =>
- this.connectorService.getConnectorFlow(action.connectorId,
action.processGroupId).pipe(
- map((flowResponse) => {
- const processGroupFlow = flowResponse.processGroupFlow;
- const flow = processGroupFlow?.flow;
- const {
- labels,
- funnels,
- inputPorts,
- outputPorts,
- remoteProcessGroups,
- processGroups,
- processors,
- connections
- } = this.sanitizeConnectorFlowPositions(flow);
+ concatLatestFrom(() => [
+ this.store.select(selectHasLoadedFlow),
+ this.store.select(selectConnectedStateChanged),
+ this.store.select(selectPendingAuthoritativeSnapshot),
+ this.store.select(selectOptimisticCacheSnapshot)
+ ]),
+ tap(([, , clusterConnectedStateChanged]) => {
+ if (clusterConnectedStateChanged) {
+
this.store.dispatch(ConnectorCanvasActions.markAuthoritativeSnapshotPending());
+ }
+ this.store.dispatch(resetConnectedStateChanged());
+ }),
+ switchMap(
+ ([
+ action,
+ hasLoadedFlow,
+ clusterConnectedStateChanged,
+ pendingAuthoritativeSnapshot,
+ reconciliationSnapshot
+ ]) => {
+ const connectedStateChanged = clusterConnectedStateChanged
|| pendingAuthoritativeSnapshot;
+ return
this.connectorService.getConnectorFlow(action.connectorId,
action.processGroupId).pipe(
+ map((flowResponse) => {
+ const processGroupFlow =
flowResponse.processGroupFlow;
+ const flow = processGroupFlow?.flow;
+ const {
+ labels,
+ funnels,
+ inputPorts,
+ outputPorts,
+ remoteProcessGroups,
+ processGroups,
+ processors,
+ connections
+ } = this.sanitizeConnectorFlowPositions(flow);
- // Extract process group IDs for navigation
- const processGroupId = processGroupFlow?.id ?? null;
- const parentProcessGroupId =
processGroupFlow?.parentGroupId ?? null;
+ // Extract process group IDs for navigation
+ const processGroupId = processGroupFlow?.id ??
null;
+ const parentProcessGroupId =
processGroupFlow?.parentGroupId ?? null;
- // Extract breadcrumb for navigation trail
- const breadcrumb = processGroupFlow?.breadcrumb ??
null;
+ // Extract breadcrumb for navigation trail
+ const breadcrumb = processGroupFlow?.breadcrumb ??
null;
- return
ConnectorCanvasActions.loadConnectorFlowSuccess({
- connectorId: action.connectorId,
- processGroupId,
- parentProcessGroupId,
- breadcrumb,
- labels,
- funnels,
- inputPorts,
- outputPorts,
- remoteProcessGroups,
- processGroups,
- processors,
- connections
- });
- }),
- catchError((error) => {
- const message = this.errorHelper.getErrorString(error);
- const errorContext = {
- errors: [message],
- context: ErrorContextKey.CONNECTORS
- };
- const failure =
ConnectorCanvasActions.loadConnectorFlowFailure({
- errorContext,
- isRefresh: hasLoadedFlow
- });
- return hasLoadedFlow
- ? from([
- failure,
- ErrorActions.addBannerError({
- errorContext: {
- errors: [message],
- context:
ErrorContextKey.CONNECTOR_CANVAS
- }
- })
- ])
- : of(failure);
- })
- )
+ return
ConnectorCanvasActions.loadConnectorFlowSuccess({
+ connectorId: action.connectorId,
+ processGroupId,
+ parentProcessGroupId,
+ breadcrumb,
+ labels,
+ funnels,
+ inputPorts,
+ outputPorts,
+ remoteProcessGroups,
+ processGroups,
+ processors,
+ connections,
+ reconciliationSnapshot,
+ connectedStateChanged
+ });
+ }),
+ catchError((error) => {
+ const message =
this.errorHelper.getErrorString(error);
+ const errorContext = {
+ errors: [message],
+ context: ErrorContextKey.CONNECTORS
+ };
+ const failure =
ConnectorCanvasActions.loadConnectorFlowFailure({
+ errorContext,
+ isRefresh: hasLoadedFlow,
+ connectedStateChanged
+ });
+ return hasLoadedFlow
+ ? from([
+ failure,
+ ErrorActions.addBannerError({
+ errorContext: {
+ errors: [message],
+ context:
ErrorContextKey.CONNECTOR_CANVAS
+ }
+ })
+ ])
+ : of(failure);
+ })
+ );
+ }
)
)
);
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.reducer.spec.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.reducer.spec.ts
index fdb59d7771d..6cacb00179e 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.reducer.spec.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.reducer.spec.ts
@@ -67,7 +67,12 @@ describe('connectorCanvasReducer', () => {
describe('loadConnectorFlow', () => {
it('should clear component data when navigating to a different process
group', () => {
- const previousState = createPopulatedState({ processGroupId:
'pg-current' });
+ const previousState = createPopulatedState({
+ processGroupId: 'pg-current',
+ addedCache: ['proc-added'],
+ removedCache: ['proc-removed'],
+ pendingAuthoritativeSnapshot: true
+ });
const result = connectorCanvasReducer(
previousState,
@@ -86,10 +91,16 @@ describe('connectorCanvasReducer', () => {
expect(result.processGroups).toEqual([]);
expect(result.processors).toEqual([]);
expect(result.connections).toEqual([]);
+ expect(result.addedCache).toEqual([]);
+ expect(result.removedCache).toEqual([]);
+ expect(result.pendingAuthoritativeSnapshot).toBe(false);
});
it('should preserve component data when refreshing the same process
group', () => {
- const previousState = createPopulatedState({ processGroupId:
'pg-current' });
+ const previousState = createPopulatedState({
+ processGroupId: 'pg-current',
+ pendingAuthoritativeSnapshot: true
+ });
const result = connectorCanvasReducer(
previousState,
@@ -107,12 +118,16 @@ describe('connectorCanvasReducer', () => {
expect(result.outputPorts).toEqual(previousState.outputPorts);
expect(result.remoteProcessGroups).toEqual(previousState.remoteProcessGroups);
expect(result.processGroups).toEqual(previousState.processGroups);
+ expect(result.pendingAuthoritativeSnapshot).toBe(true);
});
it('should clear component data when navigating to a different
connector with the same process group id', () => {
const previousState = createPopulatedState({
connectorId: 'connector-1',
- processGroupId: 'pg-current'
+ processGroupId: 'pg-current',
+ addedCache: ['proc-added'],
+ removedCache: ['proc-removed'],
+ pendingAuthoritativeSnapshot: true
});
const result = connectorCanvasReducer(
@@ -131,6 +146,9 @@ describe('connectorCanvasReducer', () => {
expect(result.processors).toEqual([]);
expect(result.connections).toEqual([]);
expect(result.parameterContext).toBeNull();
+ expect(result.addedCache).toEqual([]);
+ expect(result.removedCache).toEqual([]);
+ expect(result.pendingAuthoritativeSnapshot).toBe(false);
});
});
@@ -154,7 +172,9 @@ describe('connectorCanvasReducer', () => {
remoteProcessGroups: [],
processGroups: [],
processors,
- connections
+ connections,
+ reconciliationSnapshot: { addedCache: [], removedCache: []
},
+ connectedStateChanged: false
})
);
@@ -186,7 +206,9 @@ describe('connectorCanvasReducer', () => {
remoteProcessGroups: [],
processGroups: [],
processors: [stale],
- connections: []
+ connections: [],
+ reconciliationSnapshot: { addedCache: [], removedCache: []
},
+ connectedStateChanged: false
})
);
@@ -213,7 +235,9 @@ describe('connectorCanvasReducer', () => {
remoteProcessGroups: [],
processGroups: [],
processors: [],
- connections: []
+ connections: [],
+ reconciliationSnapshot: { addedCache: [connection.id],
removedCache: [] },
+ connectedStateChanged: false
});
const afterFirstStaleLoad = connectorCanvasReducer(state,
staleAction);
const afterSecondStaleLoad =
connectorCanvasReducer(afterFirstStaleLoad, staleAction);
@@ -250,7 +274,9 @@ describe('connectorCanvasReducer', () => {
remoteProcessGroups: [],
processGroups: [],
processors: [staleProcessor],
- connections: []
+ connections: [],
+ reconciliationSnapshot: { addedCache: [], removedCache:
['proc-removed'] },
+ connectedStateChanged: false
});
const afterFirstStaleLoad = connectorCanvasReducer(state,
staleAction);
@@ -294,13 +320,134 @@ describe('connectorCanvasReducer', () => {
remoteProcessGroups: [],
processGroups: [],
processors: [makeProcessor('proc-other')],
- connections: [connection]
+ connections: [connection],
+ reconciliationSnapshot: { addedCache: [], removedCache:
['proc-removed'] },
+ connectedStateChanged: false
})
);
expect(result.connections).toEqual([]);
expect(result.removedCache).toContain('proc-removed');
});
+
+ it('should accept a lower-revision entity from an authoritative
snapshot', () => {
+ const current = makeProcessor('proc-1', { revision: { version: 2 }
});
+ const authoritative = makeProcessor('proc-1', { revision: {
version: 1 } });
+
+ const result = connectorCanvasReducer(
+ createPopulatedState({ processors: [current] }),
+ ConnectorCanvasActions.loadConnectorFlowSuccess({
+ connectorId: 'connector-1',
+ processGroupId: 'pg-current',
+ parentProcessGroupId: 'pg-parent',
+ breadcrumb: null,
+ labels: [],
+ funnels: [],
+ inputPorts: [],
+ outputPorts: [],
+ remoteProcessGroups: [],
+ processGroups: [],
+ processors: [authoritative],
+ connections: [],
+ reconciliationSnapshot: { addedCache: [], removedCache: []
},
+ connectedStateChanged: true
+ })
+ );
+
+ expect(result.processors).toEqual([authoritative]);
+ });
+
+ it('should ignore both caches and use the authoritative component and
connection sets', () => {
+ const source = makeProcessor('proc-restored');
+ const destination = makeProcessor('proc-destination');
+ const connection = makeConnection('conn-restored', {
+ sourceId: source.id,
+ destinationId: destination.id
+ });
+
+ const result = connectorCanvasReducer(
+ createPopulatedState({
+ processors: [makeProcessor('proc-local')],
+ connections: [],
+ addedCache: ['proc-local'],
+ removedCache: [source.id],
+ pendingAuthoritativeSnapshot: true
+ }),
+ ConnectorCanvasActions.loadConnectorFlowSuccess({
+ connectorId: 'connector-1',
+ processGroupId: 'pg-current',
+ parentProcessGroupId: 'pg-parent',
+ breadcrumb: null,
+ labels: [],
+ funnels: [],
+ inputPorts: [],
+ outputPorts: [],
+ remoteProcessGroups: [],
+ processGroups: [],
+ processors: [source, destination],
+ connections: [connection],
+ reconciliationSnapshot: {
+ addedCache: ['proc-local'],
+ removedCache: [source.id]
+ },
+ connectedStateChanged: true
+ })
+ );
+
+ expect(result.processors).toEqual([source, destination]);
+ expect(result.connections).toEqual([connection]);
+ expect(result.addedCache).toEqual([]);
+ expect(result.removedCache).toEqual([]);
+ expect(result.pendingAuthoritativeSnapshot).toBe(false);
+ });
+
+ it('should preserve optimistic mutations completed while an
authoritative request is in flight', () => {
+ const local = makeProcessor('proc-local');
+ const removed = makeProcessor('proc-removed');
+ const destination = makeProcessor('proc-destination');
+ const connection = makeConnection('conn-removed', {
+ sourceId: removed.id,
+ destinationId: destination.id
+ });
+
+ const result = connectorCanvasReducer(
+ createPopulatedState({
+ processors: [local, destination],
+ addedCache: [local.id],
+ removedCache: [removed.id]
+ }),
+ ConnectorCanvasActions.loadConnectorFlowSuccess({
+ connectorId: 'connector-1',
+ processGroupId: 'pg-current',
+ parentProcessGroupId: 'pg-parent',
+ breadcrumb: null,
+ labels: [],
+ funnels: [],
+ inputPorts: [],
+ outputPorts: [],
+ remoteProcessGroups: [],
+ processGroups: [],
+ processors: [removed, destination],
+ connections: [connection],
+ reconciliationSnapshot: { addedCache: [], removedCache: []
},
+ connectedStateChanged: true
+ })
+ );
+
+ expect(result.processors).toEqual([destination, local]);
+ expect(result.connections).toEqual([]);
+ expect(result.addedCache).toEqual([local.id]);
+ expect(result.removedCache).toEqual([removed.id]);
+ });
+ });
+
+ it('should retain the authoritative requirement as soon as the cluster
signal is consumed', () => {
+ const result = connectorCanvasReducer(
+ createPopulatedState({ pendingAuthoritativeSnapshot: false }),
+ ConnectorCanvasActions.markAuthoritativeSnapshotPending()
+ );
+
+ expect(result.pendingAuthoritativeSnapshot).toBe(true);
});
describe('loadConnectorFlowFailure', () => {
@@ -314,7 +461,8 @@ describe('connectorCanvasReducer', () => {
context: ErrorContextKey.CONNECTORS,
errors: ['Connection refused']
},
- isRefresh: false
+ isRefresh: false,
+ connectedStateChanged: false
})
);
@@ -336,7 +484,8 @@ describe('connectorCanvasReducer', () => {
context: ErrorContextKey.CONNECTORS,
errors: ['Service unavailable']
},
- isRefresh: true
+ isRefresh: true,
+ connectedStateChanged: false
})
);
@@ -344,6 +493,38 @@ describe('connectorCanvasReducer', () => {
expect(result.hasLoadedFlow).toBe(true);
expect(result.processors).toEqual(previousState.processors);
});
+
+ it('should retain an authoritative snapshot requirement after a
refresh fails', () => {
+ const result = connectorCanvasReducer(
+ createPopulatedState({ loadingStatus: 'loading' }),
+ ConnectorCanvasActions.loadConnectorFlowFailure({
+ errorContext: {
+ context: ErrorContextKey.CONNECTORS,
+ errors: ['Service unavailable']
+ },
+ isRefresh: true,
+ connectedStateChanged: true
+ })
+ );
+
+ expect(result.pendingAuthoritativeSnapshot).toBe(true);
+ });
+
+ it('should preserve an existing authoritative requirement after a
later ordinary failure', () => {
+ const result = connectorCanvasReducer(
+ createPopulatedState({ loadingStatus: 'loading',
pendingAuthoritativeSnapshot: true }),
+ ConnectorCanvasActions.loadConnectorFlowFailure({
+ errorContext: {
+ context: ErrorContextKey.CONNECTORS,
+ errors: ['Service unavailable']
+ },
+ isRefresh: true,
+ connectedStateChanged: false
+ })
+ );
+
+ expect(result.pendingAuthoritativeSnapshot).toBe(true);
+ });
});
describe('createConnection', () => {
@@ -618,7 +799,9 @@ describe('connectorCanvasReducer', () => {
remoteProcessGroups: [],
processGroups: [],
processors: [makeProcessor('proc-1')],
- connections: []
+ connections: [],
+ reconciliationSnapshot: { addedCache: [], removedCache:
['proc-1'] },
+ connectedStateChanged: false
});
state = connectorCanvasReducer(state, staleLoad);
@@ -679,7 +862,9 @@ describe('connectorCanvasReducer', () => {
remoteProcessGroups: [],
processGroups: [],
processors: [makeProcessor('proc-1')],
- connections: []
+ connections: [],
+ reconciliationSnapshot: { addedCache: [], removedCache: []
},
+ connectedStateChanged: false
})
);
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.reducer.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.reducer.ts
index 59516c33ba1..418db460622 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.reducer.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.reducer.ts
@@ -48,7 +48,8 @@ function processComponentCollection<E extends { id: string;
revision: Revision }
proposedComponents: E[],
currentComponents: E[],
addedCache: string[],
- removedCache: string[]
+ removedCache: string[],
+ connectedStateChanged: boolean
): E[] {
const currentById = new Map(currentComponents.map((component) =>
[component.id, component]));
const proposedIds = new Set(proposedComponents.map((component) =>
component.id));
@@ -56,7 +57,9 @@ function processComponentCollection<E extends { id: string;
revision: Revision }
.filter((component) => !removedCache.includes(component.id))
.map((proposedComponent) => {
const currentComponent = currentById.get(proposedComponent.id);
- return currentComponent &&
!hasCurrentOrNewerRevision(currentComponent.revision,
proposedComponent.revision)
+ return currentComponent &&
+ !connectedStateChanged &&
+ !hasCurrentOrNewerRevision(currentComponent.revision,
proposedComponent.revision)
? currentComponent
: proposedComponent;
});
@@ -91,6 +94,7 @@ export const connectorCanvasReducer = createReducer(
connections: [],
addedCache: [],
removedCache: [],
+ pendingAuthoritativeSnapshot: false,
hasLoadedFlow: false,
parameterContext: null
}
@@ -117,9 +121,19 @@ export const connectorCanvasReducer = createReducer(
remoteProcessGroups,
processGroups,
processors,
- connections
+ connections,
+ reconciliationSnapshot,
+ connectedStateChanged
}
) => {
+ const requestedAddedIds = new
Set(reconciliationSnapshot.addedCache);
+ const requestedRemovedIds = new
Set(reconciliationSnapshot.removedCache);
+ const addedCache = connectedStateChanged
+ ? state.addedCache.filter((id) => !requestedAddedIds.has(id))
+ : state.addedCache;
+ const removedCache = connectedStateChanged
+ ? state.removedCache.filter((id) =>
!requestedRemovedIds.has(id))
+ : state.removedCache;
const snapshotIds = new Set(
[
...labels,
@@ -138,8 +152,7 @@ export const connectorCanvasReducer = createReducer(
]);
const reconciledConnections = connections.filter(
(connection) =>
- !state.removedCache.includes(connection.sourceId) &&
- !state.removedCache.includes(connection.destinationId)
+ !removedCache.includes(connection.sourceId) &&
!removedCache.includes(connection.destinationId)
);
return {
...state,
@@ -147,46 +160,71 @@ export const connectorCanvasReducer = createReducer(
processGroupId,
parentProcessGroupId,
breadcrumb,
- labels: processComponentCollection(labels, state.labels,
state.addedCache, state.removedCache),
- funnels: processComponentCollection(funnels, state.funnels,
state.addedCache, state.removedCache),
+ labels: processComponentCollection(
+ labels,
+ state.labels,
+ addedCache,
+ removedCache,
+ connectedStateChanged
+ ),
+ funnels: processComponentCollection(
+ funnels,
+ state.funnels,
+ addedCache,
+ removedCache,
+ connectedStateChanged
+ ),
inputPorts: processComponentCollection(
inputPorts,
state.inputPorts,
- state.addedCache,
- state.removedCache
+ addedCache,
+ removedCache,
+ connectedStateChanged
),
outputPorts: processComponentCollection(
outputPorts,
state.outputPorts,
- state.addedCache,
- state.removedCache
+ addedCache,
+ removedCache,
+ connectedStateChanged
),
remoteProcessGroups: processComponentCollection(
remoteProcessGroups,
state.remoteProcessGroups,
- state.addedCache,
- state.removedCache
+ addedCache,
+ removedCache,
+ connectedStateChanged
),
processGroups: processComponentCollection(
processGroups,
state.processGroups,
- state.addedCache,
- state.removedCache
+ addedCache,
+ removedCache,
+ connectedStateChanged
),
processors: processComponentCollection(
processors,
state.processors,
- state.addedCache,
- state.removedCache
+ addedCache,
+ removedCache,
+ connectedStateChanged
),
connections: processComponentCollection(
reconciledConnections,
state.connections,
- state.addedCache,
- state.removedCache
+ addedCache,
+ removedCache,
+ connectedStateChanged
),
- addedCache: state.addedCache.filter((id) =>
!snapshotIds.has(id)),
- removedCache: state.removedCache.filter((id) =>
snapshotIdsOrConnectionEndpoints.has(id)),
+ addedCache: connectedStateChanged
+ ? addedCache
+ : state.addedCache.filter((id) =>
!requestedAddedIds.has(id) || !snapshotIds.has(id)),
+ removedCache: connectedStateChanged
+ ? removedCache
+ : state.removedCache.filter(
+ (id) => !requestedRemovedIds.has(id) ||
snapshotIdsOrConnectionEndpoints.has(id)
+ ),
+ pendingAuthoritativeSnapshot: false,
hasLoadedFlow: true,
loadingStatus: 'success' as const,
error: null
@@ -194,12 +232,21 @@ export const connectorCanvasReducer = createReducer(
}
),
- on(ConnectorCanvasActions.loadConnectorFlowFailure, (state, {
errorContext, isRefresh }) => ({
+ on(ConnectorCanvasActions.markAuthoritativeSnapshotPending, (state) => ({
...state,
- loadingStatus: isRefresh ? ('refresh-error' as const) : ('error' as
const),
- error: errorContext.errors.join(', ')
+ pendingAuthoritativeSnapshot: true
})),
+ on(
+ ConnectorCanvasActions.loadConnectorFlowFailure,
+ (state, { errorContext, isRefresh, connectedStateChanged }) => ({
+ ...state,
+ loadingStatus: isRefresh ? ('refresh-error' as const) : ('error'
as const),
+ error: errorContext.errors.join(', '),
+ pendingAuthoritativeSnapshot: state.pendingAuthoritativeSnapshot
|| connectedStateChanged
+ })
+ ),
+
on(ConnectorCanvasActions.navigateWithoutTransform, (state) => ({
...state,
skipTransform: true
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.selectors.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.selectors.ts
index fcd9506a1b1..cd77ab1f07b 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.selectors.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.selectors.ts
@@ -78,6 +78,16 @@ export const selectLoadingStatus =
createSelector(selectConnectorCanvasState, (s
export const selectHasLoadedFlow = createSelector(selectConnectorCanvasState,
(state) => state.hasLoadedFlow);
+export const selectOptimisticCacheSnapshot =
createSelector(selectConnectorCanvasState, (state) => ({
+ addedCache: state.addedCache,
+ removedCache: state.removedCache
+}));
+
+export const selectPendingAuthoritativeSnapshot = createSelector(
+ selectConnectorCanvasState,
+ (state) => state.pendingAuthoritativeSnapshot
+);
+
export const selectError = createSelector(selectConnectorCanvasState, (state)
=> state.error);
export const selectSaving = createSelector(selectConnectorCanvasState, (state)
=> state.saving);
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/index.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/index.ts
index cbc023bd183..3040e176748 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/index.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/index.ts
@@ -55,6 +55,7 @@ export interface ConnectorCanvasState {
connections: ConnectionEntity[];
addedCache: string[];
removedCache: string[];
+ pendingAuthoritativeSnapshot: boolean;
registryClients: RegistryClientEntity[];
skipTransform: boolean;
hasLoadedFlow: boolean;
@@ -79,6 +80,7 @@ export const initialConnectorCanvasState:
ConnectorCanvasState = {
connections: [],
addedCache: [],
removedCache: [],
+ pendingAuthoritativeSnapshot: false,
registryClients: [],
skipTransform: false,
hasLoadedFlow: false,
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/ui/connector-canvas/connector-canvas.component.spec.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/ui/connector-canvas/connector-canvas.component.spec.ts
index 6e2ac199874..1463d219cdd 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/ui/connector-canvas/connector-canvas.component.spec.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/ui/connector-canvas/connector-canvas.component.spec.ts
@@ -2460,7 +2460,9 @@ describe('ConnectorCanvasComponent', () => {
remoteProcessGroups: [],
processGroups: [],
processors: [],
- connections: []
+ connections: [],
+ reconciliationSnapshot: { addedCache: [], removedCache: []
},
+ connectedStateChanged: false
})
);
