awelless commented on code in PR #11813:
URL: https://github.com/apache/nifi/pull/11813#discussion_r4563950557
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/migration/StandardControllerServiceFactory.java:
##########
@@ -123,6 +145,22 @@ public ControllerServiceNode create(final
ControllerServiceCreationDetails creat
return serviceNode;
}
+ private @Nullable String generateVersionedId(final
ControllerServiceCreationDetails creationDetails, final ControllerServiceNode
serviceNode) {
+ if (owningGroupIdentity == null) {
+ return null;
+ }
+
+ final Map<String, String> sortedProperties = new TreeMap<>();
+ for (final Map.Entry<String, String> property :
creationDetails.serviceProperties().entrySet()) {
+ if
(!serviceNode.getPropertyDescriptor(property.getKey()).isSensitive()) {
Review Comment:
Moved to the synchronizer, which reads properties after the service's own
migration, so renamed secrets are recognized. Instead of skipping sensitive
values, a Parameter reference is hashed and an inline secret leaves the service
unversioned (see the top-level comment).
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java:
##########
@@ -418,12 +425,15 @@ public void synchronize(final ProcessGroup group, final
VersionedExternalFlow ve
}
});
+ final VersionedExternalFlowMetadata metadata =
versionedExternalFlow.getMetadata();
+ final String flowIdentifier = metadata == null ? null :
metadata.getFlowIdentifier();
for (final CreatedOrModifiedExtension createdOrModifiedExtension :
createdAndModifiedExtensions) {
final ComponentNode extension =
createdOrModifiedExtension.extension();
final Map<String, String> originalPropertyValues =
createdOrModifiedExtension.propertyValues();
+ final String owningGroupIdentity =
getOwningProcessGroupIdentity(extension, group, flowIdentifier);
final ControllerServiceFactory serviceFactory = new
StandardControllerServiceFactory(context.getExtensionManager(),
context.getFlowManager(),
- context.getControllerServiceProvider(), extension);
+ context.getControllerServiceProvider(), extension,
owningGroupIdentity);
if (extension instanceof final ProcessorNode processor) {
processor.migrateConfiguration(originalPropertyValues,
serviceFactory);
Review Comment:
There is now one shared method for synchronize and paste. It differs from
your outline in three ways:
- referenced IDs replace referrer IDs (see the top-level comment);
- instead of a counter, created services that would share an ID within a
group stay unversioned;
- missing services are removed before IDs are assigned, so a created service
can take the ID of the one it replaces.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/migration/StandardControllerServiceFactory.java:
##########
@@ -104,7 +113,20 @@ public ControllerServiceNode create(final
ControllerServiceCreationDetails creat
serviceNode.setProperties(creationDetails.serviceProperties());
- final ControllerServiceFactory serviceFactory = new
StandardControllerServiceFactory(extensionManager, flowManager,
serviceProvider, serviceNode);
+ final String versionedId = generateVersionedId(creationDetails,
serviceNode);
+ if (versionedId != null) {
+ final Set<ControllerServiceNode> groupControllerServices =
group.map(processGroup ->
processGroup.getControllerServices(false)).orElse(Collections.emptySet());
+ final boolean occupied = groupControllerServices.stream()
+ .map(ControllerServiceNode::getVersionedComponentId)
+ .flatMap(Optional::stream)
+ .anyMatch(versionedId::equals);
+
+ if (!occupied) {
+ serviceNode.setVersionedComponentId(versionedId);
+ }
+ }
Review Comment:
The factory now only records created services, the nested factory shares the
collection, and the ID code is gone. One difference: the service is recorded
right after creation rather than before its migration specifically, since the
order no longer matters once IDs are computed after all migrations.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java:
##########
@@ -308,12 +313,14 @@ public ComponentAdditions
addVersionedComponentsToProcessGroup(final ProcessGrou
// update ports to final names
updatePortsToFinalNames(proposedPortFinalNames);
+ final String groupIdentity = determineProcessGroupIdentity(group);
for (final CreatedOrModifiedExtension createdOrModifiedExtension :
createdAndModifiedExtensions) {
final ComponentNode extension =
createdOrModifiedExtension.extension();
final Map<String, String> originalPropertyValues =
createdOrModifiedExtension.propertyValues();
+ final String owningGroupIdentity =
getOwningProcessGroupIdentity(extension, group, groupIdentity);
final ControllerServiceFactory serviceFactory = new
StandardControllerServiceFactory(context.getExtensionManager(),
context.getFlowManager(),
- context.getControllerServiceProvider(), extension);
+ context.getControllerServiceProvider(), extension,
owningGroupIdentity);
if (extension instanceof final ProcessorNode processor) {
processor.migrateConfiguration(originalPropertyValues,
serviceFactory);
Review Comment:
Paste now uses the shared method. It also migrates reporting tasks, as
synchronize does.
##########
nifi-system-tests/nifi-alternate-config-extensions-bundle/nifi-alternate-config-extensions/src/main/java/org/apache/nifi/processors/tests/system/MigrateToControllerService.java:
##########
@@ -0,0 +1,90 @@
+/*
+ * 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.
+ */
+
+package org.apache.nifi.processors.tests.system;
+
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.cs.tests.system.StateBackedStoreService;
+import org.apache.nifi.cs.tests.system.StoreService;
+import org.apache.nifi.migration.PropertyConfiguration;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
+
+@CapabilityDescription("""
+ Post-upgrade shape of a processor whose property migration creates the
store Controller Service that the pre-upgrade shape did not have.
+ Each execution appends a row to the store so that tests can observe
whether store contents survive flow and runtime upgrades.
+ """)
+@InputRequirement(Requirement.INPUT_FORBIDDEN)
+public class MigrateToControllerService extends AbstractProcessor {
+
+ static final PropertyDescriptor STORE_SERVICE = new
PropertyDescriptor.Builder()
+ .name("Store Service")
+ .required(false)
+ .identifiesControllerService(StoreService.class)
+ .build();
+
+ static final Relationship REL_SUCCESS = new
Relationship.Builder().name("success").build();
+
+ private static final List<PropertyDescriptor> PROPERTIES =
List.of(STORE_SERVICE);
+ private static final Set<Relationship> RELATIONSHIPS = Set.of(REL_SUCCESS);
+
+ private final AtomicLong rowCounter = new AtomicLong(0L);
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ public void migrateProperties(final PropertyConfiguration config) {
+ final String storeName = config.getPropertyValue("Store
Name").orElse(null);
+ config.removeProperty("Store Name");
+
+ if (storeName != null) {
+ final String serviceId =
config.createControllerService(StateBackedStoreService.class.getName(),
Map.of("Store Name", storeName));
+ config.setProperty(STORE_SERVICE, serviceId);
+ }
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession
session) throws ProcessException {
+ final StoreService storeService =
context.getProperty(STORE_SERVICE).asControllerService(StoreService.class);
+ if (storeService == null) {
+ context.yield();
+ return;
+ }
+
+ storeService.append("row-" + rowCounter.getAndIncrement());
Review Comment:
Row counting removed; `onTrigger` is a no-op.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java:
##########
@@ -433,6 +443,8 @@ public void synchronize(final ProcessGroup group, final
VersionedExternalFlow ve
task.migrateConfiguration(originalPropertyValues,
serviceFactory);
}
}
+
+ removeMissingControllerServices(group,
versionedExternalFlow.getFlowContents());
Review Comment:
Implemented as you suggested, with your loop. One addition: the lookup by
versioned ID skips services recorded as missing, so a reference resolves to the
service that replaces them.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizerTest.java:
##########
@@ -552,6 +571,352 @@ public void testAddProcessorWithServiceAndMigration() {
assertEquals(controllerServiceNode.getIdentifier(),
migratedProperties.get("cs"));
}
+ @Nested
+ class MigrationCreatedControllerService {
+
+ private static final String SNAPSHOT_ROOT_ID = "snapshot-root-id";
+ private static final String LOCAL_CHILD_ID = "local-child";
+ private static final String CHILD_GROUP_ID = "child-group-id";
+ private static final String PARENT_GROUP_ID = "parent-group-id";
+ private static final String CHILD_PLACEMENT_ID = "child-placement-id";
+ private static final String DESTINATION_GROUP_ID =
"destination-group-id";
+
+ private static final String INCOMING_FLOW_ID = "incoming-flow-id";
+ private static final String DESTINATION_FLOW_ID =
"destination-flow-id";
+ private static final String PARENT_FLOW_ID = "parent-flow-id";
+ private static final String CHILD_FLOW_ID = "child-flow-id";
+
+ private static final String REGISTRY_ID = "registry-id";
+ private static final String SERVICE_TYPE =
TestControllerService.class.getName();
+ private static final Map<String, String> SERVICE_PROPERTIES =
Map.of("mode", "read");
+
+ @Test
+ void
migrationInSynchronizedGroupUsesIncomingFlowIdAfterVersionControlIsCleared() {
+ final ProcessGroup targetGroup = createMockProcessGroup();
+ final VersionControlInformation oldVersionControl =
mock(VersionControlInformation.class);
+
when(oldVersionControl.getFlowIdentifier()).thenReturn("old-flow-id");
+
+ final AtomicReference<VersionControlInformation>
liveVersionControl = new AtomicReference<>(oldVersionControl);
+
when(targetGroup.getVersionControlInformation()).thenAnswer(invocation ->
liveVersionControl.get());
+ doAnswer(invocation -> {
+ liveVersionControl.set(null);
+ return null;
+ }).when(targetGroup).disconnectVersionControl();
+
+ final VersionedProcessGroup proposed = new VersionedProcessGroup();
+ proposed.setIdentifier(SNAPSHOT_ROOT_ID);
+ proposed.setProcessors(Set.of(createMinimalVersionedProcessor()));
+
+ final VersionedExternalFlow incoming = new VersionedExternalFlow();
+ incoming.setFlowContents(proposed);
+
+ final VersionedExternalFlowMetadata metadata = new
VersionedExternalFlowMetadata();
+ metadata.setFlowIdentifier(INCOMING_FLOW_ID);
+ incoming.setMetadata(metadata);
+
+ mockMigratingProcessorIn(targetGroup);
+
+ synchronizer.synchronize(targetGroup, incoming,
synchronizationOptions);
+
+ assertNull(targetGroup.getVersionControlInformation());
+ final String expectedId =
expectedMigrationServiceId(INCOMING_FLOW_ID);
+ verify(controllerServiceNode).setVersionedComponentId(expectedId);
+ }
+
+ @Test
+ void migrationInChildWithoutOwnFlowUsesChildVersionedId() {
+ final ProcessGroup parentGroup = createMockProcessGroup();
+ final ProcessGroup childGroup = createMockProcessGroup();
+ when(childGroup.getIdentifier()).thenReturn(LOCAL_CHILD_ID);
+ when(childGroup.getParent()).thenReturn(parentGroup);
+
when(childGroup.getVersionedComponentId()).thenReturn(Optional.of(CHILD_GROUP_ID));
+
when(parentGroup.getProcessGroups()).thenReturn(Set.of(childGroup));
+
+ final VersionedProcessGroup proposedChild = new
VersionedProcessGroup();
+ proposedChild.setIdentifier(CHILD_GROUP_ID);
+
proposedChild.setProcessors(Set.of(createMinimalVersionedProcessor()));
+
+ final VersionedProcessGroup proposedParent = new
VersionedProcessGroup();
+ proposedParent.setIdentifier(PARENT_GROUP_ID);
+ proposedParent.setProcessGroups(Set.of(proposedChild));
+
+ final VersionedExternalFlow incoming = new VersionedExternalFlow();
+ incoming.setFlowContents(proposedParent);
+
+ final VersionedExternalFlowMetadata metadata = new
VersionedExternalFlowMetadata();
+ metadata.setFlowIdentifier(PARENT_FLOW_ID);
+ incoming.setMetadata(metadata);
+
+ mockMigratingProcessorIn(childGroup);
+
+ synchronizer.synchronize(parentGroup, incoming,
synchronizationOptions);
+
+ final String expectedId =
expectedMigrationServiceId(CHILD_GROUP_ID);
+ verify(controllerServiceNode).setVersionedComponentId(expectedId);
+ }
+
+ @Test
+ void migrationInIndependentlyVersionedChildUsesChildFlowId() {
+ final ProcessGroup parentGroup = createMockProcessGroup();
+ final ProcessGroup childGroup = createMockProcessGroup();
+ when(childGroup.getIdentifier()).thenReturn(LOCAL_CHILD_ID);
+ when(childGroup.getParent()).thenReturn(parentGroup);
+
when(childGroup.getVersionedComponentId()).thenReturn(Optional.of(CHILD_PLACEMENT_ID));
+
when(parentGroup.getProcessGroups()).thenReturn(Set.of(childGroup));
+
+ final AtomicReference<VersionControlInformation>
childVersionControl = new AtomicReference<>();
+
when(childGroup.getVersionControlInformation()).thenAnswer(invocation ->
childVersionControl.get());
+ doAnswer(invocation -> {
+ childVersionControl.set(invocation.getArgument(0));
+ return null;
+ }).when(childGroup).setVersionControlInformation(any(), anyMap());
+
+
when(flowManager.getFlowRegistryClient(REGISTRY_ID)).thenReturn(mock(FlowRegistryClientNode.class));
+ final VersionedFlowCoordinates coordinates = new
VersionedFlowCoordinates();
+ coordinates.setRegistryId(REGISTRY_ID);
+ coordinates.setBucketId("bucket");
+ coordinates.setFlowId(CHILD_FLOW_ID);
+ coordinates.setVersion("1");
+
+ final VersionedProcessGroup proposedChild = new
VersionedProcessGroup();
+ proposedChild.setIdentifier(CHILD_PLACEMENT_ID);
+ proposedChild.setVersionedFlowCoordinates(coordinates);
+
proposedChild.setProcessors(Set.of(createMinimalVersionedProcessor()));
+
+ final VersionedProcessGroup proposedParent = new
VersionedProcessGroup();
+ proposedParent.setIdentifier(PARENT_GROUP_ID);
+ proposedParent.setProcessGroups(Set.of(proposedChild));
+
+ final VersionedExternalFlow incoming = new VersionedExternalFlow();
+ incoming.setFlowContents(proposedParent);
+
+ final VersionedExternalFlowMetadata metadata = new
VersionedExternalFlowMetadata();
+ metadata.setFlowIdentifier(PARENT_FLOW_ID);
+ incoming.setMetadata(metadata);
+
+ mockMigratingProcessorIn(childGroup);
+
+ synchronizer.synchronize(parentGroup, incoming,
synchronizationOptions);
+
+ assertNotNull(childGroup.getVersionControlInformation());
+ assertEquals(CHILD_FLOW_ID,
childGroup.getVersionControlInformation().getFlowIdentifier());
+ final String expectedId =
expectedMigrationServiceId(CHILD_FLOW_ID);
+ verify(controllerServiceNode).setVersionedComponentId(expectedId);
+ }
+
+ @Test
+ void pasteIntoVersionControlledGroupUsesDestinationFlowId() {
+ final ProcessGroup destination = createMockProcessGroup();
+
when(destination.getVersionedComponentId()).thenReturn(Optional.of(DESTINATION_GROUP_ID));
+
+ final VersionControlInformation versionControl =
mock(VersionControlInformation.class);
+
when(versionControl.getFlowIdentifier()).thenReturn(DESTINATION_FLOW_ID);
+
when(destination.getVersionControlInformation()).thenReturn(versionControl);
+
+ mockMigratingProcessorIn(destination);
+
+ synchronizer.addVersionedComponentsToProcessGroup(destination,
legacyProcessorAdditions(), synchronizationOptions);
+
+ final String expectedId =
expectedMigrationServiceId(DESTINATION_FLOW_ID);
+ verify(controllerServiceNode).setVersionedComponentId(expectedId);
+ }
+
+ @Test
+ void pasteIntoGroupWithoutOwnFlowUsesDestinationVersionedId() {
+ final ProcessGroup destination = createMockProcessGroup();
+
when(destination.getVersionedComponentId()).thenReturn(Optional.of(DESTINATION_GROUP_ID));
+ mockMigratingProcessorIn(destination);
+
+ synchronizer.addVersionedComponentsToProcessGroup(destination,
legacyProcessorAdditions(), synchronizationOptions);
+
+ final String expectedId =
expectedMigrationServiceId(DESTINATION_GROUP_ID);
+ verify(controllerServiceNode).setVersionedComponentId(expectedId);
+ }
+
+ @Test
+ void pasteWithoutDestinationIdentityLeavesServiceUnversioned() {
+ final ProcessGroup destination = createMockProcessGroup();
+ mockMigratingProcessorIn(destination);
+
+ synchronizer.addVersionedComponentsToProcessGroup(destination,
legacyProcessorAdditions(), synchronizationOptions);
+
+ verify(destination).addControllerService(controllerServiceNode);
+ verify(controllerServiceNode,
never()).setVersionedComponentId(any());
+ }
+
+ private void mockMigratingProcessorIn(final ProcessGroup owner) {
+ final ProcessorNode processor = createMockProcessor();
+
when(processor.getParentProcessGroup()).thenReturn(Optional.of(owner));
+ when(flowManager.createProcessor(any(), any(), any(),
eq(true))).thenReturn(processor);
+
+ final ControllerServiceCreationDetails details =
mock(ControllerServiceCreationDetails.class);
+ when(details.serviceIdentifier()).thenReturn("new-service");
+ when(details.type()).thenReturn(SERVICE_TYPE);
+
when(details.serviceBundleCoordinate()).thenReturn(bundleCoordinate);
+ when(details.serviceProperties()).thenReturn(SERVICE_PROPERTIES);
+
when(controllerServiceNode.getPropertyDescriptor("mode")).thenReturn(new
PropertyDescriptor.Builder().name("mode").build());
+
+ doAnswer(invocation -> {
+ final ControllerServiceFactory serviceFactory =
invocation.getArgument(1);
+ serviceFactory.create(details);
+ return null;
+ }).when(processor).migrateConfiguration(anyMap(),
any(ControllerServiceFactory.class));
+ }
+
+ private VersionedComponentAdditions legacyProcessorAdditions() {
+ final VersionedProcessor processor =
createMinimalVersionedProcessor();
+ processor.setGroupIdentifier("source-group-not-destination");
+ return new
VersionedComponentAdditions.Builder().setProcessors(Set.of(processor)).build();
+ }
+
+ private String expectedMigrationServiceId(final String
owningGroupIdentity) {
+ final FlowManager expectedFlowManager = mock(FlowManager.class);
+ final ControllerServiceNode expectedService =
mock(ControllerServiceNode.class);
+ when(expectedService.getPropertyDescriptor("mode")).thenReturn(new
PropertyDescriptor.Builder().name("mode").build());
+ when(expectedFlowManager.createControllerService(eq(SERVICE_TYPE),
anyString(), any(), anySet(), eq(true), eq(true),
eq(null))).thenReturn(expectedService);
+
+ final ComponentNode creator = mock(ComponentNode.class);
+
when(creator.getParentProcessGroup()).thenReturn(Optional.of(mock(ProcessGroup.class)));
+ final StandardControllerServiceFactory factory = new
StandardControllerServiceFactory(
+ mock(ExtensionManager.class), expectedFlowManager,
mock(ControllerServiceProvider.class), creator, owningGroupIdentity);
+ final ControllerServiceCreationDetails details =
mock(ControllerServiceCreationDetails.class);
+ when(details.serviceIdentifier()).thenReturn("expected-service");
+ when(details.type()).thenReturn(SERVICE_TYPE);
+
when(details.serviceBundleCoordinate()).thenReturn(bundleCoordinate);
+ when(details.serviceProperties()).thenReturn(SERVICE_PROPERTIES);
+
+ factory.create(details);
+
+ final ArgumentCaptor<String> versionedId =
ArgumentCaptor.forClass(String.class);
+
verify(expectedService).setVersionedComponentId(versionedId.capture());
+ return versionedId.getValue();
+ }
+ }
Review Comment:
The tests compare IDs between runs, and `expectedMigrationServiceId` is
gone. The order test runs in both directions with a `LinkedHashSet`. It uses
Parameter-referenced secrets plus one inline secret rather than an old property
name, because the rename happens in the service's own migration, which the
system test covers.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizerTest.java:
##########
@@ -552,6 +571,352 @@ public void testAddProcessorWithServiceAndMigration() {
assertEquals(controllerServiceNode.getIdentifier(),
migratedProperties.get("cs"));
}
+ @Nested
+ class MigrationCreatedControllerService {
+
+ private static final String SNAPSHOT_ROOT_ID = "snapshot-root-id";
+ private static final String LOCAL_CHILD_ID = "local-child";
+ private static final String CHILD_GROUP_ID = "child-group-id";
+ private static final String PARENT_GROUP_ID = "parent-group-id";
+ private static final String CHILD_PLACEMENT_ID = "child-placement-id";
+ private static final String DESTINATION_GROUP_ID =
"destination-group-id";
+
+ private static final String INCOMING_FLOW_ID = "incoming-flow-id";
+ private static final String DESTINATION_FLOW_ID =
"destination-flow-id";
+ private static final String PARENT_FLOW_ID = "parent-flow-id";
+ private static final String CHILD_FLOW_ID = "child-flow-id";
+
+ private static final String REGISTRY_ID = "registry-id";
+ private static final String SERVICE_TYPE =
TestControllerService.class.getName();
+ private static final Map<String, String> SERVICE_PROPERTIES =
Map.of("mode", "read");
+
+ @Test
+ void
migrationInSynchronizedGroupUsesIncomingFlowIdAfterVersionControlIsCleared() {
+ final ProcessGroup targetGroup = createMockProcessGroup();
+ final VersionControlInformation oldVersionControl =
mock(VersionControlInformation.class);
+
when(oldVersionControl.getFlowIdentifier()).thenReturn("old-flow-id");
+
+ final AtomicReference<VersionControlInformation>
liveVersionControl = new AtomicReference<>(oldVersionControl);
+
when(targetGroup.getVersionControlInformation()).thenAnswer(invocation ->
liveVersionControl.get());
+ doAnswer(invocation -> {
+ liveVersionControl.set(null);
+ return null;
+ }).when(targetGroup).disconnectVersionControl();
+
+ final VersionedProcessGroup proposed = new VersionedProcessGroup();
+ proposed.setIdentifier(SNAPSHOT_ROOT_ID);
+ proposed.setProcessors(Set.of(createMinimalVersionedProcessor()));
+
+ final VersionedExternalFlow incoming = new VersionedExternalFlow();
+ incoming.setFlowContents(proposed);
+
+ final VersionedExternalFlowMetadata metadata = new
VersionedExternalFlowMetadata();
+ metadata.setFlowIdentifier(INCOMING_FLOW_ID);
+ incoming.setMetadata(metadata);
+
+ mockMigratingProcessorIn(targetGroup);
+
+ synchronizer.synchronize(targetGroup, incoming,
synchronizationOptions);
+
+ assertNull(targetGroup.getVersionControlInformation());
+ final String expectedId =
expectedMigrationServiceId(INCOMING_FLOW_ID);
+ verify(controllerServiceNode).setVersionedComponentId(expectedId);
+ }
+
+ @Test
+ void migrationInChildWithoutOwnFlowUsesChildVersionedId() {
+ final ProcessGroup parentGroup = createMockProcessGroup();
+ final ProcessGroup childGroup = createMockProcessGroup();
+ when(childGroup.getIdentifier()).thenReturn(LOCAL_CHILD_ID);
+ when(childGroup.getParent()).thenReturn(parentGroup);
+
when(childGroup.getVersionedComponentId()).thenReturn(Optional.of(CHILD_GROUP_ID));
+
when(parentGroup.getProcessGroups()).thenReturn(Set.of(childGroup));
+
+ final VersionedProcessGroup proposedChild = new
VersionedProcessGroup();
+ proposedChild.setIdentifier(CHILD_GROUP_ID);
+
proposedChild.setProcessors(Set.of(createMinimalVersionedProcessor()));
+
+ final VersionedProcessGroup proposedParent = new
VersionedProcessGroup();
+ proposedParent.setIdentifier(PARENT_GROUP_ID);
+ proposedParent.setProcessGroups(Set.of(proposedChild));
+
+ final VersionedExternalFlow incoming = new VersionedExternalFlow();
+ incoming.setFlowContents(proposedParent);
+
+ final VersionedExternalFlowMetadata metadata = new
VersionedExternalFlowMetadata();
+ metadata.setFlowIdentifier(PARENT_FLOW_ID);
+ incoming.setMetadata(metadata);
+
+ mockMigratingProcessorIn(childGroup);
+
+ synchronizer.synchronize(parentGroup, incoming,
synchronizationOptions);
+
+ final String expectedId =
expectedMigrationServiceId(CHILD_GROUP_ID);
+ verify(controllerServiceNode).setVersionedComponentId(expectedId);
+ }
+
+ @Test
+ void migrationInIndependentlyVersionedChildUsesChildFlowId() {
+ final ProcessGroup parentGroup = createMockProcessGroup();
+ final ProcessGroup childGroup = createMockProcessGroup();
+ when(childGroup.getIdentifier()).thenReturn(LOCAL_CHILD_ID);
+ when(childGroup.getParent()).thenReturn(parentGroup);
+
when(childGroup.getVersionedComponentId()).thenReturn(Optional.of(CHILD_PLACEMENT_ID));
+
when(parentGroup.getProcessGroups()).thenReturn(Set.of(childGroup));
+
+ final AtomicReference<VersionControlInformation>
childVersionControl = new AtomicReference<>();
+
when(childGroup.getVersionControlInformation()).thenAnswer(invocation ->
childVersionControl.get());
+ doAnswer(invocation -> {
+ childVersionControl.set(invocation.getArgument(0));
+ return null;
+ }).when(childGroup).setVersionControlInformation(any(), anyMap());
+
+
when(flowManager.getFlowRegistryClient(REGISTRY_ID)).thenReturn(mock(FlowRegistryClientNode.class));
+ final VersionedFlowCoordinates coordinates = new
VersionedFlowCoordinates();
+ coordinates.setRegistryId(REGISTRY_ID);
+ coordinates.setBucketId("bucket");
+ coordinates.setFlowId(CHILD_FLOW_ID);
+ coordinates.setVersion("1");
+
+ final VersionedProcessGroup proposedChild = new
VersionedProcessGroup();
+ proposedChild.setIdentifier(CHILD_PLACEMENT_ID);
+ proposedChild.setVersionedFlowCoordinates(coordinates);
+
proposedChild.setProcessors(Set.of(createMinimalVersionedProcessor()));
+
+ final VersionedProcessGroup proposedParent = new
VersionedProcessGroup();
+ proposedParent.setIdentifier(PARENT_GROUP_ID);
+ proposedParent.setProcessGroups(Set.of(proposedChild));
+
+ final VersionedExternalFlow incoming = new VersionedExternalFlow();
+ incoming.setFlowContents(proposedParent);
+
+ final VersionedExternalFlowMetadata metadata = new
VersionedExternalFlowMetadata();
+ metadata.setFlowIdentifier(PARENT_FLOW_ID);
+ incoming.setMetadata(metadata);
+
+ mockMigratingProcessorIn(childGroup);
+
+ synchronizer.synchronize(parentGroup, incoming,
synchronizationOptions);
+
+ assertNotNull(childGroup.getVersionControlInformation());
+ assertEquals(CHILD_FLOW_ID,
childGroup.getVersionControlInformation().getFlowIdentifier());
+ final String expectedId =
expectedMigrationServiceId(CHILD_FLOW_ID);
+ verify(controllerServiceNode).setVersionedComponentId(expectedId);
+ }
+
+ @Test
+ void pasteIntoVersionControlledGroupUsesDestinationFlowId() {
+ final ProcessGroup destination = createMockProcessGroup();
+
when(destination.getVersionedComponentId()).thenReturn(Optional.of(DESTINATION_GROUP_ID));
+
+ final VersionControlInformation versionControl =
mock(VersionControlInformation.class);
+
when(versionControl.getFlowIdentifier()).thenReturn(DESTINATION_FLOW_ID);
+
when(destination.getVersionControlInformation()).thenReturn(versionControl);
+
+ mockMigratingProcessorIn(destination);
+
+ synchronizer.addVersionedComponentsToProcessGroup(destination,
legacyProcessorAdditions(), synchronizationOptions);
+
+ final String expectedId =
expectedMigrationServiceId(DESTINATION_FLOW_ID);
+ verify(controllerServiceNode).setVersionedComponentId(expectedId);
+ }
+
+ @Test
+ void pasteIntoGroupWithoutOwnFlowUsesDestinationVersionedId() {
+ final ProcessGroup destination = createMockProcessGroup();
+
when(destination.getVersionedComponentId()).thenReturn(Optional.of(DESTINATION_GROUP_ID));
+ mockMigratingProcessorIn(destination);
+
+ synchronizer.addVersionedComponentsToProcessGroup(destination,
legacyProcessorAdditions(), synchronizationOptions);
+
+ final String expectedId =
expectedMigrationServiceId(DESTINATION_GROUP_ID);
+ verify(controllerServiceNode).setVersionedComponentId(expectedId);
+ }
+
+ @Test
+ void pasteWithoutDestinationIdentityLeavesServiceUnversioned() {
+ final ProcessGroup destination = createMockProcessGroup();
+ mockMigratingProcessorIn(destination);
+
+ synchronizer.addVersionedComponentsToProcessGroup(destination,
legacyProcessorAdditions(), synchronizationOptions);
+
+ verify(destination).addControllerService(controllerServiceNode);
+ verify(controllerServiceNode,
never()).setVersionedComponentId(any());
+ }
+
+ private void mockMigratingProcessorIn(final ProcessGroup owner) {
+ final ProcessorNode processor = createMockProcessor();
+
when(processor.getParentProcessGroup()).thenReturn(Optional.of(owner));
+ when(flowManager.createProcessor(any(), any(), any(),
eq(true))).thenReturn(processor);
+
+ final ControllerServiceCreationDetails details =
mock(ControllerServiceCreationDetails.class);
+ when(details.serviceIdentifier()).thenReturn("new-service");
+ when(details.type()).thenReturn(SERVICE_TYPE);
+
when(details.serviceBundleCoordinate()).thenReturn(bundleCoordinate);
+ when(details.serviceProperties()).thenReturn(SERVICE_PROPERTIES);
+
when(controllerServiceNode.getPropertyDescriptor("mode")).thenReturn(new
PropertyDescriptor.Builder().name("mode").build());
+
+ doAnswer(invocation -> {
+ final ControllerServiceFactory serviceFactory =
invocation.getArgument(1);
+ serviceFactory.create(details);
+ return null;
+ }).when(processor).migrateConfiguration(anyMap(),
any(ControllerServiceFactory.class));
+ }
+
+ private VersionedComponentAdditions legacyProcessorAdditions() {
+ final VersionedProcessor processor =
createMinimalVersionedProcessor();
+ processor.setGroupIdentifier("source-group-not-destination");
+ return new
VersionedComponentAdditions.Builder().setProcessors(Set.of(processor)).build();
+ }
+
+ private String expectedMigrationServiceId(final String
owningGroupIdentity) {
+ final FlowManager expectedFlowManager = mock(FlowManager.class);
+ final ControllerServiceNode expectedService =
mock(ControllerServiceNode.class);
+ when(expectedService.getPropertyDescriptor("mode")).thenReturn(new
PropertyDescriptor.Builder().name("mode").build());
+ when(expectedFlowManager.createControllerService(eq(SERVICE_TYPE),
anyString(), any(), anySet(), eq(true), eq(true),
eq(null))).thenReturn(expectedService);
+
+ final ComponentNode creator = mock(ComponentNode.class);
+
when(creator.getParentProcessGroup()).thenReturn(Optional.of(mock(ProcessGroup.class)));
+ final StandardControllerServiceFactory factory = new
StandardControllerServiceFactory(
+ mock(ExtensionManager.class), expectedFlowManager,
mock(ControllerServiceProvider.class), creator, owningGroupIdentity);
+ final ControllerServiceCreationDetails details =
mock(ControllerServiceCreationDetails.class);
+ when(details.serviceIdentifier()).thenReturn("expected-service");
+ when(details.type()).thenReturn(SERVICE_TYPE);
+
when(details.serviceBundleCoordinate()).thenReturn(bundleCoordinate);
+ when(details.serviceProperties()).thenReturn(SERVICE_PROPERTIES);
+
+ factory.create(details);
+
+ final ArgumentCaptor<String> versionedId =
ArgumentCaptor.forClass(String.class);
+
verify(expectedService).setVersionedComponentId(versionedId.capture());
+ return versionedId.getValue();
+ }
+ }
+
+ @Nested
+ class MissingControllerServiceRemoval {
+
+ private static final String ROOT_GROUP_ID = "root-group-id";
+ private static final String PARENT_GROUP_ID = "parent-group-id";
+ private static final String CHILD_GROUP_ID = "child-group-id";
+ private static final String VERSIONED_CHILD_GROUP_ID =
"versioned-child-group-id";
+
+ @Test
+ void
synchronizationKeepsMissingServicesReachableFromRemainingProcessor() {
+ final ProcessGroup targetGroup = createMockProcessGroup();
+ final ControllerServiceNode directlyReferenced =
createMockControllerService();
+ final ControllerServiceNode referencedByDirect =
createMockControllerService();
+ final ControllerServiceNode referencedByIndirect =
createMockControllerService();
+ setReferences(directlyReferenced, createMockProcessor());
+ setReferences(referencedByDirect, directlyReferenced);
+ setReferences(referencedByIndirect, referencedByDirect);
+
when(targetGroup.getControllerServices(false)).thenReturn(Set.of(directlyReferenced,
referencedByDirect, referencedByIndirect));
+
+ synchronizer.synchronize(targetGroup,
incomingFlowWithoutServices(), synchronizationOptions);
+
+ verify(controllerServiceProvider,
never()).removeControllerService(directlyReferenced);
+ verify(controllerServiceProvider,
never()).removeControllerService(referencedByDirect);
+ verify(controllerServiceProvider,
never()).removeControllerService(referencedByIndirect);
+ }
+
+ @Test
+ void
synchronizationRemovesMissingServicesReferencedOnlyByOtherMissingServices() {
+ final ProcessGroup targetGroup = createMockProcessGroup();
+ final ControllerServiceNode cycleFirst =
createMockControllerService();
+ final ControllerServiceNode cycleSecond =
createMockControllerService();
+ final ControllerServiceNode unreferenced =
createMockControllerService();
+ setReferences(cycleFirst, cycleSecond);
+ setReferences(cycleSecond, cycleFirst);
+
when(targetGroup.getControllerServices(false)).thenReturn(Set.of(cycleFirst,
cycleSecond, unreferenced));
+
+ synchronizer.synchronize(targetGroup,
incomingFlowWithoutServices(), synchronizationOptions);
+
+
verify(controllerServiceProvider).removeControllerService(cycleFirst);
+
verify(controllerServiceProvider).removeControllerService(cycleSecond);
+
verify(controllerServiceProvider).removeControllerService(unreferenced);
+ }
+
+ @Test
+ void
synchronizationKeepsMissingServiceReferencedAfterPropertyMigration() {
+ final ProcessGroup targetGroup = createMockProcessGroup();
+ final ControllerServiceNode service =
createMockControllerService();
+
when(targetGroup.getControllerServices(false)).thenReturn(Set.of(service));
+
+ final ProcessorNode processor = createMockProcessor();
+
when(processor.getParentProcessGroup()).thenReturn(Optional.of(targetGroup));
+ when(flowManager.createProcessor(any(), any(), any(),
eq(true))).thenReturn(processor);
+ doAnswer(invocation ->
service.getReferences().getReferencingComponents().add(processor))
+ .when(processor).migrateConfiguration(anyMap(),
any(ControllerServiceFactory.class));
+
+ final VersionedProcessGroup proposed = new VersionedProcessGroup();
+ proposed.setIdentifier(ROOT_GROUP_ID);
+ proposed.setProcessors(Set.of(createMinimalVersionedProcessor()));
+ final VersionedExternalFlow incoming = new VersionedExternalFlow();
+ incoming.setFlowContents(proposed);
+
+ synchronizer.synchronize(targetGroup, incoming,
synchronizationOptions);
+
+ verify(controllerServiceProvider,
never()).removeControllerService(service);
+ }
+
+ @Test
+ void
synchronizationRemovesMissingServicesOnlyInSynchronizedChildGroups() {
+ final ProcessGroup parentGroup = createMockProcessGroup();
+ final ProcessGroup childGroup = createMockProcessGroup();
+
when(childGroup.getVersionedComponentId()).thenReturn(Optional.of(CHILD_GROUP_ID));
+ final ProcessGroup versionedChildGroup = createMockProcessGroup();
+
when(versionedChildGroup.getVersionedComponentId()).thenReturn(Optional.of(VERSIONED_CHILD_GROUP_ID));
+ when(parentGroup.getProcessGroups()).thenReturn(Set.of(childGroup,
versionedChildGroup));
+
+ final ControllerServiceNode childService =
createMockControllerService();
+
when(childGroup.getControllerServices(false)).thenReturn(Set.of(childService));
+ final ControllerServiceNode versionedChildService =
createMockControllerService();
+
when(versionedChildGroup.getControllerServices(false)).thenReturn(Set.of(versionedChildService));
+
+ final ProcessorNode versionedChildProcessor =
createMappableProcessor(versionedChildGroup);
+
when(versionedChildGroup.getProcessors()).thenReturn(Set.of(versionedChildProcessor));
+ final ControllerServiceNode parentService =
createMockControllerService();
+ setReferences(parentService, versionedChildProcessor);
+
when(parentGroup.getControllerServices(false)).thenReturn(Set.of(parentService));
+
+ final VersionedFlowCoordinates coordinates = new
VersionedFlowCoordinates();
+ coordinates.setRegistryId("registry-id");
+ coordinates.setBucketId("bucket");
+ coordinates.setFlowId("child-flow-id");
+ coordinates.setVersion("1");
+
+ final VersionedProcessGroup proposedChild = new
VersionedProcessGroup();
+ proposedChild.setIdentifier(CHILD_GROUP_ID);
+ final VersionedProcessGroup proposedVersionedChild = new
VersionedProcessGroup();
+ proposedVersionedChild.setIdentifier(VERSIONED_CHILD_GROUP_ID);
+ proposedVersionedChild.setVersionedFlowCoordinates(coordinates);
+ final VersionedProcessGroup proposedParent = new
VersionedProcessGroup();
+ proposedParent.setIdentifier(PARENT_GROUP_ID);
+ proposedParent.setProcessGroups(Set.of(proposedChild,
proposedVersionedChild));
+ final VersionedExternalFlow incoming = new VersionedExternalFlow();
+ incoming.setFlowContents(proposedParent);
+
+ final FlowSynchronizationOptions parentOnlyOptions =
FlowSynchronizationOptions.Builder.from(synchronizationOptions)
+ .updateDescendantVersionedFlows(false)
+ .build();
+
+ synchronizer.synchronize(parentGroup, incoming, parentOnlyOptions);
+
+
verify(controllerServiceProvider).removeControllerService(childService);
+ verify(controllerServiceProvider,
never()).removeControllerService(versionedChildService);
+ verify(controllerServiceProvider,
never()).removeControllerService(parentService);
+ }
+
+ private VersionedExternalFlow incomingFlowWithoutServices() {
+ final VersionedProcessGroup proposed = new VersionedProcessGroup();
+ proposed.setIdentifier(ROOT_GROUP_ID);
+
+ final VersionedExternalFlow incoming = new VersionedExternalFlow();
+ incoming.setFlowContents(proposed);
+ return incoming;
+ }
+ }
Review Comment:
Folded and strengthened as suggested; the child-group test is removed.
##########
nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/migration/MigrationCreatedControllerServiceVersioningIT.java:
##########
@@ -0,0 +1,323 @@
+/*
+ * 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.
+ */
+
+package org.apache.nifi.tests.system.migration;
+
+import org.apache.nifi.tests.system.AbstractNarSwapMigrationIT;
+import org.apache.nifi.toolkit.client.NiFiClientException;
+import org.apache.nifi.web.api.dto.ComponentStateDTO;
+import org.apache.nifi.web.api.dto.ProcessorDTO;
+import org.apache.nifi.web.api.dto.StateEntryDTO;
+import org.apache.nifi.web.api.entity.ComponentStateEntity;
+import org.apache.nifi.web.api.entity.ControllerServiceEntity;
+import org.apache.nifi.web.api.entity.FlowRegistryClientEntity;
+import org.apache.nifi.web.api.entity.ProcessGroupEntity;
+import org.apache.nifi.web.api.entity.ProcessorEntity;
+import org.apache.nifi.web.api.entity.VersionedFlowUpdateRequestEntity;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Verifies that startup and import migration create matching Controller
Service identities and that
+ * flow version changes preserve the service instance, references, and state.
+ */
+class MigrationCreatedControllerServiceVersioningIT extends
AbstractNarSwapMigrationIT {
+
+ private static final String TEST_FLOWS_BUCKET = "test-flows";
+ private static final String FLOW_ID =
"0831859c-f4a1-4ff6-99b3-58aeb35cb17f";
+ private static final String FLOW_DEFINITION_FILENAME = "snapshot.json";
+ private static final String VERSION_1 = "1";
+ private static final String VERSION_2 = "2";
+ private static final String VERSION_3 = "3";
+ private static final Path FLOW_SNAPSHOT =
Path.of("src/test/resources/versioned-flows", TEST_FLOWS_BUCKET, FLOW_ID,
VERSION_1, FLOW_DEFINITION_FILENAME);
+
+ private static final String STORE_SERVICE_PROPERTY = "Store Service";
+ private static final String STORE_SERVICE_TYPE = TEST_CS_PACKAGE +
".StateBackedStoreService";
+
+ private static final String MIGRATING_PROCESSOR_NAME =
"MigrateToControllerService";
+ private static final String PROCESSOR_TYPE = TEST_PROCESSORS_PACKAGE + "."
+ MIGRATING_PROCESSOR_NAME;
+ private static final String ADDED_PROCESSOR_NAME = "Added Processor";
+
+ private static final String ROOT_PG = "root";
+
+ private static final String CREATED_STATE_KEY = "created";
+ private static final String ROW_COUNT_STATE_KEY = "rowCount";
+ private static final long CONDITION_POLL_MILLIS = 100L;
+
+ @TempDir
+ Path registryDirectory;
+
+ @Test
+ void testStartupAndImportMigratedServicesSurviveFlowVersionUpdates()
throws NiFiClientException, IOException, InterruptedException {
Review Comment:
Strengthened in the same single method: a required password renamed by the
service, a second processor with shared and separate services, and a consumer
that skipped a version. One difference: instead of a different inline password
per instance, the legacy snapshot references `#{password}`, because a service
with an inline secret stays unversioned. Secret separation is covered by the
flows where processors use different secrets.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/migration/TestStandardControllerServiceFactory.java:
##########
@@ -142,6 +155,144 @@ public void testServiceIdDeterministic() {
assertEquals(4, Set.of(initialServiceId, secondGroupId, thirdId,
fourthId).size());
}
+ @Nested
+ class VersionedIdAssignment {
+ private static final String ENDPOINT_PROPERTY = "Endpoint";
+ private static final String ENDPOINT_PARAMETER = "#{endpoint}";
+ private static final String MODE_PROPERTY = "Mode";
+ private static final String READ_MODE = "read";
+ private static final String PASSWORD_PROPERTY = "Password";
+
+ private static final String OWNING_GROUP_IDENTITY = "shared-version";
+
+ private static final String CONSUMER_GROUP_ID = "consumer-instance";
+
+ @Test
+ void testVersionedIdentityDependsOnSuppliedIdentityTypeAndProperties()
{
+ final Map<String, String> properties = new LinkedHashMap<>();
+ properties.put(ENDPOINT_PROPERTY, ENDPOINT_PARAMETER);
+ properties.put(MODE_PROPERTY, READ_MODE);
+ final String first =
createServiceAndGetVersionedId("author-instance", OWNING_GROUP_IDENTITY,
IMPL_CLASS, properties);
+
+ final Map<String, String> reordered = new LinkedHashMap<>();
+ reordered.put(MODE_PROPERTY, READ_MODE);
+ reordered.put(ENDPOINT_PROPERTY, ENDPOINT_PARAMETER);
+ final String second =
createServiceAndGetVersionedId(CONSUMER_GROUP_ID, OWNING_GROUP_IDENTITY,
IMPL_CLASS, reordered);
+
+ assertNotNull(first);
+ assertEquals(first, second);
+ assertNotEquals(first,
createServiceAndGetVersionedId(CONSUMER_GROUP_ID, "different-version",
IMPL_CLASS, properties));
+ assertNotEquals(first,
createServiceAndGetVersionedId(CONSUMER_GROUP_ID, OWNING_GROUP_IDENTITY,
"AnotherService", properties));
+ assertNotEquals(first,
createServiceAndGetVersionedId(CONSUMER_GROUP_ID, OWNING_GROUP_IDENTITY,
IMPL_CLASS, Map.of()));
+ }
+
+ @Test
+ void testSensitiveValuesDoNotAffectVersionedIdentity() {
+ final Map<String, String> authorProperties =
Map.of(ENDPOINT_PROPERTY, ENDPOINT_PARAMETER, PASSWORD_PROPERTY,
"author-password");
+ final Map<String, String> consumerProperties =
Map.of(ENDPOINT_PROPERTY, ENDPOINT_PARAMETER, PASSWORD_PROPERTY,
"consumer-password");
+ final String authorId =
createServiceAndGetVersionedId("author-instance", OWNING_GROUP_IDENTITY,
IMPL_CLASS, authorProperties);
+ final String consumerId =
createServiceAndGetVersionedId(CONSUMER_GROUP_ID, OWNING_GROUP_IDENTITY,
IMPL_CLASS, consumerProperties);
+
+ assertEquals(authorId, consumerId);
+ assertEquals(authorId,
createServiceAndGetVersionedId(CONSUMER_GROUP_ID, OWNING_GROUP_IDENTITY,
IMPL_CLASS, Map.of(ENDPOINT_PROPERTY, ENDPOINT_PARAMETER)));
+ assertNotEquals(authorId,
createServiceAndGetVersionedId(CONSUMER_GROUP_ID, OWNING_GROUP_IDENTITY,
IMPL_CLASS,
+ Map.of(ENDPOINT_PROPERTY, "another-endpoint",
PASSWORD_PROPERTY, "consumer-password")));
+ assertNotEquals(factory.determineServiceId(IMPL_CLASS,
authorProperties), factory.determineServiceId(IMPL_CLASS, consumerProperties));
+ }
+
+ @Test
+ void testOccupiedVersionedIdentityLeavesNewServiceUnversioned() {
+ final Map<String, String> properties = Map.of(ENDPOINT_PROPERTY,
ENDPOINT_PARAMETER);
+ final String occupiedId =
createServiceAndGetVersionedId("author-instance", OWNING_GROUP_IDENTITY,
IMPL_CLASS, properties);
+ final ControllerServiceNode existingService =
mock(ControllerServiceNode.class);
+
when(existingService.getVersionedComponentId()).thenReturn(Optional.of(occupiedId));
+
+ final ControllerServiceNode newService =
createService(CONSUMER_GROUP_ID, OWNING_GROUP_IDENTITY, IMPL_CLASS, properties,
Set.of(existingService));
+
+ verify(newService, never()).setVersionedComponentId(any());
+ verify(existingService, never()).setVersionedComponentId(any());
+ verify(newService).migrateConfiguration(eq(properties), any());
+ }
+
+ @Test
+ void testUnversionedAndDifferentVersionedServicesDoNotOccupyIdentity()
{
+ final Map<String, String> properties = Map.of(ENDPOINT_PROPERTY,
ENDPOINT_PARAMETER);
+ final String expectedId =
createServiceAndGetVersionedId("author-instance", OWNING_GROUP_IDENTITY,
IMPL_CLASS, properties);
+ final ControllerServiceNode unversionedService =
mock(ControllerServiceNode.class);
+
when(unversionedService.getVersionedComponentId()).thenReturn(Optional.empty());
+ final ControllerServiceNode differentlyVersionedService =
mock(ControllerServiceNode.class);
+
when(differentlyVersionedService.getVersionedComponentId()).thenReturn(Optional.of("another-versioned-id"));
+
+ final ControllerServiceNode newService =
createService(CONSUMER_GROUP_ID, OWNING_GROUP_IDENTITY, IMPL_CLASS, properties,
+ Set.of(unversionedService, differentlyVersionedService));
+
+ verify(newService).setVersionedComponentId(expectedId);
+ verify(unversionedService, never()).setVersionedComponentId(any());
+ verify(differentlyVersionedService,
never()).setVersionedComponentId(any());
+ }
+
+ @Test
+ void testCreationWithoutOwningGroupIdentity() {
+ final ComponentNode creator = mock(ComponentNode.class);
+ final FlowManager flowManager = mock(FlowManager.class);
+ final StandardControllerServiceFactory factory = new
StandardControllerServiceFactory(
+ mock(ExtensionManager.class), flowManager,
mock(ControllerServiceProvider.class), creator);
+ final ControllerServiceNode service =
mock(ControllerServiceNode.class);
+ final ControllerServiceCreationDetails details = new
ControllerServiceCreationDetails("service-id", IMPL_CLASS,
+ createCoordinate(LONE_BUNDLE, VERSION_2), Map.of(),
ControllerServiceCreationDetails.CreationState.SERVICE_TO_BE_CREATED);
+ when(flowManager.createControllerService(eq(IMPL_CLASS),
anyString(), any(), any(), eq(true), eq(true), eq(null))).thenReturn(service);
+ when(creator.getParentProcessGroup()).thenReturn(Optional.empty());
+
+ factory.create(details);
+
+ verify(flowManager).addRootControllerService(service);
+ verify(service, never()).setVersionedComponentId(any());
+
+ final ProcessGroup group = mock(ProcessGroup.class);
+
when(creator.getParentProcessGroup()).thenReturn(Optional.of(group));
+
+ factory.create(details);
+
+ verify(group).addControllerService(service);
+ verify(service, never()).setVersionedComponentId(any());
+ }
+
+ private String createServiceAndGetVersionedId(final String
localGroupId, final String owningGroupIdentity, final String type, final
Map<String, String> properties) {
+ final ControllerServiceNode service = createService(localGroupId,
owningGroupIdentity, type, properties, Set.of());
+ final ArgumentCaptor<String> versionedId =
ArgumentCaptor.forClass(String.class);
+ verify(service).setVersionedComponentId(versionedId.capture());
+ return versionedId.getValue();
+ }
+
+ private ControllerServiceNode createService(final String localGroupId,
final String owningGroupIdentity, final String type,
+ final Map<String, String>
properties, final Set<ControllerServiceNode> existingServices) {
+ final ComponentNode component = mock(ComponentNode.class);
+ final ProcessGroup group = mock(ProcessGroup.class);
+
when(component.getParentProcessGroup()).thenReturn(Optional.of(group));
+
when(component.getProcessGroupIdentifier()).thenReturn(localGroupId);
+ final FlowManager manager = mock(FlowManager.class);
+ final ControllerServiceNode service =
mock(ControllerServiceNode.class);
+
when(service.getPropertyDescriptor(anyString())).thenAnswer(invocation -> {
+ final String name = invocation.getArgument(0);
+ return new
PropertyDescriptor.Builder().name(name).sensitive(PASSWORD_PROPERTY.equals(name)).build();
+ });
+ final Set<ControllerServiceNode> groupServices = new
HashSet<>(existingServices);
+ groupServices.add(service);
+ when(group.getControllerServices(false)).thenReturn(groupServices);
+ when(manager.createControllerService(eq(type), anyString(), any(),
any(), eq(true), eq(true), eq(null))).thenReturn(service);
+
+ final StandardControllerServiceFactory serviceFactory = new
StandardControllerServiceFactory(
+ mock(ExtensionManager.class), manager,
mock(ControllerServiceProvider.class), component, owningGroupIdentity);
+ final String instanceId = serviceFactory.determineServiceId(type,
properties);
+ final ControllerServiceCreationDetails details = new
ControllerServiceCreationDetails(instanceId, type,
createCoordinate(LONE_BUNDLE, VERSION_2), properties,
+
ControllerServiceCreationDetails.CreationState.SERVICE_TO_BE_CREATED);
+ serviceFactory.create(details);
+
+ verify(service).setProperties(properties);
+ return service;
+ }
+ }
Review Comment:
Removed; the file matches `main`.
--
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]
