This is an automated email from the ASF dual-hosted git repository.
mcgilman 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 0156816b87d NIFI-16543: Wait for every recursive Controller Service
reference (#11887)
0156816b87d is described below
commit 0156816b87d09da101c40a20d5309a3bda983cbf
Author: Rob Fellows <[email protected]>
AuthorDate: Wed Nov 4 16:29:09 2026 -0500
NIFI-16543: Wait for every recursive Controller Service reference (#11887)
Enable and disable could advance after one sibling Controller Service
branch completed. Require every sibling and nested branch to reach the accepted
state, and stop recursion at referenceCycle nodes whose children the server
omits. Add a regression spec covering mixed sibling and nested states, active
threads, empty references, cycles, and unauthorized descendants.
---
.../controller-service-state.effects.spec.ts | 416 +++++++++++++++++++++
.../controller-service-state.effects.ts | 120 +++---
2 files changed, 482 insertions(+), 54 deletions(-)
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/state/contoller-service-state/controller-service-state.effects.spec.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/state/contoller-service-state/controller-service-state.effects.spec.ts
new file mode 100644
index 00000000000..d478f674255
--- /dev/null
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/state/contoller-service-state/controller-service-state.effects.spec.ts
@@ -0,0 +1,416 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { TestBed } from '@angular/core/testing';
+import { ReplaySubject } from 'rxjs';
+import { MatDialog } from '@angular/material/dialog';
+import { provideMockActions } from '@ngrx/effects/testing';
+import { provideMockStore } from '@ngrx/store/testing';
+import { Action } from '@ngrx/store';
+import { ControllerServiceReferencingComponentEntity, Permissions } from
'@nifi/shared';
+
+import { ControllerServiceStateEffects } from
'./controller-service-state.effects';
+import { ControllerServiceStateService } from
'../../service/controller-service-state.service';
+import { ErrorHelper } from '../../service/error-helper.service';
+import { ControllerServiceEntity } from '../shared';
+import { SetEnableRequest, SetEnableStep } from './index';
+
+describe('ControllerServiceStateEffects', () => {
+ const authorizedPermissions: Permissions = { canRead: true, canWrite: true
};
+ const unauthorizedPermissions: Permissions = { canRead: true, canWrite:
false };
+
+ function createService(
+ id: string,
+ state: string,
+ children: ControllerServiceReferencingComponentEntity[] = [],
+ options: {
+ referenceCycle?: boolean;
+ permissions?: Permissions;
+ omitReferencingComponents?: boolean;
+ } = {}
+ ): ControllerServiceReferencingComponentEntity {
+ const component:
ControllerServiceReferencingComponentEntity['component'] = {
+ groupId: 'group-1',
+ id,
+ name: id,
+ type: 'org.apache.nifi.TestService',
+ state,
+ properties: {},
+ descriptors: {},
+ validationErrors: [],
+ referenceType: 'ControllerService',
+ referenceCycle: options.referenceCycle ?? false,
+ ...(options.omitReferencingComponents ? {} : {
referencingComponents: children })
+ };
+
+ return {
+ id,
+ revision: { version: 1 },
+ permissions: options.permissions ?? authorizedPermissions,
+ bulletins: [],
+ operatePermissions: authorizedPermissions,
+ component
+ };
+ }
+
+ function createSchedulable(
+ id: string,
+ referenceType: 'Processor' | 'ReportingTask',
+ state: string,
+ activeThreadCount?: number
+ ): ControllerServiceReferencingComponentEntity {
+ return {
+ id,
+ revision: { version: 1 },
+ permissions: authorizedPermissions,
+ bulletins: [],
+ operatePermissions: authorizedPermissions,
+ component: {
+ groupId: 'group-1',
+ id,
+ name: id,
+ type: `org.apache.nifi.${referenceType}`,
+ state,
+ properties: {},
+ descriptors: {},
+ validationErrors: [],
+ referenceType,
+ referenceCycle: false,
+ referencingComponents: [],
+ ...(activeThreadCount === undefined ? {} : { activeThreadCount
})
+ }
+ };
+ }
+
+ function createControllerServiceEntity(
+ referencingComponents: ControllerServiceReferencingComponentEntity[],
+ runStatus = 'ENABLED'
+ ): ControllerServiceEntity {
+ return {
+ id: 'root-service',
+ uri: '/controller-services/root-service',
+ permissions: authorizedPermissions,
+ revision: { version: 1 },
+ bulletins: [],
+ status: { runStatus, validationStatus: 'VALID' },
+ component: {
+ id: 'root-service',
+ name: 'Root Service',
+ type: 'org.apache.nifi.RootService',
+ state: runStatus,
+ referencingComponents
+ }
+ };
+ }
+
+ function createSetEnableRequest(
+ currentStep: SetEnableStep,
+ enable: boolean,
+ scope: SetEnableRequest['scope'] = 'SERVICE_AND_REFERENCING_COMPONENTS'
+ ): SetEnableRequest {
+ return {
+ enable,
+ currentStep,
+ scope
+ };
+ }
+
+ async function setup() {
+ const actions$ = new ReplaySubject<Action>(1);
+
+ await TestBed.configureTestingModule({
+ providers: [
+ ControllerServiceStateEffects,
+ provideMockActions(() => actions$),
+ provideMockStore(),
+ {
+ provide: ControllerServiceStateService,
+ useValue: {
+ setEnable: vi.fn(),
+ getControllerService: vi.fn(),
+ updateReferencingServices: vi.fn(),
+ updateReferencingSchedulableComponents: vi.fn()
+ }
+ },
+ {
+ provide: ErrorHelper,
+ useValue: {
+ getErrorString: vi.fn()
+ }
+ },
+ {
+ provide: MatDialog,
+ useValue: {
+ open: vi.fn()
+ }
+ }
+ ]
+ }).compileComponents();
+
+ const effects = TestBed.inject(ControllerServiceStateEffects);
+
+ return { effects, actions$ };
+ }
+
+ beforeEach(() => {
+ vi.clearAllMocks();
+ });
+
+ describe('isReferencingServicesActionComplete / getNextStep enable', () =>
{
+ it('stays at EnableReferencingServices when one sibling remains
DISABLED', async () => {
+ const { effects } = await setup();
+ const controllerService = createControllerServiceEntity([
+ createService('s1', 'ENABLED'),
+ createService('s2', 'DISABLED')
+ ]);
+
+ const nextStep = effects.getNextStep(
+
createSetEnableRequest(SetEnableStep.EnableReferencingServices, true),
+ controllerService
+ );
+
+ expect(nextStep).toBe(SetEnableStep.EnableReferencingServices);
+ });
+
+ it('advances to StartReferencingComponents when every sibling is
ENABLING or ENABLED', async () => {
+ const { effects } = await setup();
+ const controllerService = createControllerServiceEntity([
+ createService('s1', 'ENABLING'),
+ createService('s2', 'ENABLED')
+ ]);
+
+ const nextStep = effects.getNextStep(
+
createSetEnableRequest(SetEnableStep.EnableReferencingServices, true),
+ controllerService
+ );
+
+ expect(nextStep).toBe(SetEnableStep.StartReferencingComponents);
+ });
+
+ it('stays at EnableReferencingServices when a nested service remains
DISABLED', async () => {
+ const { effects } = await setup();
+ const controllerService = createControllerServiceEntity([
+ createService('s1', 'ENABLED', [createService('s1a',
'DISABLED')]),
+ createService('s2', 'ENABLED')
+ ]);
+
+ const nextStep = effects.getNextStep(
+
createSetEnableRequest(SetEnableStep.EnableReferencingServices, true),
+ controllerService
+ );
+
+ expect(nextStep).toBe(SetEnableStep.EnableReferencingServices);
+ });
+ });
+
+ describe('isReferencingServicesActionComplete / getNextStep disable', ()
=> {
+ it('stays at DisableReferencingServices when one sibling remains
DISABLING', async () => {
+ const { effects } = await setup();
+ const controllerService = createControllerServiceEntity([
+ createService('s1', 'DISABLED'),
+ createService('s2', 'DISABLING')
+ ]);
+
+ const nextStep = effects.getNextStep(
+
createSetEnableRequest(SetEnableStep.DisableReferencingServices, false),
+ controllerService
+ );
+
+ expect(nextStep).toBe(SetEnableStep.DisableReferencingServices);
+ });
+
+ it('advances to DisableService when every sibling and nested service
is DISABLED', async () => {
+ const { effects } = await setup();
+ const controllerService = createControllerServiceEntity([
+ createService('s1', 'DISABLED', [createService('s1a',
'DISABLED')]),
+ createService('s2', 'DISABLED')
+ ]);
+
+ const nextStep = effects.getNextStep(
+
createSetEnableRequest(SetEnableStep.DisableReferencingServices, false),
+ controllerService
+ );
+
+ expect(nextStep).toBe(SetEnableStep.DisableService);
+ });
+ });
+
+ describe('areReferencingComponentsStopped / getNextStep stop', () => {
+ it('stays at StopReferencingComponents when one nested branch still
has a RUNNING processor', async () => {
+ const { effects } = await setup();
+ const controllerService = createControllerServiceEntity([
+ createService('s1', 'ENABLED', [createSchedulable('p1',
'Processor', 'STOPPED')]),
+ createService('s2', 'ENABLED', [createSchedulable('p2',
'Processor', 'RUNNING')])
+ ]);
+
+ const nextStep = effects.getNextStep(
+
createSetEnableRequest(SetEnableStep.StopReferencingComponents, false),
+ controllerService
+ );
+
+ expect(nextStep).toBe(SetEnableStep.StopReferencingComponents);
+ });
+
+ it('stays at StopReferencingComponents when a nested reporting task
has active threads', async () => {
+ const { effects } = await setup();
+ const controllerService = createControllerServiceEntity([
+ createService('s1', 'ENABLED', [createSchedulable('rt1',
'ReportingTask', 'STOPPED', 2)])
+ ]);
+
+ const nextStep = effects.getNextStep(
+
createSetEnableRequest(SetEnableStep.StopReferencingComponents, false),
+ controllerService
+ );
+
+ expect(nextStep).toBe(SetEnableStep.StopReferencingComponents);
+ });
+
+ it('advances to DisableReferencingServices when every nested branch is
stopped with zero threads', async () => {
+ const { effects } = await setup();
+ const controllerService = createControllerServiceEntity([
+ createService('s1', 'ENABLED', [createSchedulable('p1',
'Processor', 'STOPPED', 0)]),
+ createService('s2', 'ENABLED', [createSchedulable('rt1',
'ReportingTask', 'STOPPED')])
+ ]);
+
+ const nextStep = effects.getNextStep(
+
createSetEnableRequest(SetEnableStep.StopReferencingComponents, false),
+ controllerService
+ );
+
+ expect(nextStep).toBe(SetEnableStep.DisableReferencingServices);
+ });
+
+ it('stays at StopReferencingComponents when a direct processor is
RUNNING', async () => {
+ const { effects } = await setup();
+ const controllerService = createControllerServiceEntity([
+ createSchedulable('p1', 'Processor', 'RUNNING'),
+ createService('s1', 'ENABLED', [createSchedulable('p2',
'Processor', 'STOPPED')])
+ ]);
+
+ const nextStep = effects.getNextStep(
+
createSetEnableRequest(SetEnableStep.StopReferencingComponents, false),
+ controllerService
+ );
+
+ expect(nextStep).toBe(SetEnableStep.StopReferencingComponents);
+ });
+ });
+
+ describe('empty references and reference cycles', () => {
+ it('treats empty referencing components as complete for enable and
stop checks', async () => {
+ const { effects } = await setup();
+
+ expect(effects.isReferencingServicesActionComplete([], ['ENABLED',
'ENABLING'])).toBe(true);
+ expect(effects.isReferencingServicesActionComplete(undefined,
['DISABLED'])).toBe(true);
+ expect(effects.areReferencingComponentsStopped([])).toBe(true);
+
expect(effects.areReferencingComponentsStopped(undefined)).toBe(true);
+ expect(effects.hasUnauthorizedReferences([])).toBe(false);
+ expect(effects.hasUnauthorizedReferences(undefined)).toBe(false);
+ });
+
+ it('treats a leaf service with an empty referencingComponents list as
complete', async () => {
+ const { effects } = await setup();
+ const leaf = [createService('s1', 'ENABLED', [])];
+
+ expect(effects.isReferencingServicesActionComplete(leaf,
['ENABLED', 'ENABLING'])).toBe(true);
+ expect(effects.areReferencingComponentsStopped(leaf)).toBe(true);
+ });
+
+ it('does not throw on cycle nodes and treats an accepted cycle sibling
as complete', async () => {
+ const { effects } = await setup();
+ const referencingComponents = [
+ createService('s1', 'ENABLED'),
+ createService('s-cycle', 'ENABLED', [], {
+ referenceCycle: true,
+ omitReferencingComponents: true
+ })
+ ];
+
+ expect(() =>
+
effects.isReferencingServicesActionComplete(referencingComponents, ['ENABLED',
'ENABLING'])
+ ).not.toThrow();
+
expect(effects.isReferencingServicesActionComplete(referencingComponents,
['ENABLED', 'ENABLING'])).toBe(
+ true
+ );
+
+ expect(() =>
effects.areReferencingComponentsStopped(referencingComponents)).not.toThrow();
+
expect(effects.areReferencingComponentsStopped(referencingComponents)).toBe(true);
+
+ expect(() =>
effects.hasUnauthorizedReferences(referencingComponents)).not.toThrow();
+
expect(effects.hasUnauthorizedReferences(referencingComponents)).toBe(false);
+ });
+
+ it('returns false when a cycle node is outside the accepted state',
async () => {
+ const { effects } = await setup();
+ const referencingComponents = [
+ createService('s1', 'ENABLED'),
+ createService('s-cycle', 'DISABLED', [], {
+ referenceCycle: true,
+ omitReferencingComponents: true
+ })
+ ];
+
+
expect(effects.isReferencingServicesActionComplete(referencingComponents,
['ENABLED', 'ENABLING'])).toBe(
+ false
+ );
+ });
+
+ it('keeps waiting when a service is expanded under one sibling and
flagged under another', async () => {
+ const { effects } = await setup();
+ const expanded = createService('shared', 'ENABLED',
[createService('child', 'DISABLED')]);
+ const flagged = createService('shared', 'ENABLED', [], {
+ referenceCycle: true,
+ omitReferencingComponents: true
+ });
+
+ const nextStep = effects.getNextStep(
+
createSetEnableRequest(SetEnableStep.EnableReferencingServices, true),
+ createControllerServiceEntity([
+ createService('s1', 'ENABLED', [expanded]),
+ createService('s2', 'ENABLED', [flagged])
+ ])
+ );
+
+ expect(nextStep).toBe(SetEnableStep.EnableReferencingServices);
+ });
+
+ it('detects unauthorized permissions without recursing into cycle
nodes', async () => {
+ const { effects } = await setup();
+ const referencingComponents = [
+ createService('s1', 'ENABLED', [], { permissions:
unauthorizedPermissions }),
+ createService('s-cycle', 'ENABLED', [], {
+ referenceCycle: true,
+ omitReferencingComponents: true
+ })
+ ];
+
+ expect(() =>
effects.hasUnauthorizedReferences(referencingComponents)).not.toThrow();
+
expect(effects.hasUnauthorizedReferences(referencingComponents)).toBe(true);
+ });
+
+ it('detects unauthorized permissions in a nested descendant', async ()
=> {
+ const { effects } = await setup();
+ const referencingComponents = [
+ createService('s1', 'ENABLED', [
+ createService('s1a', 'ENABLED', [], { permissions:
unauthorizedPermissions })
+ ])
+ ];
+
+
expect(effects.hasUnauthorizedReferences(referencingComponents)).toBe(true);
+ });
+ });
+});
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/state/contoller-service-state/controller-service-state.effects.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/state/contoller-service-state/controller-service-state.effects.ts
index 88a940a3a2f..c8391854cdd 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/state/contoller-service-state/controller-service-state.effects.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/state/contoller-service-state/controller-service-state.effects.ts
@@ -26,12 +26,27 @@ import { selectControllerService,
selectControllerServiceSetEnableRequest } from
import { OkDialog } from '../../ui/common/ok-dialog/ok-dialog.component';
import { MatDialog } from '@angular/material/dialog';
import { ControllerServiceStateService } from
'../../service/controller-service-state.service';
-import { ControllerServiceEntity, ControllerServiceReferencingComponentEntity
} from '../shared';
+import { ControllerServiceEntity } from '../shared';
import { SetEnableRequest, SetEnableStep } from './index';
-import { isDefinedAndNotNull, MEDIUM_DIALOG } from '@nifi/shared';
+import {
+ ControllerServiceReferencingComponentDTO,
+ ControllerServiceReferencingComponentEntity,
+ isDefinedAndNotNull,
+ MEDIUM_DIALOG
+} from '@nifi/shared';
import { HttpErrorResponse } from '@angular/common/http';
import { ErrorHelper } from '../../service/error-helper.service';
+type ReferencingControllerServiceEntity =
ControllerServiceReferencingComponentEntity & {
+ component: ControllerServiceReferencingComponentDTO;
+};
+
+function isReferencingControllerService(
+ referencingComponent: ControllerServiceReferencingComponentEntity
+): referencingComponent is ReferencingControllerServiceEntity {
+ return referencingComponent.component?.referenceType ===
'ControllerService';
+}
+
@Injectable()
export class ControllerServiceStateEffects {
private actions$ = inject(Actions);
@@ -332,14 +347,18 @@ export class ControllerServiceStateEffects {
{ dispatch: false }
);
- hasUnauthorizedReferences(referencingComponents:
ControllerServiceReferencingComponentEntity[]): boolean {
+ hasUnauthorizedReferences(
+ referencingComponents: ControllerServiceReferencingComponentEntity[] |
undefined
+ ): boolean {
+ const components = referencingComponents ?? [];
+
// if there are no referencing components there is nothing unauthorized
- if (referencingComponents.length === 0) {
+ if (components.length === 0) {
return false;
}
// identify any unauthorized referencing components
- const unauthorized: boolean =
referencingComponents.some((referencingComponentEntity) => {
+ const unauthorized: boolean =
components.some((referencingComponentEntity) => {
return !referencingComponentEntity.permissions.canRead ||
!referencingComponentEntity.permissions.canWrite;
});
@@ -349,19 +368,17 @@ export class ControllerServiceStateEffects {
}
// consider the components that are referencing the referencingServices
- const referencingServices =
referencingComponents.filter((referencingComponent) => {
- return referencingComponent.component.referenceType ===
'ControllerService';
- });
+ const referencingServices =
components.filter(isReferencingControllerService);
- if (referencingServices.length === 0) {
- // if there are no more nested services all references are
authorized
- return false;
- } else {
- // if there are nested services, check if they have any
referencing components are unauthorized
- return referencingServices.some((referencingService) => {
- return
this.hasUnauthorizedReferences(referencingService.component.referencingComponents);
- });
- }
+ // if there are nested services, check if they have any referencing
components that are unauthorized;
+ // a cycle is evaluated where the service was first encountered
+ return referencingServices.some((referencingService) => {
+ const referencingComponent = referencingService.component;
+ return (
+ !referencingComponent.referenceCycle &&
+
this.hasUnauthorizedReferences(referencingComponent.referencingComponents)
+ );
+ });
}
getNextStep(request: SetEnableRequest, controllerServiceEntity:
ControllerServiceEntity): SetEnableStep {
@@ -407,45 +424,43 @@ export class ControllerServiceStateEffects {
}
isReferencingServicesActionComplete(
- referencingComponents: ControllerServiceReferencingComponentEntity[],
+ referencingComponents: ControllerServiceReferencingComponentEntity[] |
undefined,
acceptedRunStatus: string[]
): boolean {
- const referencingServices =
referencingComponents.filter((referencingComponent) => {
- return referencingComponent.component.referenceType ===
'ControllerService';
- });
-
- if (referencingServices.length === 0) {
- return true;
- }
-
- return referencingServices.some((referencingService) => {
- const isEnabled: boolean =
acceptedRunStatus.includes(referencingService.component.state);
-
- if (isEnabled) {
- // if this service isn't enabled, there is no need to check
further...
- return this.isReferencingServicesActionComplete(
- referencingService.component.referencingComponents,
- acceptedRunStatus
- );
+ const referencingServices = (referencingComponents ??
[]).filter(isReferencingControllerService);
+
+ // every referencing service, and every service beneath it, must reach
an accepted state;
+ // a cycle is evaluated where the service was first encountered
+ return referencingServices.every((referencingService) => {
+ const referencingComponent = referencingService.component;
+ if (referencingComponent.state === undefined ||
!acceptedRunStatus.includes(referencingComponent.state)) {
+ return false;
}
- return isEnabled;
+ return (
+ referencingComponent.referenceCycle ||
+
this.isReferencingServicesActionComplete(referencingComponent.referencingComponents,
acceptedRunStatus)
+ );
});
}
- areReferencingComponentsStopped(referencingComponents:
ControllerServiceReferencingComponentEntity[]): boolean {
+ areReferencingComponentsStopped(
+ referencingComponents: ControllerServiceReferencingComponentEntity[] |
undefined
+ ): boolean {
+ const components = referencingComponents ?? [];
+
// consider the schedulable components in the referencingComponents
- const referencingScheduleableComponents =
referencingComponents.filter((referencingComponent) => {
+ const referencingScheduleableComponents =
components.filter((referencingComponent) => {
return (
- referencingComponent.component.referenceType === 'Processor' ||
- referencingComponent.component.referenceType ===
'ReportingTask'
+ referencingComponent.component?.referenceType === 'Processor'
||
+ referencingComponent.component?.referenceType ===
'ReportingTask'
);
});
const stillRunning: boolean =
referencingScheduleableComponents.some((referencingComponentEntity) => {
const referencingComponent = referencingComponentEntity.component;
return (
- referencingComponent.state === 'RUNNING' ||
- (referencingComponent.activeThreadCount &&
referencingComponent.activeThreadCount > 0)
+ referencingComponent?.state === 'RUNNING' ||
+ (referencingComponent?.activeThreadCount !== undefined &&
referencingComponent.activeThreadCount > 0)
);
});
@@ -455,18 +470,15 @@ export class ControllerServiceStateEffects {
}
// consider the scheduleable components that are referencing the
referencingServices
- const referencingServices =
referencingComponents.filter((referencingComponent) => {
- return referencingComponent.component.referenceType ===
'ControllerService';
- });
+ const referencingServices =
components.filter(isReferencingControllerService);
- if (referencingServices.length === 0) {
- // if there are no more nested services all schedulable components
have stopped
- return true;
- } else {
- // if there are nested services, check if they have any
referencing components that are still running
- return referencingServices.some((referencingService) => {
- return
this.areReferencingComponentsStopped(referencingService.component.referencingComponents);
- });
- }
+ // every nested branch must have stopped; a cycle is evaluated where
the service was first encountered
+ return referencingServices.every((referencingService) => {
+ const referencingComponent = referencingService.component;
+ return (
+ referencingComponent.referenceCycle ||
+
this.areReferencingComponentsStopped(referencingComponent.referencingComponents)
+ );
+ });
}
}
