markap14 commented on code in PR #11813:
URL: https://github.com/apache/nifi/pull/11813#discussion_r4460865282
##########
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:
[Claude Opus 5.5]
This check runs before `serviceNode.migrateConfiguration(...)` on line 130,
and that is where the service renames old property names to its current ones.
For a name the service doesn't recognize, `getPropertyDescriptor` returns a
non-sensitive placeholder, so the value is included in the hash.
The two real migrations that pass secrets both use old names:
- `AbstractAwsProcessor` passes `Secret Key`.
`AWSCredentialsProviderControllerService` only renames it to `Secret Access
Key` in its own `migrateProperties`.
- `ProxyServiceMigration` passes `proxy-user-password`.
`StandardProxyConfigurationService` calls it `Proxy User Password`.
So the password goes into the hash. The author and each consumer get
different IDs, the service is replaced on update and loses its state, and a
hash of the password is saved in the registry.
Please move this logic into the synchronizer, which runs after the service's
migration (see my comment on lines 428-445 of
`StandardVersionedComponentSynchronizer`). The factory shouldn't compute the ID
at all.
##########
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:
[Claude Opus 5.5]
This "first one created keeps the ID" rule can point a processor at another
processor's service.
Example: two AWS processors in the same group with different keys. With
sensitive values properly left out, both services hash to the same ID.
Whichever processor migrates first gets the ID, and the other service stays
unversioned. Migration order follows the order of a hash set of processors, so
it isn't guaranteed to be the same on the author and the consumer.
If the order differs, the consumer's ID is on processor 2's service while
the author's is on processor 1's. On the next version update, the synchronizer
points processor 1 at the service holding processor 2's credentials. Processor
2 gets a new, empty service.
Please change the factory so that it only records what it creates:
- Replace the `owningGroupIdentity` constructor argument and field (lines 62
and 69-76) with a `Collection<ControllerServiceNode> createdServices`. The
4-argument constructor, used by `VersionedFlowSynchronizer` for
controller-level components, can pass `new ArrayList<>()`.
- Right after `serviceNode.setProperties(...)` on line 114, add
`createdServices.add(serviceNode)`. It has to come before line 130, so that a
service is always in the list before any service its own migration creates.
- Pass the same `createdServices` into the nested factory on line 129.
- Delete lines 116-127 and `generateVersionedId` (lines 148-162).
##########
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:
[Claude Opus 5.5]
This builds the list of services to remove after property migration, so it
includes services that migration has just created. The factory enables those
right away. If one of them isn't referenced through a Controller Service
property, `removeControllerService` calls `verifyCanDelete`, which throws
because the service isn't disabled. That fails the whole update after
everything else has already been applied. At startup, NiFi would not start.
Recording the missing services during the per-group `synchronize`, before
migration runs, fixes this. It also replaces lines 1207-1321 with much less
code:
1. Add `private final Set<ControllerServiceNode> missingControllerServices =
new HashSet<>();` next to `createdAndModifiedExtensions` (line 180), and clear
it next to `createdAndModifiedExtensions.clear()` (line 351).
2. In the per-group `synchronize`, right before
`synchronizeControllerServices(...)` on line 616, add every entry of
`controllerServicesByVersionedId` whose key isn't the identifier of a service
in `proposed.getControllerServices()`. The per-group method only runs for
groups that are being synchronized, so no separate walk over child groups is
needed.
3. Replace this line with a call to:
```java
private void removeUnreferencedMissingControllerServices() {
final Set<ComponentNode> keptServices = new HashSet<>();
boolean serviceKept = true;
while (serviceKept) {
serviceKept = false;
for (final ControllerServiceNode service :
missingControllerServices) {
if (!keptServices.contains(service)) {
for (final ComponentNode referencingComponent :
service.getReferences().getReferencingComponents()) {
if
(!missingControllerServices.contains(referencingComponent) ||
keptServices.contains(referencingComponent)) {
keptServices.add(service);
serviceKept = true;
break;
}
}
}
}
}
for (final ControllerServiceNode service : missingControllerServices) {
if (keptServices.contains(service)) {
LOG.info("Keeping {} although the proposed flow does not contain
it, because it is still referenced by {}", service,
service.getReferences().getReferencingComponents());
} else {
LOG.info("Removing {} from {}", service,
service.getProcessGroup());
context.getControllerServiceProvider().removeControllerService(service);
}
}
}
```
This keeps chains of services that something still uses, and removes cycles
that nothing outside the cycle uses. Removal order doesn't matter, because
`StandardProcessGroup.removeControllerService` already clears the references
the removed service held.
4. Delete lines 1207-1321. On line 757, restore the original
`childCoordinates == null || syncOptions.isUpdateDescendantVersionedFlows()`
condition, and update the comment on lines 602-603 to say the services are
recorded here and removed at the end.
##########
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:
[Claude Opus 5.5]
With the IDs assigned in the synchronizer, please rework these tests around
the contract: the same inputs give the same ID on every instance, and different
inputs give different IDs. Compare the IDs captured from
`setVersionedComponentId` across runs, instead of rebuilding the expected value
with a second factory. `expectedMigrationServiceId` can then be deleted.
- Keep the synchronized-group test and the two child-group tests. Assert
that the ID changes when the flow ID or child versioned ID changes, not that it
equals a value rebuilt here.
- Keep one paste test. The other two cover the same branches of
`determineProcessGroupIdentity` as the child-group tests, and the factory no
longer has a "no identity" case to test.
- Add one test for migration order, which is the actual bug:
- Set up two processors in the same group. Each migration creates a
service of the same type that differs only in a sensitive value.
- Pass the sensitive value under an old name that the mock service
reports, after its own migration, under a sensitive descriptor with the current
name. That is what the AWS and proxy services do.
- Use a `LinkedHashSet` for `proposed.getProcessors()` so you can run the
sync once with the processors in one order and once in the other.
- Assert that each processor's service gets the same ID in both runs and
that the two IDs differ.
This test fails with the current first-wins rule.
##########
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:
[Claude Opus 5.5]
Please replace this loop (lines 316-330) with a call to the shared method
described in my comment on lines 428-445:
`migrateCreatedAndModifiedExtensions(group,
determineProcessGroupIdentity(group))`. A paste has to assign versioned IDs the
same way as an import or a version change.
##########
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:
[Claude Opus 5.5]
This test covers the right paths: startup and import migration, then updates
with and without the service. But it can't catch either problem described in
the review:
- The migration passes one value that isn't sensitive and keeps its name, so
every instance hashes the same input.
- There is only one service per group (the processor added in V2 has no
`Store Name`, so it creates nothing), so two services never compete for an ID.
Please strengthen this test rather than adding a new one:
1. **Add a secret that the service renames, like the AWS processors do.**
- Pre-upgrade `MigrateToControllerService`: add a required, sensitive
`Store Password`.
- Post-upgrade `MigrateToControllerService`: pass it as
`"store-password"`.
- `StateBackedStoreService`: declare a required, sensitive `Store
Password` and rename `store-password` to it in `migrateProperties`.
- `snapshot.json`: add the descriptor with no value.
- Before the upgrade, set different passwords on the author and on the
startup consumer. Set the import consumer's password on its service after
import.
Because the property is required, the existing "enabled" checks also show
that each secret survives the update. With the current code this fails on line
111.
2. **Add a second migrating processor with the same `Store Name` and a
different password.**
- Replace the "exactly one service" checks (`waitForSingleStoreService`
and line 220) with checks for each processor: the same service instance ID and
`created` timestamp before and after each update.
- Check that the two services have different versioned IDs.
- Check that each consumer processor's service has the same versioned ID
as the author's service for the same processor.
This shows that each processor keeps its own service and secret.
##########
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:
[Claude Opus 5.5]
These can come down to two tests:
- Fold the chain case (line 805) and the cycle case (line 823) into the
existing `testControllerServiceRemoved` (line 1844), which this PR already
strengthens.
- Its "unreferenced service is removed" case duplicates the `unreferenced`
service in the cycle test, so drop one of them.
- Its last `verify` (line 1878) repeats line 1856.
- Keep `synchronizationKeepsMissingServiceReferencedAfterPropertyMigration`,
which is the real regression test for the Jira. Strengthen it with a second
service that migration creates and nothing references, and verify that it isn't
removed. That covers the failure described in my comment on line 447 of the
synchronizer.
- `synchronizationRemovesMissingServicesOnlyInSynchronizedChildGroups` can
be removed. Once missing services are recorded in the per-group `synchronize`,
it only checks behavior that existed before this PR.
##########
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:
[Claude Opus 5.5]
This is where the versioned IDs should be assigned, after every migration in
the update has finished.
Please replace this loop (lines 428-445) and the near-copy in
`addVersionedComponentsToProcessGroup` (lines 316-330) with one private method
that both call. `synchronize` would call it with `group` and the metadata
`flowIdentifier`. `addVersionedComponentsToProcessGroup` would call it with
`group` and `determineProcessGroupIdentity(group)`. Roughly:
```java
private void migrateCreatedAndModifiedExtensions(final ProcessGroup
synchronizedGroup, final String synchronizedGroupIdentity) {
final List<ControllerServiceNode> createdServices = new ArrayList<>();
for (final CreatedOrModifiedExtension createdOrModifiedExtension :
createdAndModifiedExtensions) {
final ComponentNode extension =
createdOrModifiedExtension.extension();
final Map<String, String> originalPropertyValues =
createdOrModifiedExtension.propertyValues();
final ControllerServiceFactory serviceFactory = new
StandardControllerServiceFactory(context.getExtensionManager(),
context.getFlowManager(),
context.getControllerServiceProvider(), extension,
createdServices);
if (extension instanceof final ProcessorNode processor) {
processor.migrateConfiguration(originalPropertyValues,
serviceFactory);
} else if (extension instanceof final ControllerServiceNode service)
{
service.migrateConfiguration(originalPropertyValues,
serviceFactory);
} else if (extension instanceof final ReportingTaskNode task) {
task.migrateConfiguration(originalPropertyValues,
serviceFactory);
}
}
// A service is added to the list before any service that its own
migration creates, so the referencing service has its versioned ID first
for (final ControllerServiceNode service : createdServices) {
assignVersionedComponentId(service, synchronizedGroup,
synchronizedGroupIdentity);
}
}
```
`assignVersionedComponentId` builds the ID from the following parts. Each
one is the same on every instance and doesn't depend on migration order.
1. `getOwningProcessGroupIdentity(service, synchronizedGroup,
synchronizedGroupIdentity)`. If it is null, leave the service unversioned, as
now.
2. `service.getCanonicalClassName()`.
3. A `TreeMap` of property name to value from
`service.getRawPropertyValues()`. Look up each descriptor with
`service.getPropertyDescriptor(name)` so that sensitive dynamic properties are
recognized. Skip it if it is sensitive or if `getControllerServiceDefinition()
!= null`. Instance IDs of other services differ between instances, and the
reference is captured by step 4 of the referenced service anyway.
4. The sorted versioned IDs of
`service.getReferences().getReferencingComponents()`. Use a `switch` on
`ProcessorNode` and `ControllerServiceNode` to call
`getVersionedComponentId()`. Avoid an `instanceof VersionedComponent` check,
because `org.apache.nifi.flow.VersionedComponent` is already imported in this
class. If any referencing component has no versioned ID, leave the service
unversioned.
Hash the four parts with `UUID.nameUUIDFromBytes`, the same way as today. If
that ID is already used by another service in `service.getProcessGroup()`, add
a counter (1, 2, ...) to the hashed text until it's unused. That can only
happen when a single component's migration creates two services of the same
type that differ only in secret values. The order within one migration is fixed
by its code, so the counter gives the same result on every instance.
With this in place, the `occupied` check in the factory goes away, along
with the extra constructor argument.
##########
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:
[Claude Opus 5.5]
This processor has no run schedule, so it writes the service's local state
on every run, nonstop, for the whole test. The `created` value the service
writes when it's first enabled already proves the service wasn't recreated.
Please drop the row counting:
- `onTrigger` can do nothing.
- Remove `append` from `StoreService` and `StateBackedStoreService`, along
with the `rowCount` and `lastRow` keys.
- Remove `countRows` and the waits on lines 207 and 235-236 of
`MigrationCreatedControllerServiceVersioningIT`.
- The "still present" check in `readCreatedTimestamp` can also go, because
both callers have just confirmed the service exists.
##########
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:
[Claude Opus 5.5]
Once the factory only records the services it creates, this nested class can
be removed. The recording is simple wiring, and the synchronizer tests cover
it. That also removes two assertions about behavior this PR doesn't change: the
instance ID assertion on line 200, and the controller-level service path in
`testCreationWithoutOwningGroupIdentity`.
For what it's worth, the sensitive-value test passed only because the mock
service on line 277 already recognizes `Password` under the exact name passed
in. The real services only learn the current name in their own migration. The
synchronizer test should use an old name (see my comment there).
--
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]
