[PR] NIFI-16310 Preserve Controller Services created by migrateProperties during versioned flow synchronization [nifi]

["awelless (via GitHub)" <gi...@apache.org>]

awelless opened a new pull request, #11670:
URL: https://github.com/apache/nifi/pull/11670

   <!-- 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. -->
   
   # Summary
   
   [NIFI-16310](https://issues.apache.org/jira/browse/NIFI-16310)
   
   Controller Services created by `migrateProperties` method were always recreated during a versioned process group upgrade,  causing the trigger of `@OnRemoved` hook and loss of component state.
   Now migration-created services are matched with their counterparts in a versioned process group. If match is successful, a local controller service assumes a versioned id without being recreated.
   
   If no match happens, a controller services is kept unversioned, but not removed.
   
   ### Verification
   
   - Synchronizer unit tests.
   - System tests.
   
   ### Additional changes
   
   `FileSystemFlowRegistryClient` which is used in system tests was moved into a separate NAR, to be available before and after NAR swap.
   
   ## Suggested review order
   
   - `MigrationCreatedControllerServiceVersioningIT` and its dependencies to see the relevant use cases.
   - `StandardVersionedComponentSynchronizer` for actual implementation.
   - `StandardVersionedComponentSynchronizerTest` to see the desired behavior for edge cases.
   
   # Tracking
   
   Please complete the following tracking steps prior to pull request creation.
   
   ### Issue Tracking
   
   - [Apache NiFi Jira](https://issues.apache.org/jira/browse/NIFI) issue created
   
   ### Pull Request Tracking
   
   - Pull Request title starts with Apache NiFi Jira issue number, such as `NIFI-00000`
   - Pull Request commit message starts with Apache NiFi Jira issue number, as such `NIFI-00000`
   - Pull request contains [commits signed](https://docs.github.com/en/authentication/managing-commit-signature-verification/signing-commits) with a registered key indicating `Verified` status
   
   ### Pull Request Formatting
   
   - Pull Request based on current revision of the `main` branch
   - Pull Request refers to a feature branch with one commit containing changes
   
   # Verification
   
   Please indicate the verification steps performed prior to pull request creation.
   
   ### Build
   
   - [x] Build completed using `./mvnw clean install -P contrib-check`
     - [x] JDK 21
     - [ ] JDK 25
   
   ### Licensing
   
   - [ ] New dependencies are compatible with the [Apache License 2.0](https://apache.org/licenses/LICENSE-2.0) according to the [License Policy](https://www.apache.org/legal/resolved.html)
   - [ ] New dependencies are documented in applicable `LICENSE` and `NOTICE` files
   
   ### Documentation
   
   - [ ] Documentation formatting appears as expected in rendered files
   


-- 
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: issues-unsubscribe@nifi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] NIFI-16310 Preserve Controller Services created by migrateProperties during versioned flow synchronization [nifi]

["awelless (via GitHub)" <gi...@apache.org>]

awelless commented on code in PR #11670:
URL: https://github.com/apache/nifi/pull/11670#discussion_r3990190596


##########
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.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;
+
+/**
+ * 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 {

Review Comment:
   Initially I wanted to reuse `MigrateProperties` processor, but some of its modifications blocked flow upgrade, so I proceeded with a dedicated processor - controller service pair



##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java:
##########
@@ -1169,9 +1169,225 @@ private void removeMissingRpg(final ProcessGroup group, final VersionedProcessGr
         removeMissingComponents(group, proposed, rpgsByVersionedId, VersionedProcessGroup::getRemoteProcessGroups, ProcessGroup::removeRemoteProcessGroup);
     }
 
+    /**
+     * Assigns a proposed versioned id to a Controller Service created by property migration.
+     * The service must have exactly one referencer.
+     * The proposed counterpart of that referencer must point at a Controller Service of the same type.
+     * If those do not hold, the service stays unversioned.
+     * A proposed id is assigned to at most one local service.
+     */
+    private void assignVersionedIdsToMigrationCreatedControllerServices(final ProcessGroup group, final VersionedProcessGroup proposed) {
+        final Collection<ControllerServiceNode> groupServices = group.getControllerServices(false);
+        if (groupServices == null || groupServices.isEmpty()) {
+            return;
+        }
+
+        final List<ControllerServiceNode> migrationCreatedServices = new ArrayList<>();
+        for (final ControllerServiceNode localService : groupServices) {
+            if (localService.getVersionedComponentId().isEmpty() && isMigrationCreated(localService)) {
+                migrationCreatedServices.add(localService);
+            }
+        }
+
+        if (migrationCreatedServices.isEmpty()) {
+            return;
+        }
+
+        final Set<String> claimedVersionedIds = HashSet.newHashSet(groupServices.size());
+        for (final ControllerServiceNode localService : groupServices) {
+            localService.getVersionedComponentId().ifPresent(claimedVersionedIds::add);
+        }
+
+        final Map<String, VersionedConfigurableExtension> proposedComponentsByVersionedId = indexByVersionedId(proposed.getControllerServices(), proposed.getProcessors());
+
+        for (final ControllerServiceNode localService : orderByReferencerChain(migrationCreatedServices)) {
+            final ComponentNode referencer = getSoleReferencer(localService);
+            if (referencer == null) {
+                LOG.debug("Leaving {} in {} unversioned because it is not referenced by exactly one component", localService, group);
+                continue;
+            }
+
+            final VersionedConfigurableExtension proposedReferencer = getProposedReferencer(referencer, proposedComponentsByVersionedId);
+            if (proposedReferencer == null) {
+                LOG.debug("Leaving {} in {} unversioned because its referencer {} has no counterpart in the proposed flow", localService, group, referencer);
+                continue;
+            }
+
+            final ProposedControllerServiceMatch match = findMatchingProposedControllerService(localService, referencer, proposedReferencer, proposedComponentsByVersionedId, group);
+            if (match == null) {
+                LOG.debug("Leaving {} in {} unversioned because no proposed Controller Service matches the referencing property of {}", localService, group, referencer);
+                continue;
+            }
+
+            if (!claimedVersionedIds.add(match.versionedId())) {
+                LOG.debug("Leaving {} in {} unversioned because versioned id {} is already used by another Controller Service", localService, group, match.versionedId());
+                continue;
+            }
+
+            localService.setVersionedComponentId(match.versionedId());
+            updatedVersionedComponentIds.add(match.versionedId());
+            LOG.info("Matched {} in {} to the Controller Service with versioned id {} that the proposed flow declares, based on the {} property of {}",
+                localService, group, match.versionedId(), match.propertyName(), referencer);
+        }
+    }
+
+    private List<ControllerServiceNode> orderByReferencerChain(final List<ControllerServiceNode> migrationCreatedServices) {
+        final BitSet visited = new BitSet(migrationCreatedServices.size());
+        final List<ControllerServiceNode> ordered = new ArrayList<>(migrationCreatedServices.size());
+
+        for (int i = 0; i < migrationCreatedServices.size(); i++) {
+            appendReferencerChain(i, migrationCreatedServices, visited, ordered);
+        }
+
+        return ordered;
+    }
+
+    private void appendReferencerChain(
+            final int index,
+            final List<ControllerServiceNode> migrationCreatedServices,
+            final BitSet visited,
+            final List<ControllerServiceNode> ordered
+    ) {
+        if (visited.get(index)) {
+            return;
+        }
+        visited.set(index);
+
+        final ControllerServiceNode localService = migrationCreatedServices.get(index);
+        final ComponentNode referencer = getSoleReferencer(localService);
+        if (referencer instanceof ControllerServiceNode referencingService && isMigrationCreated(referencingService)) {
+            final int referencerIndex = migrationCreatedServices.indexOf(referencingService);
+            if (referencerIndex >= 0) {
+                // Visit the creator first so we can assign the versioned id to the creator first.
+                appendReferencerChain(referencerIndex, migrationCreatedServices, visited, ordered);
+            }
+        }
+
+        ordered.add(localService);
+    }
+
+    private boolean isMigrationCreated(final ControllerServiceNode service) {
+        return StandardControllerServiceFactory.MIGRATION_CREATED_COMMENT.equals(service.getComments());
+    }
+
+    private Map<String, VersionedConfigurableExtension> indexByVersionedId(
+            final Collection<? extends VersionedConfigurableExtension> controllerServices,
+            final Collection<? extends VersionedConfigurableExtension> processors
+    ) {
+        final int serviceCount = controllerServices == null ? 0 : controllerServices.size();
+        final int processorCount = processors == null ? 0 : processors.size();
+        final Map<String, VersionedConfigurableExtension> byVersionedId = HashMap.newHashMap(serviceCount + processorCount);
+        addByVersionedId(byVersionedId, controllerServices);
+        addByVersionedId(byVersionedId, processors);
+        return byVersionedId;
+    }
+
+    private void addByVersionedId(
+            final Map<String, VersionedConfigurableExtension> byVersionedId,
+            final Collection<? extends VersionedConfigurableExtension> components
+    ) {
+        if (components == null) {
+            return;
+        }
+
+        for (final VersionedConfigurableExtension component : components) {
+            byVersionedId.put(component.getIdentifier(), component);
+        }
+    }
+
+    private ComponentNode getSoleReferencer(final ControllerServiceNode localService) {
+        final ControllerServiceReference references = localService.getReferences();
+        if (references == null) {
+            return null;
+        }
+
+        final Set<ComponentNode> referencers = references.getReferencingComponents();
+        if (referencers == null || referencers.size() != 1) {
+            return null;
+        }
+
+        return referencers.iterator().next();
+    }
+
+    private VersionedConfigurableExtension getProposedReferencer(
+            final ComponentNode referencer,
+            final Map<String, VersionedConfigurableExtension> proposedComponentsByVersionedId
+    ) {
+        if (!(referencer instanceof org.apache.nifi.components.VersionedComponent versionedReferencer)) {
+            return null;
+        }
+
+        return versionedReferencer.getVersionedComponentId()
+            .map(proposedComponentsByVersionedId::get)
+            .orElse(null);
+    }
+
+    private ProposedControllerServiceMatch findMatchingProposedControllerService(
+            final ControllerServiceNode localService,
+            final ComponentNode referencer,
+            final VersionedConfigurableExtension proposedReferencer,
+            final Map<String, VersionedConfigurableExtension> proposedComponentsByVersionedId,
+            final ProcessGroup group
+    ) {
+        final Map<PropertyDescriptor, String> rawPropertyValues = referencer.getRawPropertyValues();
+        if (rawPropertyValues == null) {
+            LOG.debug("Leaving {} in {} unversioned because referencer {} has no property values", localService, group, referencer);
+            return null;
+        }
+
+        for (final Map.Entry<PropertyDescriptor, String> propertyEntry : rawPropertyValues.entrySet()) {
+            final PropertyDescriptor descriptor = propertyEntry.getKey();
+            final String propertyName = descriptor.getName();
+            if (descriptor.getControllerServiceDefinition() == null) {
+                continue;
+            }
+
+            if (!localService.getIdentifier().equals(propertyEntry.getValue())) {
+                continue;
+            }
+
+            final Map<String, String> proposedProperties = proposedReferencer.getProperties();
+            final String proposedServiceId = proposedProperties == null ? null : proposedProperties.get(propertyName);
+            if (proposedServiceId == null) {
+                LOG.debug("Leaving {} in {} unversioned because the proposed {} does not set the {} property",
+                    localService, group, proposedReferencer, propertyName);
+                continue;
+            }
+
+            // In versioned flow, the service identifier is the versioned component id of the service.
+            final VersionedConfigurableExtension proposedService = proposedComponentsByVersionedId.get(proposedServiceId);
+            if (proposedService == null || !proposedService.getType().equals(localService.getCanonicalClassName())) {
+                LOG.debug("Leaving {} in {} unversioned because proposed Controller Service {} is missing or has a different type than {}",
+                    localService, group, proposedServiceId, localService);
+                continue;
+            }
+
+            return new ProposedControllerServiceMatch(proposedServiceId, propertyName);
+        }
+
+        return null;
+    }
+
+    private record ProposedControllerServiceMatch(String versionedId, String propertyName) {
+    }
+
     private void removeMissingControllerServices(final ProcessGroup group, final VersionedProcessGroup proposed, final Map<String, ControllerServiceNode> servicesByVersionedId) {
-        final BiConsumer<ProcessGroup, ControllerServiceNode> componentRemoval = (grp, service) -> context.getControllerServiceProvider().removeControllerService(service);
-        removeMissingComponents(group, proposed, servicesByVersionedId, VersionedProcessGroup::getControllerServices, componentRemoval);
+        // Do not remove Controller Services created by migrateProperties.
+        final Map<String, ControllerServiceNode> servicesEligibleForRemoval = HashMap.newHashMap(servicesByVersionedId.size());
+        for (final Map.Entry<String, ControllerServiceNode> entry : servicesByVersionedId.entrySet()) {
+            final ControllerServiceNode service = entry.getValue();
+            if (isMigrationCreated(service)) {
+                if (service.getVersionedComponentId().isEmpty()) {
+                    LOG.info("Keeping {} in {} because it was created by property migration and is not present in the proposed flow",
+                            service, group);

Review Comment:
   I reckon that we shouldn't just drop migration services if they're absent in a versioned flow. So that if we run into any edge case, the service is preserved and a user has a chance to resolve the issue.



-- 
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: issues-unsubscribe@nifi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] NIFI-16310 Preserve Controller Services created by migrateProperties during versioned flow synchronization [nifi]

["exceptionfactory (via GitHub)" <gi...@apache.org>]

exceptionfactory commented on code in PR #11670:
URL: https://github.com/apache/nifi/pull/11670#discussion_r4184054938


##########
nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/MigrateToControllerService.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.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 org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.List;
+import java.util.Set;
+
+@CapabilityDescription("""
+        Pre-upgrade shape of a processor that keeps its store location in a plain property.
+        The post-upgrade shape of the same processor, in the alternate-config extensions bundle, migrates that property into a Controller Service.
+        """)
+@InputRequirement(Requirement.INPUT_FORBIDDEN)
+public class MigrateToControllerService extends AbstractProcessor {
+
+    static final PropertyDescriptor STORE_NAME = new PropertyDescriptor.Builder()
+            .name("store-name")
+            .displayName("Store Name")
+            .required(true)

Review Comment:
   The `displayName` is not needed and should be removed.
   
   ```suggestion
               .name("Store Name")
   ```



##########
nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/MigrateToControllerService.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.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 org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.List;
+import java.util.Set;
+
+@CapabilityDescription("""
+        Pre-upgrade shape of a processor that keeps its store location in a plain property.
+        The post-upgrade shape of the same processor, in the alternate-config extensions bundle, migrates that property into a Controller Service.
+        """)
+@InputRequirement(Requirement.INPUT_FORBIDDEN)
+public class MigrateToControllerService extends AbstractProcessor {
+
+    static final PropertyDescriptor STORE_NAME = new PropertyDescriptor.Builder()
+            .name("store-name")
+            .displayName("Store Name")
+            .required(true)
+            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+            .build();
+
+    static final Relationship REL_SUCCESS = new Relationship.Builder().name("success").build();
+
+    private static final List<PropertyDescriptor> PROPERTIES = List.of(STORE_NAME);
+    private static final Set<Relationship> RELATIONSHIPS = Set.of(REL_SUCCESS);
+
+    @Override
+    protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+        return PROPERTIES;
+    }
+
+    @Override
+    public Set<Relationship> getRelationships() {
+        return RELATIONSHIPS;
+    }
+
+    @Override
+    public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+        context.yield();

Review Comment:
   Is there a reason for calling `yield()` instead of having an empty body?



##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java:
##########
@@ -1169,9 +1172,228 @@ private void removeMissingRpg(final ProcessGroup group, final VersionedProcessGr
         removeMissingComponents(group, proposed, rpgsByVersionedId, VersionedProcessGroup::getRemoteProcessGroups, ProcessGroup::removeRemoteProcessGroup);
     }
 
+    /**
+     * Assigns a proposed versioned id to a Controller Service created by property migration.
+     * The service must have exactly one referencer.
+     * The proposed counterpart of that referencer must point at a Controller Service of the same type.
+     * If those do not hold, the service stays unversioned.
+     * A proposed id is assigned to at most one local service.
+     */
+    private void assignVersionedIdsToMigrationCreatedControllerServices(final ProcessGroup group, final VersionedProcessGroup proposed) {
+        final Collection<ControllerServiceNode> groupServices = group.getControllerServices(false);
+        if (groupServices.isEmpty()) {
+            return;
+        }
+
+        final List<ControllerServiceNode> migrationCreatedServices = getUnversionedMigrationCreatedServices(groupServices);
+        if (migrationCreatedServices.isEmpty()) {
+            return;
+        }
+
+        final Set<String> claimedVersionedIds = getClaimedVersionedIds(groupServices);
+
+        final Set<VersionedControllerService> proposedControllerServices = Objects.requireNonNullElse(proposed.getControllerServices(), Set.of());
+        final Set<VersionedProcessor> proposedProcessors = Objects.requireNonNullElse(proposed.getProcessors(), Set.of());
+        final Map<String, VersionedConfigurableExtension> proposedComponentsByVersionedId = indexByVersionedId(proposedControllerServices, proposedProcessors);
+
+        for (final ControllerServiceNode localService : orderByReferencerChain(migrationCreatedServices)) {
+            assignMatchingVersionedId(group, localService, claimedVersionedIds, proposedComponentsByVersionedId);
+        }
+    }
+
+    private List<ControllerServiceNode> getUnversionedMigrationCreatedServices(final Collection<ControllerServiceNode> groupServices) {
+        final List<ControllerServiceNode> migrationCreatedServices = new ArrayList<>();
+        for (final ControllerServiceNode localService : groupServices) {
+            if (localService.getVersionedComponentId().isEmpty() && isMigrationCreated(localService)) {
+                migrationCreatedServices.add(localService);
+            }
+        }
+        return migrationCreatedServices;
+    }
+
+    private Set<String> getClaimedVersionedIds(final Collection<ControllerServiceNode> groupServices) {
+        final Set<String> claimedVersionedIds = HashSet.newHashSet(groupServices.size());
+        for (final ControllerServiceNode localService : groupServices) {
+            localService.getVersionedComponentId().ifPresent(claimedVersionedIds::add);
+        }
+        return claimedVersionedIds;
+    }
+
+    private void assignMatchingVersionedId(
+            final ProcessGroup group,
+            final ControllerServiceNode localService,
+            final Set<String> claimedVersionedIds,
+            final Map<String, VersionedConfigurableExtension> proposedComponentsByVersionedId
+    ) {
+        final ComponentNode referencer = getSoleReferencer(localService);
+        if (referencer == null) {
+            LOG.debug("Leaving {} in {} unversioned because it is not referenced by exactly one component", localService, group);
+            return;
+        }
+
+        final VersionedConfigurableExtension proposedReferencer = getProposedReferencer(referencer, proposedComponentsByVersionedId);
+        if (proposedReferencer == null) {
+            LOG.debug("Leaving {} in {} unversioned because its referencer {} has no counterpart in the proposed flow", localService, group, referencer);
+            return;
+        }
+
+        final ProposedControllerServiceMatch match = findMatchingProposedControllerService(localService, referencer, proposedReferencer, proposedComponentsByVersionedId, group);
+        if (match == null) {
+            LOG.debug("Leaving {} in {} unversioned because no proposed Controller Service matches the referencing property of {}", localService, group, referencer);
+            return;
+        }
+
+        if (!claimedVersionedIds.add(match.versionedId())) {
+            LOG.debug("Leaving {} in {} unversioned because versioned id {} is already used by another Controller Service", localService, group, match.versionedId());
+            return;
+        }
+
+        localService.setVersionedComponentId(match.versionedId());
+        updatedVersionedComponentIds.add(match.versionedId());
+        LOG.info("Matched {} in {} to the Controller Service with versioned id {} that the proposed flow declares, based on the {} property of {}",
+            localService, group, match.versionedId(), match.propertyName(), referencer);
+    }
+
+    private List<ControllerServiceNode> orderByReferencerChain(final List<ControllerServiceNode> migrationCreatedServices) {

Review Comment:
   The reason behind the ordering is not quite clear, it would be helpful to add a comment that explains why the ordering is needed



##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java:
##########
@@ -1169,9 +1172,228 @@ private void removeMissingRpg(final ProcessGroup group, final VersionedProcessGr
         removeMissingComponents(group, proposed, rpgsByVersionedId, VersionedProcessGroup::getRemoteProcessGroups, ProcessGroup::removeRemoteProcessGroup);
     }
 
+    /**
+     * Assigns a proposed versioned id to a Controller Service created by property migration.
+     * The service must have exactly one referencer.
+     * The proposed counterpart of that referencer must point at a Controller Service of the same type.
+     * If those do not hold, the service stays unversioned.
+     * A proposed id is assigned to at most one local service.
+     */
+    private void assignVersionedIdsToMigrationCreatedControllerServices(final ProcessGroup group, final VersionedProcessGroup proposed) {
+        final Collection<ControllerServiceNode> groupServices = group.getControllerServices(false);
+        if (groupServices.isEmpty()) {
+            return;
+        }
+
+        final List<ControllerServiceNode> migrationCreatedServices = getUnversionedMigrationCreatedServices(groupServices);
+        if (migrationCreatedServices.isEmpty()) {
+            return;
+        }
+
+        final Set<String> claimedVersionedIds = getClaimedVersionedIds(groupServices);
+
+        final Set<VersionedControllerService> proposedControllerServices = Objects.requireNonNullElse(proposed.getControllerServices(), Set.of());
+        final Set<VersionedProcessor> proposedProcessors = Objects.requireNonNullElse(proposed.getProcessors(), Set.of());
+        final Map<String, VersionedConfigurableExtension> proposedComponentsByVersionedId = indexByVersionedId(proposedControllerServices, proposedProcessors);
+
+        for (final ControllerServiceNode localService : orderByReferencerChain(migrationCreatedServices)) {
+            assignMatchingVersionedId(group, localService, claimedVersionedIds, proposedComponentsByVersionedId);
+        }
+    }
+
+    private List<ControllerServiceNode> getUnversionedMigrationCreatedServices(final Collection<ControllerServiceNode> groupServices) {
+        final List<ControllerServiceNode> migrationCreatedServices = new ArrayList<>();
+        for (final ControllerServiceNode localService : groupServices) {
+            if (localService.getVersionedComponentId().isEmpty() && isMigrationCreated(localService)) {
+                migrationCreatedServices.add(localService);
+            }
+        }
+        return migrationCreatedServices;
+    }
+
+    private Set<String> getClaimedVersionedIds(final Collection<ControllerServiceNode> groupServices) {
+        final Set<String> claimedVersionedIds = HashSet.newHashSet(groupServices.size());
+        for (final ControllerServiceNode localService : groupServices) {
+            localService.getVersionedComponentId().ifPresent(claimedVersionedIds::add);
+        }
+        return claimedVersionedIds;
+    }
+
+    private void assignMatchingVersionedId(
+            final ProcessGroup group,
+            final ControllerServiceNode localService,
+            final Set<String> claimedVersionedIds,
+            final Map<String, VersionedConfigurableExtension> proposedComponentsByVersionedId
+    ) {
+        final ComponentNode referencer = getSoleReferencer(localService);
+        if (referencer == null) {
+            LOG.debug("Leaving {} in {} unversioned because it is not referenced by exactly one component", localService, group);
+            return;
+        }
+
+        final VersionedConfigurableExtension proposedReferencer = getProposedReferencer(referencer, proposedComponentsByVersionedId);
+        if (proposedReferencer == null) {
+            LOG.debug("Leaving {} in {} unversioned because its referencer {} has no counterpart in the proposed flow", localService, group, referencer);
+            return;
+        }
+
+        final ProposedControllerServiceMatch match = findMatchingProposedControllerService(localService, referencer, proposedReferencer, proposedComponentsByVersionedId, group);
+        if (match == null) {
+            LOG.debug("Leaving {} in {} unversioned because no proposed Controller Service matches the referencing property of {}", localService, group, referencer);
+            return;
+        }
+
+        if (!claimedVersionedIds.add(match.versionedId())) {
+            LOG.debug("Leaving {} in {} unversioned because versioned id {} is already used by another Controller Service", localService, group, match.versionedId());
+            return;
+        }
+
+        localService.setVersionedComponentId(match.versionedId());
+        updatedVersionedComponentIds.add(match.versionedId());
+        LOG.info("Matched {} in {} to the Controller Service with versioned id {} that the proposed flow declares, based on the {} property of {}",
+            localService, group, match.versionedId(), match.propertyName(), referencer);
+    }
+
+    private List<ControllerServiceNode> orderByReferencerChain(final List<ControllerServiceNode> migrationCreatedServices) {
+        final BitSet visited = new BitSet(migrationCreatedServices.size());
+        final List<ControllerServiceNode> ordered = new ArrayList<>(migrationCreatedServices.size());
+
+        for (int i = 0; i < migrationCreatedServices.size(); i++) {
+            appendReferencerChain(i, migrationCreatedServices, visited, ordered);
+        }
+
+        return ordered;
+    }
+
+    private void appendReferencerChain(
+            final int index,
+            final List<ControllerServiceNode> migrationCreatedServices,
+            final BitSet visited,
+            final List<ControllerServiceNode> ordered
+    ) {
+        if (visited.get(index)) {
+            return;
+        }
+        visited.set(index);
+
+        final ControllerServiceNode localService = migrationCreatedServices.get(index);
+        final ComponentNode referencer = getSoleReferencer(localService);
+        if (referencer instanceof ControllerServiceNode referencingService && isMigrationCreated(referencingService)) {
+            final int referencerIndex = migrationCreatedServices.indexOf(referencingService);
+            if (referencerIndex >= 0) {
+                // Visit the creator first so we can assign the versioned id to the creator first.
+                appendReferencerChain(referencerIndex, migrationCreatedServices, visited, ordered);
+            }
+        }
+
+        ordered.add(localService);
+    }
+
+    private boolean isMigrationCreated(final ControllerServiceNode service) {
+        return StandardControllerServiceFactory.MIGRATION_CREATED_COMMENT.equals(service.getComments());
+    }
+
+    private Map<String, VersionedConfigurableExtension> indexByVersionedId(
+            final Collection<? extends VersionedConfigurableExtension> controllerServices,
+            final Collection<? extends VersionedConfigurableExtension> processors
+    ) {
+        final Map<String, VersionedConfigurableExtension> byVersionedId = HashMap.newHashMap(controllerServices.size() + processors.size());
+        addByVersionedId(byVersionedId, controllerServices);
+        addByVersionedId(byVersionedId, processors);
+        return byVersionedId;
+    }
+
+    private void addByVersionedId(
+            final Map<String, VersionedConfigurableExtension> byVersionedId,
+            final Collection<? extends VersionedConfigurableExtension> components
+    ) {
+        for (final VersionedConfigurableExtension component : components) {
+            byVersionedId.put(component.getIdentifier(), component);
+        }
+    }
+
+    private ComponentNode getSoleReferencer(final ControllerServiceNode localService) {
+        final Set<ComponentNode> referencers = localService.getReferences().getReferencingComponents();
+        if (referencers.size() != 1) {
+            return null;
+        }
+
+        return referencers.iterator().next();
+    }
+
+    private VersionedConfigurableExtension getProposedReferencer(
+            final ComponentNode referencer,
+            final Map<String, VersionedConfigurableExtension> proposedComponentsByVersionedId
+    ) {
+        if (!(referencer instanceof org.apache.nifi.components.VersionedComponent versionedReferencer)) {
+            return null;
+        }
+
+        return versionedReferencer.getVersionedComponentId()
+            .map(proposedComponentsByVersionedId::get)
+            .orElse(null);
+    }
+
+    private ProposedControllerServiceMatch findMatchingProposedControllerService(
+            final ControllerServiceNode localService,
+            final ComponentNode referencer,
+            final VersionedConfigurableExtension proposedReferencer,
+            final Map<String, VersionedConfigurableExtension> proposedComponentsByVersionedId,
+            final ProcessGroup group
+    ) {
+        for (final Map.Entry<PropertyDescriptor, String> propertyEntry : referencer.getRawPropertyValues().entrySet()) {
+            final PropertyDescriptor descriptor = propertyEntry.getKey();
+            final String propertyName = descriptor.getName();
+            if (descriptor.getControllerServiceDefinition() == null) {
+                continue;
+            }
+
+            if (!localService.getIdentifier().equals(propertyEntry.getValue())) {
+                continue;
+            }
+
+            final Map<String, String> proposedProperties = proposedReferencer.getProperties();
+            final String proposedServiceId = proposedProperties == null ? null : proposedProperties.get(propertyName);
+            if (proposedServiceId == null) {
+                LOG.debug("Leaving {} in {} unversioned because the proposed {} does not set the {} property",
+                    localService, group, proposedReferencer, propertyName);
+                continue;
+            }
+
+            // In versioned flow, the service identifier is the versioned component id of the service.
+            final VersionedConfigurableExtension proposedService = proposedComponentsByVersionedId.get(proposedServiceId);
+            if (proposedService == null || !proposedService.getType().equals(localService.getCanonicalClassName())) {
+                LOG.debug("Leaving {} in {} unversioned because proposed Controller Service {} is missing or has a different type than {}",
+                    localService, group, proposedServiceId, localService);
+                continue;
+            }
+
+            return new ProposedControllerServiceMatch(proposedServiceId, propertyName);
+        }
+
+        return null;
+    }
+
+    private record ProposedControllerServiceMatch(String versionedId, String propertyName) {
+    }

Review Comment:
   I recommend declaring this at the end of the class instead of in between methods



##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java:
##########
@@ -1169,9 +1172,228 @@ private void removeMissingRpg(final ProcessGroup group, final VersionedProcessGr
         removeMissingComponents(group, proposed, rpgsByVersionedId, VersionedProcessGroup::getRemoteProcessGroups, ProcessGroup::removeRemoteProcessGroup);
     }
 
+    /**
+     * Assigns a proposed versioned id to a Controller Service created by property migration.
+     * The service must have exactly one referencer.
+     * The proposed counterpart of that referencer must point at a Controller Service of the same type.
+     * If those do not hold, the service stays unversioned.
+     * A proposed id is assigned to at most one local service.
+     */
+    private void assignVersionedIdsToMigrationCreatedControllerServices(final ProcessGroup group, final VersionedProcessGroup proposed) {
+        final Collection<ControllerServiceNode> groupServices = group.getControllerServices(false);
+        if (groupServices.isEmpty()) {
+            return;
+        }
+
+        final List<ControllerServiceNode> migrationCreatedServices = getUnversionedMigrationCreatedServices(groupServices);
+        if (migrationCreatedServices.isEmpty()) {
+            return;
+        }
+
+        final Set<String> claimedVersionedIds = getClaimedVersionedIds(groupServices);
+
+        final Set<VersionedControllerService> proposedControllerServices = Objects.requireNonNullElse(proposed.getControllerServices(), Set.of());
+        final Set<VersionedProcessor> proposedProcessors = Objects.requireNonNullElse(proposed.getProcessors(), Set.of());
+        final Map<String, VersionedConfigurableExtension> proposedComponentsByVersionedId = indexByVersionedId(proposedControllerServices, proposedProcessors);
+
+        for (final ControllerServiceNode localService : orderByReferencerChain(migrationCreatedServices)) {
+            assignMatchingVersionedId(group, localService, claimedVersionedIds, proposedComponentsByVersionedId);
+        }
+    }
+
+    private List<ControllerServiceNode> getUnversionedMigrationCreatedServices(final Collection<ControllerServiceNode> groupServices) {
+        final List<ControllerServiceNode> migrationCreatedServices = new ArrayList<>();
+        for (final ControllerServiceNode localService : groupServices) {
+            if (localService.getVersionedComponentId().isEmpty() && isMigrationCreated(localService)) {
+                migrationCreatedServices.add(localService);
+            }
+        }
+        return migrationCreatedServices;
+    }
+
+    private Set<String> getClaimedVersionedIds(final Collection<ControllerServiceNode> groupServices) {
+        final Set<String> claimedVersionedIds = HashSet.newHashSet(groupServices.size());
+        for (final ControllerServiceNode localService : groupServices) {
+            localService.getVersionedComponentId().ifPresent(claimedVersionedIds::add);
+        }
+        return claimedVersionedIds;
+    }
+
+    private void assignMatchingVersionedId(
+            final ProcessGroup group,
+            final ControllerServiceNode localService,
+            final Set<String> claimedVersionedIds,
+            final Map<String, VersionedConfigurableExtension> proposedComponentsByVersionedId
+    ) {
+        final ComponentNode referencer = getSoleReferencer(localService);
+        if (referencer == null) {
+            LOG.debug("Leaving {} in {} unversioned because it is not referenced by exactly one component", localService, group);
+            return;
+        }
+
+        final VersionedConfigurableExtension proposedReferencer = getProposedReferencer(referencer, proposedComponentsByVersionedId);
+        if (proposedReferencer == null) {
+            LOG.debug("Leaving {} in {} unversioned because its referencer {} has no counterpart in the proposed flow", localService, group, referencer);
+            return;
+        }
+
+        final ProposedControllerServiceMatch match = findMatchingProposedControllerService(localService, referencer, proposedReferencer, proposedComponentsByVersionedId, group);
+        if (match == null) {
+            LOG.debug("Leaving {} in {} unversioned because no proposed Controller Service matches the referencing property of {}", localService, group, referencer);
+            return;
+        }
+
+        if (!claimedVersionedIds.add(match.versionedId())) {
+            LOG.debug("Leaving {} in {} unversioned because versioned id {} is already used by another Controller Service", localService, group, match.versionedId());
+            return;
+        }
+
+        localService.setVersionedComponentId(match.versionedId());
+        updatedVersionedComponentIds.add(match.versionedId());
+        LOG.info("Matched {} in {} to the Controller Service with versioned id {} that the proposed flow declares, based on the {} property of {}",
+            localService, group, match.versionedId(), match.propertyName(), referencer);
+    }
+
+    private List<ControllerServiceNode> orderByReferencerChain(final List<ControllerServiceNode> migrationCreatedServices) {
+        final BitSet visited = new BitSet(migrationCreatedServices.size());
+        final List<ControllerServiceNode> ordered = new ArrayList<>(migrationCreatedServices.size());
+
+        for (int i = 0; i < migrationCreatedServices.size(); i++) {
+            appendReferencerChain(i, migrationCreatedServices, visited, ordered);
+        }
+
+        return ordered;
+    }
+
+    private void appendReferencerChain(
+            final int index,
+            final List<ControllerServiceNode> migrationCreatedServices,
+            final BitSet visited,
+            final List<ControllerServiceNode> ordered
+    ) {
+        if (visited.get(index)) {
+            return;
+        }
+        visited.set(index);
+
+        final ControllerServiceNode localService = migrationCreatedServices.get(index);
+        final ComponentNode referencer = getSoleReferencer(localService);
+        if (referencer instanceof ControllerServiceNode referencingService && isMigrationCreated(referencingService)) {
+            final int referencerIndex = migrationCreatedServices.indexOf(referencingService);
+            if (referencerIndex >= 0) {
+                // Visit the creator first so we can assign the versioned id to the creator first.
+                appendReferencerChain(referencerIndex, migrationCreatedServices, visited, ordered);
+            }
+        }
+
+        ordered.add(localService);
+    }
+
+    private boolean isMigrationCreated(final ControllerServiceNode service) {
+        return StandardControllerServiceFactory.MIGRATION_CREATED_COMMENT.equals(service.getComments());
+    }
+
+    private Map<String, VersionedConfigurableExtension> indexByVersionedId(
+            final Collection<? extends VersionedConfigurableExtension> controllerServices,
+            final Collection<? extends VersionedConfigurableExtension> processors
+    ) {
+        final Map<String, VersionedConfigurableExtension> byVersionedId = HashMap.newHashMap(controllerServices.size() + processors.size());
+        addByVersionedId(byVersionedId, controllerServices);
+        addByVersionedId(byVersionedId, processors);
+        return byVersionedId;
+    }
+
+    private void addByVersionedId(
+            final Map<String, VersionedConfigurableExtension> byVersionedId,
+            final Collection<? extends VersionedConfigurableExtension> components
+    ) {
+        for (final VersionedConfigurableExtension component : components) {
+            byVersionedId.put(component.getIdentifier(), component);
+        }
+    }
+
+    private ComponentNode getSoleReferencer(final ControllerServiceNode localService) {
+        final Set<ComponentNode> referencers = localService.getReferences().getReferencingComponents();
+        if (referencers.size() != 1) {
+            return null;
+        }
+
+        return referencers.iterator().next();

Review Comment:
   This assumes there will never be `0` referencing components, which seems unsafe



-- 
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: issues-unsubscribe@nifi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] NIFI-16310 Preserve Controller Services created by migrateProperties during versioned flow synchronization [nifi]

["exceptionfactory (via GitHub)" <gi...@apache.org>]

exceptionfactory commented on code in PR #11670:
URL: https://github.com/apache/nifi/pull/11670#discussion_r4059056017


##########
nifi-system-tests/nifi-alternate-config-extensions-bundle/nifi-alternate-config-extensions/src/main/java/org/apache/nifi/cs/tests/system/StateBackedStoreService.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * 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.cs.tests.system;
+
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Store whose contents live in the component's local state, so that they are subject to the framework's own state
+ * lifecycle. The store records when it was first created and how many rows it holds.
+ *
+ * Removing a Controller Service clears its state, so a service that was torn down and recreated comes back with a
+ * later creation timestamp and no rows. Comparing the creation timestamp across an operation therefore distinguishes
+ * a preserved service from one that was replaced, even when the replacement carries the same identifier.
+ */
+@Stateful(scopes = Scope.LOCAL, description = "Holds the creation timestamp of the store and the number of rows written to it.")
+public class StateBackedStoreService extends AbstractControllerService implements StoreService {
+
+    static final PropertyDescriptor STORE_NAME = new PropertyDescriptor.Builder()
+            .name("Store Name")
+            .required(true)
+            .defaultValue("store")
+            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+            .build();
+
+    static final String CREATED_KEY = "created";
+    static final String ROW_COUNT_KEY = "rowCount";
+    static final String LAST_ROW_KEY = "lastRow";
+
+    private static final List<PropertyDescriptor> PROPERTIES = List.of(STORE_NAME);
+
+    @Override
+    protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+        return PROPERTIES;
+    }
+
+    /**
+     * Establishes the store on first enable and leaves it alone afterwards, so that the creation timestamp survives
+     * the disable and re-enable cycle that a version change performs.
+     */
+    @OnEnabled
+    public void onEnabled(final ConfigurationContext context) throws IOException {
+        final StateManager stateManager = getStateManager();
+        final StateMap stateMap = stateManager.getState(Scope.LOCAL);
+        if (stateMap.get(CREATED_KEY) != null) {
+            return;
+        }
+
+        final Map<String, String> state = new HashMap<>();
+        state.put(CREATED_KEY, String.valueOf(System.currentTimeMillis()));
+        state.put(ROW_COUNT_KEY, "0");
+        stateManager.setState(state, Scope.LOCAL);
+    }
+
+    @Override
+    public synchronized void append(final String row) {

Review Comment:
   Why is this `synchronized`, is it necessary?



##########
nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/migration/MigrationCreatedControllerServiceVersioningIT.java:
##########
@@ -0,0 +1,365 @@
+/*
+ * 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.migration.StandardControllerServiceFactory;
+import org.apache.nifi.tests.system.AbstractNarSwapMigrationIT;
+import org.apache.nifi.tests.system.ExceptionalBooleanSupplier;
+import org.apache.nifi.toolkit.client.NiFiClientException;
+import org.apache.nifi.web.api.dto.ComponentStateDTO;
+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 java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * Verifies that a Controller Service created by property migration survives the operations a deployed flow goes
+ * through: a runtime upgrade that introduces the migration, a plain runtime restart, and version changes of the
+ * enclosing versioned Process Group.
+ *
+ * Two independent flow lineages are pre-seeded in the registry. In the first, a later version declares the store
+ * Controller Service itself, as a published flow would once the vendor adds it. In the second, no version ever
+ * declares the service, so the flow relies entirely on property migration to create it, and the later version only
+ * adds an unrelated processor.
+ */
+public class MigrationCreatedControllerServiceVersioningIT extends AbstractNarSwapMigrationIT {

Review Comment:
   For a new test class, the `public` modifiers are not needed at the class and method level.



##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java:
##########
@@ -1169,9 +1169,225 @@ private void removeMissingRpg(final ProcessGroup group, final VersionedProcessGr
         removeMissingComponents(group, proposed, rpgsByVersionedId, VersionedProcessGroup::getRemoteProcessGroups, ProcessGroup::removeRemoteProcessGroup);
     }
 
+    /**
+     * Assigns a proposed versioned id to a Controller Service created by property migration.
+     * The service must have exactly one referencer.
+     * The proposed counterpart of that referencer must point at a Controller Service of the same type.
+     * If those do not hold, the service stays unversioned.
+     * A proposed id is assigned to at most one local service.
+     */
+    private void assignVersionedIdsToMigrationCreatedControllerServices(final ProcessGroup group, final VersionedProcessGroup proposed) {
+        final Collection<ControllerServiceNode> groupServices = group.getControllerServices(false);
+        if (groupServices == null || groupServices.isEmpty()) {

Review Comment:
   This method should never return `null`



##########
nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/migration/MigrationCreatedControllerServiceVersioningIT.java:
##########
@@ -0,0 +1,365 @@
+/*
+ * 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.migration.StandardControllerServiceFactory;
+import org.apache.nifi.tests.system.AbstractNarSwapMigrationIT;
+import org.apache.nifi.tests.system.ExceptionalBooleanSupplier;
+import org.apache.nifi.toolkit.client.NiFiClientException;
+import org.apache.nifi.web.api.dto.ComponentStateDTO;
+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 java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * Verifies that a Controller Service created by property migration survives the operations a deployed flow goes
+ * through: a runtime upgrade that introduces the migration, a plain runtime restart, and version changes of the
+ * enclosing versioned Process Group.
+ *
+ * Two independent flow lineages are pre-seeded in the registry. In the first, a later version declares the store
+ * Controller Service itself, as a published flow would once the vendor adds it. In the second, no version ever
+ * declares the service, so the flow relies entirely on property migration to create it, and the later version only
+ * adds an unrelated processor.
+ */
+public class MigrationCreatedControllerServiceVersioningIT extends AbstractNarSwapMigrationIT {
+    private static final String TEST_FLOWS_BUCKET = "test-flows";
+    private static final String SERVICE_DECLARED_FLOW_ID = "11111111-2222-3333-4444-555555555555";
+    private static final String SERVICE_ABSENT_FLOW_ID = "22222222-3333-4444-5555-666666666666";
+    private static final String DECLARED_SERVICE_VERSIONED_ID = "99999999-8888-7777-6666-555555555555";
+    private static final String STORE_SERVICE_PROPERTY = "Store Service";
+    private static final String STORE_SERVICE_TYPE = "org.apache.nifi.cs.tests.system.StateBackedStoreService";
+    private static final String PROCESSOR_TYPE = "org.apache.nifi.processors.tests.system.MigrateToControllerService";
+    private static final String MIGRATING_PROCESSOR_NAME = "MigrateToControllerService";
+    private static final String ADDED_PROCESSOR_NAME = "Added Processor";
+    private static final String VERSIONED_FLOWS_DIRECTORY = "src/test/resources/versioned-flows";
+    private static final String CREATED_STATE_KEY = "created";
+    private static final String ROW_COUNT_STATE_KEY = "rowCount";
+    private static final Duration CONDITION_TIMEOUT = Duration.ofSeconds(30);
+    private static final Duration CONDITION_POLL = Duration.ofMillis(100);
+
+    /**
+     * After the runtime is upgraded, the Controller Service that property migration creates must be present,
+     * enabled and referenced, and the flow that was running before the upgrade must be running again, with no manual action.
+     */
+    @Test
+    public void testRuntimeUpgradeCreatesEnabledServiceAndKeepsFlowRunning() throws NiFiClientException, IOException, InterruptedException {
+        final MigratedFlow flow = importAndUpgradeRuntime(SERVICE_DECLARED_FLOW_ID);
+        final ControllerServiceEntity service = waitForSingleStoreService(flow.groupId());
+        final String serviceId = service.getComponent().getId();
+
+        assertEquals(StandardControllerServiceFactory.MIGRATION_CREATED_COMMENT, service.getComponent().getComments());
+        assertBelongsToLocalFlowOnly(service);
+        getClientUtil().waitForControllerServiceRunStatus(serviceId, "ENABLED");
+        getClientUtil().waitForRunningProcessor(flow.processorId());
+        assertEquals(serviceId, getStoreServiceId(flow.processorId()));
+
+        final Collection<String> validationErrors = getNifiClient().getProcessorClient().getProcessor(flow.processorId()).getComponent().getValidationErrors();
+        final boolean processorValid = validationErrors == null || validationErrors.isEmpty();
+        assertTrue(processorValid, "Processor must be valid after the runtime upgrade");
+
+        waitForCondition(() -> countRows(serviceId) > 0, "store row count > 0");
+
+        final String versionedFlowState = getClientUtil().getVersionedFlowState(flow.groupId(), "root");
+        assertNotEquals("LOCALLY_MODIFIED", versionedFlowState, "The migration-created Controller Service must not make the flow dirty");
+        assertNotEquals("LOCALLY_MODIFIED_AND_STALE", versionedFlowState, "The migration-created Controller Service must not make the flow dirty");
+
+        final boolean serviceReportedAsLocalModification = getNifiClient().getProcessGroupClient().getLocalModifications(flow.groupId())
+                .getComponentDifferences().stream()
+                .anyMatch(diff -> serviceId.equals(diff.getComponentId()));
+        assertFalse(serviceReportedAsLocalModification);
+    }
+
+    /**
+     * Upgrading to a flow version that declares the store Controller Service must keep using the service that
+     * property migration already created, rather than removing it and substituting the one the published version declares.
+     */
+    @Test
+    public void testFlowUpgradePreservesMigrationCreatedControllerService() throws NiFiClientException, IOException, InterruptedException {
+        final MigratedFlow flow = importAndUpgradeRuntime(SERVICE_DECLARED_FLOW_ID);
+        final MigratedStore store = awaitPopulatedStoreService(flow);
+        assertBelongsToLocalFlowOnly(waitForSingleStoreService(flow.groupId()));
+
+        final VersionedFlowUpdateRequestEntity upgradeRequest = getClientUtil().changeFlowVersion(flow.groupId(), "2", false);
+
+        assertStorePreserved(flow, store, "flow upgrade");
+        getClientUtil().assertFlowUpToDate(flow.groupId());
+
+        final ControllerServiceEntity serviceAfterUpgrade = waitForSingleStoreService(flow.groupId());
+        assertEquals(DECLARED_SERVICE_VERSIONED_ID, serviceAfterUpgrade.getComponent().getVersionedComponentId(),
+                "The service must track the Controller Service that version 2 declares, instead of belonging to the local flow only");
+        assertEquals("", serviceAfterUpgrade.getComponent().getComments(),
+                "Comments must come from version 2 once the service tracks the declared Controller Service");
+        assertNull(upgradeRequest.getRequest().getFailureReason(),
+                "The migration-created Controller Service must not prevent the versioned flow from upgrading");
+    }
+
+    /**
+     * A flow whose definition never declares the store Controller Service relies entirely on property migration to
+     * create it. Upgrading such a flow to a version that only adds an unrelated processor, leaving the migrating
+     * processor untouched, must not disturb the service: the version change has nothing to say about it.
+     */
+    @Test
+    public void testFlowUpgradeAddingUnrelatedProcessorPreservesMigrationCreatedControllerService() throws NiFiClientException, IOException, InterruptedException {
+        final MigratedFlow flow = importAndUpgradeRuntime(SERVICE_ABSENT_FLOW_ID);
+        final MigratedStore store = awaitPopulatedStoreService(flow);
+
+        final VersionedFlowUpdateRequestEntity upgradeRequest = getClientUtil().changeFlowVersion(flow.groupId(), "2", false);
+
+        assertTrue(hasProcessorNamed(flow.groupId(), ADDED_PROCESSOR_NAME), "The upgrade must have added the unrelated processor");
+
+        assertStorePreserved(flow, store, "flow upgrade");
+        getClientUtil().assertFlowUpToDate(flow.groupId());
+
+        final ControllerServiceEntity serviceAfterUpgrade = waitForSingleStoreService(flow.groupId());
+        assertBelongsToLocalFlowOnly(serviceAfterUpgrade);
+        assertEquals(StandardControllerServiceFactory.MIGRATION_CREATED_COMMENT, serviceAfterUpgrade.getComponent().getComments());
+        assertNull(upgradeRequest.getRequest().getFailureReason(),
+                "The migration-created Controller Service must not prevent the versioned flow from upgrading");
+    }
+
+    /**
+     * Restarting the runtime with no NAR change must reuse the Controller Service that property migration
+     * created previously rather than creating a second one.
+     */
+    @Test
+    public void testRuntimeRestartDoesNotRecreateMigrationCreatedControllerService() throws NiFiClientException, IOException, InterruptedException {
+        final MigratedFlow flow = importAndUpgradeRuntime(SERVICE_DECLARED_FLOW_ID);
+        final MigratedStore store = awaitPopulatedStoreService(flow);
+
+        getNiFiInstance().stop();
+        getNiFiInstance().start(true);
+
+        assertStorePreserved(flow, store, "runtime restart");
+        assertBelongsToLocalFlowOnly(waitForSingleStoreService(flow.groupId()));
+    }
+
+    /**
+     * Asserts that the given Controller Service has no counterpart in the flow definition. Such a service either
+     * carries no versioned component id at all, or carries the placeholder that the framework derives from its own
+     * instance id while mapping the group. A service that tracks a Controller Service declared by the flow definition
+     * carries the identifier from the definition instead.
+     */
+    private void assertBelongsToLocalFlowOnly(final ControllerServiceEntity service) {
+        final String versionedComponentId = service.getComponent().getVersionedComponentId();
+        if (versionedComponentId == null) {
+            return;
+        }
+
+        final String derivedFromInstanceId = UUID.nameUUIDFromBytes(service.getComponent().getId().getBytes(StandardCharsets.UTF_8)).toString();
+        assertEquals(derivedFromInstanceId, versionedComponentId,
+                "The service tracks a Controller Service declared by the flow definition, so it no longer belongs to the local flow only");
+    }
+
+    /**
+     * Imports version 1 of the given pre-seeded flow, waits for its processor to be running, and then simulates a
+     * runtime upgrade by stopping NiFi, swapping in the alternate-config extensions and starting NiFi again. Nothing
+     * in the flow is stopped or started by hand, so the assertions that follow observe what the upgrade does on its own.
+     */
+    private MigratedFlow importAndUpgradeRuntime(final String flowId) throws NiFiClientException, IOException, InterruptedException {
+        final FlowRegistryClientEntity registryClient = registerClient(new File(VERSIONED_FLOWS_DIRECTORY));
+        final ProcessGroupEntity group = getClientUtil().importFlowFromRegistry("root", registryClient.getId(), TEST_FLOWS_BUCKET, flowId, "1");
+        final ProcessorEntity processor = findProcessor(group.getId(), MIGRATING_PROCESSOR_NAME);
+        getClientUtil().waitForRunningProcessor(processor.getId());
+
+        getNiFiInstance().stop();
+        switchOutNars();
+        getNiFiInstance().start(true);
+
+        return new MigratedFlow(group.getId(), processor.getId());
+    }
+
+    /**
+     * Waits until the migration-created Controller Service is enabled and its store has accumulated at least one row,
+     * so that the flow is known to be working before the operation under test runs, and returns its identifier.
+     */
+    private MigratedStore awaitPopulatedStoreService(final MigratedFlow flow) throws NiFiClientException, IOException, InterruptedException {
+        final ControllerServiceEntity service = waitForSingleStoreService(flow.groupId());
+        final String serviceId = service.getComponent().getId();
+        getClientUtil().waitForControllerServiceRunStatus(serviceId, "ENABLED");
+        waitForCondition(() -> countRows(serviceId) > 0, "store row count > 0");
+
+        return new MigratedStore(serviceId, readCreatedTimestamp(flow, serviceId));
+    }
+
+    /**
+     * Asserts that the given operation left the migration-created Controller Service in place and the flow working.
+     *
+     * Two independent properties are checked. An unchanged creation timestamp proves the service was never torn down,
+     * since removing a Controller Service clears its state and a replacement records a new timestamp. A row count that
+     * keeps climbing afterwards proves the flow is doing work again, which a RUNNING state alone does not show: a
+     * processor whose Controller Service reference is broken still reports RUNNING while writing nothing.
+     *
+     * The store is read through the service the processor references now rather than the one recorded earlier, so
+     * that a service substituted by the operation is detected rather than silently passing.
+     */
+    private void assertStorePreserved(final MigratedFlow flow, final MigratedStore store, final String operation)
+            throws NiFiClientException, IOException, InterruptedException {
+
+        waitForCondition(() -> !findStoreServices(flow.groupId()).isEmpty(), "store Controller Service found");
+
+        final List<ControllerServiceEntity> servicesAfter = findControllerServicesInGroup(flow.groupId());
+        assertEquals(1, servicesAfter.size(),
+                "The " + operation + " left " + servicesAfter.size() + " Controller Services in the group, so one was created alongside the migration-created one");
+
+        final ControllerServiceEntity serviceAfter = servicesAfter.getFirst();
+        assertEquals(store.serviceId(), serviceAfter.getComponent().getId(),
+                "The " + operation + " replaced the migration-created Controller Service with a different one");
+        assertEquals(store.serviceId(), getStoreServiceId(flow.processorId()));
+
+        getClientUtil().waitForControllerServiceRunStatus(store.serviceId(), "ENABLED");
+        getClientUtil().waitForRunningProcessor(flow.processorId());
+
+        assertEquals(store.created(), readCreatedTimestamp(flow, store.serviceId()),
+                "The migration-created Controller Service was removed during the " + operation + ": its store was created again, so the state it held is gone");
+
+        final long rowsAfterOperation = countRows(getStoreServiceId(flow.processorId()));
+        waitForCondition(() -> countRows(getStoreServiceId(flow.processorId())) > rowsAfterOperation, "store row count increase");
+    }
+
+
+    private ProcessorEntity findProcessor(final String groupId, final String name) throws NiFiClientException, IOException {
+        final List<ProcessorEntity> matching = getNifiClient().getFlowClient().getProcessGroup(groupId).getProcessGroupFlow().getFlow().getProcessors().stream()
+                .filter(processor -> PROCESSOR_TYPE.equals(processor.getComponent().getType()))
+                .filter(processor -> name.equals(processor.getComponent().getName()))
+                .toList();
+
+        if (matching.size() != 1) {
+            throw new AssertionError("Expected exactly one processor named " + name + " in group " + groupId + " but found " + matching.size());
+        }
+
+        return matching.getFirst();
+    }
+
+    private boolean hasProcessorNamed(final String groupId, final String name) throws NiFiClientException, IOException {
+        return getNifiClient().getFlowClient().getProcessGroup(groupId).getProcessGroupFlow().getFlow().getProcessors().stream()
+                .anyMatch(processor -> name.equals(processor.getComponent().getName()));
+    }
+
+    private List<ControllerServiceEntity> findControllerServicesInGroup(final String groupId) throws NiFiClientException, IOException {
+        return getNifiClient().getFlowClient().getControllerServices(groupId).getControllerServices().stream()
+                .filter(service -> groupId.equals(service.getComponent().getParentGroupId()))
+                .toList();
+    }
+
+    private List<ControllerServiceEntity> findStoreServices(final String groupId) throws NiFiClientException, IOException {
+        return findControllerServicesInGroup(groupId).stream()
+                .filter(service -> STORE_SERVICE_TYPE.equals(service.getComponent().getType()))
+                .toList();
+    }
+
+    private ControllerServiceEntity waitForSingleStoreService(final String groupId) throws NiFiClientException, IOException, InterruptedException {
+        waitForCondition(() -> findStoreServices(groupId).size() == 1, "exactly one store Controller Service");
+        return findStoreServices(groupId).getFirst();
+    }
+
+    private String getStoreServiceId(final String processorId) throws NiFiClientException, IOException {
+        final Map<String, String> properties = getNifiClient().getProcessorClient().getProcessor(processorId).getComponent().getConfig().getProperties();
+        return properties.get(STORE_SERVICE_PROPERTY);
+    }
+
+    /**
+     * Reads the store's local component state. Removing a Controller Service clears its state, so an empty map means
+     * either that the service has not been enabled yet or that it was torn down.
+     */
+    private Map<String, String> readStoreState(final String serviceId) throws NiFiClientException, IOException {
+        final ComponentStateEntity stateEntity = getNifiClient().getControllerServicesClient().getControllerServiceState(serviceId);
+        final ComponentStateDTO componentState = stateEntity.getComponentState();
+        if (componentState == null || componentState.getLocalState() == null || componentState.getLocalState().getState() == null) {
+            return Map.of();
+        }
+
+        final Map<String, String> state = new HashMap<>();
+        for (final StateEntryDTO entry : componentState.getLocalState().getState()) {
+            state.put(entry.getKey(), entry.getValue());
+        }
+
+        return state;
+    }
+
+    /**
+     * Returns the timestamp recorded when the store was established, or null if the service is no longer in the group.
+     * A service that has been removed outright has no state to read, and reporting that as a missing timestamp lets the
+     * caller fail on the comparison rather than on a lookup error.
+     */
+    private String readCreatedTimestamp(final MigratedFlow flow, final String serviceId) throws NiFiClientException, IOException {
+        final boolean stillPresent = findStoreServices(flow.groupId()).stream()
+                .anyMatch(service -> serviceId.equals(service.getComponent().getId()));
+        if (!stillPresent) {
+            return null;
+        }
+
+        return readStoreState(serviceId).get(CREATED_STATE_KEY);
+    }
+
+    private long countRows(final String serviceId) throws NiFiClientException, IOException {
+        final String rowCount = readStoreState(serviceId).get(ROW_COUNT_STATE_KEY);
+        return rowCount == null ? 0 : Long.parseLong(rowCount);
+    }
+
+    /**
+     * Polls the given condition until it holds, failing the test once the timeout elapses. Conditions query a NiFi that
+     * may still be starting up or replacing components, so a failing query is treated as the condition not holding yet.
+     */
+    private void waitForCondition(final ExceptionalBooleanSupplier condition, final String conditionDescription) throws InterruptedException {

Review Comment:
   This method appears to be duplicative of the `waitFor()` method in the base class.



##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizerTest.java:
##########
@@ -552,6 +528,284 @@ public void testAddProcessorWithServiceAndMigration() {
         assertEquals(controllerServiceNode.getIdentifier(), migratedProperties.get("cs"));
     }
 
+    @Test
+    public void testUserAddedControllerServiceRemovedWhenAbsentFromProposedFlow() {
+        final ProcessGroup processGroup = createMockProcessGroup();
+
+        final PropertyDescriptor descriptor = new PropertyDescriptor.Builder().name("abc").build();
+        final ControllerServiceNode serviceNode = createMockControllerService();
+        when(serviceNode.getComments()).thenReturn("Added by a user");
+        when(serviceNode.getName()).thenReturn("name");
+        when(serviceNode.getCanonicalClassName()).thenReturn("ControllerServiceImpl");
+        when(serviceNode.getProperties()).thenReturn(Map.of(descriptor, new PropertyConfiguration("123", null, null, null)));
+        when(serviceNode.getRawPropertyValues()).thenReturn(Map.of(descriptor, "123"));
+        when(serviceNode.getVersionedComponentId()).thenReturn(Optional.empty());

Review Comment:
   Repeated literals in multiple methods should be declared as static final variables and reused.



##########
nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/migration/MigrationCreatedControllerServiceVersioningIT.java:
##########
@@ -0,0 +1,365 @@
+/*
+ * 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.migration.StandardControllerServiceFactory;
+import org.apache.nifi.tests.system.AbstractNarSwapMigrationIT;
+import org.apache.nifi.tests.system.ExceptionalBooleanSupplier;
+import org.apache.nifi.toolkit.client.NiFiClientException;
+import org.apache.nifi.web.api.dto.ComponentStateDTO;
+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 java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * Verifies that a Controller Service created by property migration survives the operations a deployed flow goes
+ * through: a runtime upgrade that introduces the migration, a plain runtime restart, and version changes of the
+ * enclosing versioned Process Group.
+ *
+ * Two independent flow lineages are pre-seeded in the registry. In the first, a later version declares the store
+ * Controller Service itself, as a published flow would once the vendor adds it. In the second, no version ever
+ * declares the service, so the flow relies entirely on property migration to create it, and the later version only
+ * adds an unrelated processor.
+ */
+public class MigrationCreatedControllerServiceVersioningIT extends AbstractNarSwapMigrationIT {
+    private static final String TEST_FLOWS_BUCKET = "test-flows";
+    private static final String SERVICE_DECLARED_FLOW_ID = "11111111-2222-3333-4444-555555555555";
+    private static final String SERVICE_ABSENT_FLOW_ID = "22222222-3333-4444-5555-666666666666";
+    private static final String DECLARED_SERVICE_VERSIONED_ID = "99999999-8888-7777-6666-555555555555";
+    private static final String STORE_SERVICE_PROPERTY = "Store Service";
+    private static final String STORE_SERVICE_TYPE = "org.apache.nifi.cs.tests.system.StateBackedStoreService";
+    private static final String PROCESSOR_TYPE = "org.apache.nifi.processors.tests.system.MigrateToControllerService";
+    private static final String MIGRATING_PROCESSOR_NAME = "MigrateToControllerService";
+    private static final String ADDED_PROCESSOR_NAME = "Added Processor";
+    private static final String VERSIONED_FLOWS_DIRECTORY = "src/test/resources/versioned-flows";
+    private static final String CREATED_STATE_KEY = "created";
+    private static final String ROW_COUNT_STATE_KEY = "rowCount";
+    private static final Duration CONDITION_TIMEOUT = Duration.ofSeconds(30);
+    private static final Duration CONDITION_POLL = Duration.ofMillis(100);
+
+    /**
+     * After the runtime is upgraded, the Controller Service that property migration creates must be present,
+     * enabled and referenced, and the flow that was running before the upgrade must be running again, with no manual action.
+     */
+    @Test
+    public void testRuntimeUpgradeCreatesEnabledServiceAndKeepsFlowRunning() throws NiFiClientException, IOException, InterruptedException {
+        final MigratedFlow flow = importAndUpgradeRuntime(SERVICE_DECLARED_FLOW_ID);
+        final ControllerServiceEntity service = waitForSingleStoreService(flow.groupId());
+        final String serviceId = service.getComponent().getId();
+
+        assertEquals(StandardControllerServiceFactory.MIGRATION_CREATED_COMMENT, service.getComponent().getComments());
+        assertBelongsToLocalFlowOnly(service);
+        getClientUtil().waitForControllerServiceRunStatus(serviceId, "ENABLED");
+        getClientUtil().waitForRunningProcessor(flow.processorId());
+        assertEquals(serviceId, getStoreServiceId(flow.processorId()));
+
+        final Collection<String> validationErrors = getNifiClient().getProcessorClient().getProcessor(flow.processorId()).getComponent().getValidationErrors();
+        final boolean processorValid = validationErrors == null || validationErrors.isEmpty();
+        assertTrue(processorValid, "Processor must be valid after the runtime upgrade");
+
+        waitForCondition(() -> countRows(serviceId) > 0, "store row count > 0");
+
+        final String versionedFlowState = getClientUtil().getVersionedFlowState(flow.groupId(), "root");
+        assertNotEquals("LOCALLY_MODIFIED", versionedFlowState, "The migration-created Controller Service must not make the flow dirty");
+        assertNotEquals("LOCALLY_MODIFIED_AND_STALE", versionedFlowState, "The migration-created Controller Service must not make the flow dirty");
+
+        final boolean serviceReportedAsLocalModification = getNifiClient().getProcessGroupClient().getLocalModifications(flow.groupId())
+                .getComponentDifferences().stream()
+                .anyMatch(diff -> serviceId.equals(diff.getComponentId()));
+        assertFalse(serviceReportedAsLocalModification);
+    }
+
+    /**
+     * Upgrading to a flow version that declares the store Controller Service must keep using the service that
+     * property migration already created, rather than removing it and substituting the one the published version declares.
+     */
+    @Test
+    public void testFlowUpgradePreservesMigrationCreatedControllerService() throws NiFiClientException, IOException, InterruptedException {
+        final MigratedFlow flow = importAndUpgradeRuntime(SERVICE_DECLARED_FLOW_ID);
+        final MigratedStore store = awaitPopulatedStoreService(flow);
+        assertBelongsToLocalFlowOnly(waitForSingleStoreService(flow.groupId()));
+
+        final VersionedFlowUpdateRequestEntity upgradeRequest = getClientUtil().changeFlowVersion(flow.groupId(), "2", false);
+
+        assertStorePreserved(flow, store, "flow upgrade");
+        getClientUtil().assertFlowUpToDate(flow.groupId());
+
+        final ControllerServiceEntity serviceAfterUpgrade = waitForSingleStoreService(flow.groupId());
+        assertEquals(DECLARED_SERVICE_VERSIONED_ID, serviceAfterUpgrade.getComponent().getVersionedComponentId(),

Review Comment:
   Many of these assertion methods are more verbose than necessary and should be shortened.



##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java:
##########
@@ -1169,9 +1169,225 @@ private void removeMissingRpg(final ProcessGroup group, final VersionedProcessGr
         removeMissingComponents(group, proposed, rpgsByVersionedId, VersionedProcessGroup::getRemoteProcessGroups, ProcessGroup::removeRemoteProcessGroup);
     }
 
+    /**
+     * Assigns a proposed versioned id to a Controller Service created by property migration.
+     * The service must have exactly one referencer.
+     * The proposed counterpart of that referencer must point at a Controller Service of the same type.
+     * If those do not hold, the service stays unversioned.
+     * A proposed id is assigned to at most one local service.
+     */
+    private void assignVersionedIdsToMigrationCreatedControllerServices(final ProcessGroup group, final VersionedProcessGroup proposed) {
+        final Collection<ControllerServiceNode> groupServices = group.getControllerServices(false);
+        if (groupServices == null || groupServices.isEmpty()) {
+            return;
+        }
+
+        final List<ControllerServiceNode> migrationCreatedServices = new ArrayList<>();
+        for (final ControllerServiceNode localService : groupServices) {
+            if (localService.getVersionedComponentId().isEmpty() && isMigrationCreated(localService)) {
+                migrationCreatedServices.add(localService);
+            }
+        }
+
+        if (migrationCreatedServices.isEmpty()) {
+            return;
+        }
+
+        final Set<String> claimedVersionedIds = HashSet.newHashSet(groupServices.size());
+        for (final ControllerServiceNode localService : groupServices) {
+            localService.getVersionedComponentId().ifPresent(claimedVersionedIds::add);
+        }
+
+        final Map<String, VersionedConfigurableExtension> proposedComponentsByVersionedId = indexByVersionedId(proposed.getControllerServices(), proposed.getProcessors());

Review Comment:
   This method is on the longer side, it looks an opportunity to break it up into two methods after the initial checks.



##########
nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/MigrateToControllerService.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.components.PropertyDescriptor;
+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 org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Pre-upgrade shape of a processor that keeps its store location in a plain property.
+ * The post-upgrade shape of the same processor, in the alternate-config extensions bundle,
+ * migrates that property into a Controller Service.
+ */

Review Comment:
   It would be better to use this as the Capability Description annotation, instead of the class comment for semantic definition.



-- 
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: issues-unsubscribe@nifi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] NIFI-16310 Preserve Controller Services created by migrateProperties during versioned flow synchronization [nifi]

["awelless (via GitHub)" <gi...@apache.org>]

awelless commented on PR #11670:
URL: https://github.com/apache/nifi/pull/11670#issuecomment-6147353338

   Thanks for the feedback, @markap14.
   I think it's fine to support only flow versions created after the fix.
   
   Comments regarding the suggested approach.
   1. I'm going to use process group version id instead of creator id. In that case if the same service is attempted to be created from 2 creators, it always gets the same versioned id, no matter what creator creates it.
   2. Before controller service is committed into a newer flow version, we should remove `MIGRATION_CREATED_COMMENT` to treat it normally from now on. In this PR it has to be done manually, I'm considering a follow up where this comment is removed automatically.
   
   I'm going to follow the suggested approach.


-- 
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: issues-unsubscribe@nifi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] NIFI-16310 Preserve Controller Services created by migrateProperties during versioned flow synchronization [nifi]

["awelless (via GitHub)" <gi...@apache.org>]

awelless commented on PR #11670:
URL: https://github.com/apache/nifi/pull/11670#issuecomment-5891924380

   Thanks for review, @exceptionfactory.
   I addressed all of your comments. Could you give this PR another look?


-- 
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: issues-unsubscribe@nifi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] NIFI-16310 Preserve Controller Services created by migrateProperties during versioned flow synchronization [nifi]

["awelless (via GitHub)" <gi...@apache.org>]

awelless commented on PR #11670:
URL: https://github.com/apache/nifi/pull/11670#issuecomment-6217927029

   @markap14 @exceptionfactory 
   Thanks for the suggestions. 
   
   I decided to proceed with the approach suggested by Mark's. Therefore I’m closing this PR in favor of #11813. Let's continue the discussion there.
   
   I addressed your comments in that PR.


-- 
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: issues-unsubscribe@nifi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] NIFI-16310 Preserve Controller Services created by migrateProperties during versioned flow synchronization [nifi]

["markap14 (via GitHub)" <gi...@apache.org>]

markap14 commented on PR #11670:
URL: https://github.com/apache/nifi/pull/11670#issuecomment-6001979759

   [claude-opus 5.5]
   
   Thanks for digging into this, @awelless. The Jira lays out the problem clearly, and the system test reproduces it well. I'd like to suggest a much simpler direction for the fix, though. The current approach adds about 235 lines of matching logic to `StandardVersionedComponentSynchronizer`, which is already one of the most complex classes in the framework. I think the same result can be reached with about a dozen lines of logic spread across two existing methods.
   
   ## Why the service gets replaced
   
   When a new flow version is applied, the synchronizer matches local components to the components in that version purely by Versioned Component ID. A service created by `migrateProperties` has no Versioned Component ID, so the synchronizer falls back to an ID generated from the service's instance ID. The instance ID is built from the Process Group's instance ID, and that differs between the NiFi instance where the flow author committed the flow and the NiFi instance that later imports it. The two IDs never match, so the synchronizer treats them as different services. It creates the declared service from scratch, points the processor at it, and removes the local service along with its state.
   
   ## Suggested approach
   
   **1. Give migration-created services a deterministic Versioned Component ID when they are created.**
   
   In `StandardControllerServiceFactory.create`, set the Versioned Component ID from the creator's Versioned Component ID, the implementation class name, and the sorted property values. The creator's Versioned Component ID comes from the versioned flow itself, so it is the same on every NiFi instance that runs the flow, not just on every node of one cluster:
   
   ```java
   serviceNode.setProperties(creationDetails.serviceProperties());
   
   // The Versioned Component ID must be set before the service's own migration so that any service it creates can derive its ID from this one.
   determineVersionedServiceId(creationDetails.type(), creationDetails.serviceProperties()).ifPresent(serviceNode::setVersionedComponentId);
   ```
   
   ```java
   private Optional<String> determineVersionedServiceId(final String className, final Map<String, String> propertyValues) {
       final Optional<String> creatorVersionedId = switch (creator) {
           case final ProcessorNode processor -> processor.getVersionedComponentId();
           case final ControllerServiceNode service -> service.getVersionedComponentId();
           default -> Optional.empty();
       };
   
       final SortedMap<String, String> sortedProperties = new TreeMap<>(propertyValues);
       return creatorVersionedId.map(versionedId -> {
           final String componentDescription = versionedId + className + sortedProperties;
           return UUID.nameUUIDFromBytes(componentDescription.getBytes(StandardCharsets.UTF_8)).toString();
       });
   }
   ```
   
   Here is what then happens in the scenario from the Jira:
   
   1. The flow author upgrades their NiFi. Migration creates the service with the deterministic ID.
   2. The author commits the next version. The flow mapper keeps an existing Versioned Component ID, so the registry stores the service under that same ID.
   3. Another NiFi instance imports the flow and upgrades its NARs. Migration creates the service with the same ID.
   4. That instance upgrades to the new version. The synchronizer finds a service with a matching ID and updates it in place. The service keeps its state, and the group ends up with one service, not two.
   
   The existing synchronizer logic does all the matching. There is no need to look up the referencing component, compare property names, order chains of services, or pick a "first" match when there are several candidates.
   
   **2. Don't remove migration-created services that the proposed flow does not declare.**
   
   This handles the second case in the Jira, where the new version only adds an unrelated processor and never mentions the service:
   
   ```java
   private void removeMissingControllerServices(final ProcessGroup group, final VersionedProcessGroup proposed, final Map<String, ControllerServiceNode> servicesByVersionedId) {
       // Controller Services created by property migration are not removed when the proposed flow does not declare them. A versioned flow
       // may never declare such a service while still containing the component that references it, and removing the service would discard its state.
       final Map<String, ControllerServiceNode> removableServicesByVersionedId = new HashMap<>();
       for (final Map.Entry<String, ControllerServiceNode> entry : servicesByVersionedId.entrySet()) {
           if (!StandardControllerServiceFactory.MIGRATION_CREATED_COMMENT.equals(entry.getValue().getComments())) {
               removableServicesByVersionedId.put(entry.getKey(), entry.getValue());
           }
       }
   
       final BiConsumer<ProcessGroup, ControllerServiceNode> componentRemoval = (grp, service) -> context.getControllerServiceProvider().removeControllerService(service);
       removeMissingComponents(group, proposed, removableServicesByVersionedId, VersionedProcessGroup::getControllerServices, componentRemoval);
   }
   ```
   
   ## Why I'd prefer this
   
   - **It is much smaller.** The change is about a dozen lines of logic in two existing methods. Nearly all of the new code in `StandardVersionedComponentSynchronizer` goes away.
   - **It reuses the matching NiFi already does.** Matching by Versioned Component ID is how every other component type works, so migration-created services stop being a special case during updates.
   - **Every node in a cluster reaches the same result.** In the current approach, the tie-break for several local services matching one declared service depends on iteration order over `ProcessGroup.getControllerServices(false)`, which returns a `HashSet`. Different nodes can attach the versioned ID to different local services. The deterministic ID removes that tie-break.
   - **There are fewer edge cases.** The Jira lists a long set of edge cases, mostly about the matching heuristics. With this approach, either the IDs match and the service is updated in place, or they don't and the service is kept.
   
   One limitation to note: flow versions that are already in a registry won't benefit. A migration-created service committed before this change has an ID derived from the author's Process Group instance ID, which no other instance can reproduce. Only versions committed after this change get the in-place update. I think that trade-off is well worth the reduction in complexity.
   
   ## Tests
   
   - **System test fixture.** The version 2 snapshot hard-codes `99999999-8888-7777-6666-555555555555` as the declared service's ID. No NiFi instance would ever produce that ID, so the test checks a scenario that can't happen in practice. It would be better for the test to create version 2 the way a real flow author would: upgrade the NARs, let migration create the service, and commit that version to the registry. Then import it into a fresh group and upgrade that group. If that's not practical, the fixture should at least use the ID that the deterministic algorithm produces.
   - **`assertBelongsToLocalFlowOnly`** asserts that the Versioned Component ID is derived from the instance ID. That will no longer hold, and it ties the test to how the ID is calculated rather than to the behavior we care about. The behavior to check is that the service survives with its state and that the group ends up with exactly one service, which `assertStorePreserved` already covers.
   - **`testFlowUpgradePreservesMigrationCreatedControllerService`** asserts that the comments are empty after the upgrade. That reflects the hand-written fixture rather than anything the framework guarantees, so I'd drop that assertion.
   - **`StandardVersionedComponentSynchronizerTest`.** Most of the roughly 300 new lines test the matching heuristics, which would no longer exist. I'd replace them with an update to an existing test in `TestStandardControllerServiceFactory` that confirms the Versioned Component ID is set and comes out the same for the same creator, class, and properties. I'd also add a single synchronizer test confirming that a migration-created service the proposed flow doesn't declare is not removed.
   - Moving `FileSystemFlowRegistryClient` into its own NAR so it is available on both sides of the NAR swap makes sense, and I'd keep it.
   
   I have a local version of the two production changes above if it helps. I'm happy to look again once the PR moves in this direction.
   


-- 
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: issues-unsubscribe@nifi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] NIFI-16310 Preserve Controller Services created by migrateProperties during versioned flow synchronization [nifi]

["awelless (via GitHub)" <gi...@apache.org>]

awelless commented on code in PR #11670:
URL: https://github.com/apache/nifi/pull/11670#discussion_r4072626630


##########
nifi-system-tests/nifi-alternate-config-extensions-bundle/nifi-alternate-config-extensions/src/main/java/org/apache/nifi/cs/tests/system/StateBackedStoreService.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * 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.cs.tests.system;
+
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Store whose contents live in the component's local state, so that they are subject to the framework's own state
+ * lifecycle. The store records when it was first created and how many rows it holds.
+ *
+ * Removing a Controller Service clears its state, so a service that was torn down and recreated comes back with a
+ * later creation timestamp and no rows. Comparing the creation timestamp across an operation therefore distinguishes
+ * a preserved service from one that was replaced, even when the replacement carries the same identifier.
+ */
+@Stateful(scopes = Scope.LOCAL, description = "Holds the creation timestamp of the store and the number of rows written to it.")
+public class StateBackedStoreService extends AbstractControllerService implements StoreService {
+
+    static final PropertyDescriptor STORE_NAME = new PropertyDescriptor.Builder()
+            .name("Store Name")
+            .required(true)
+            .defaultValue("store")
+            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+            .build();
+
+    static final String CREATED_KEY = "created";
+    static final String ROW_COUNT_KEY = "rowCount";
+    static final String LAST_ROW_KEY = "lastRow";
+
+    private static final List<PropertyDescriptor> PROPERTIES = List.of(STORE_NAME);
+
+    @Override
+    protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+        return PROPERTIES;
+    }
+
+    /**
+     * Establishes the store on first enable and leaves it alone afterwards, so that the creation timestamp survives
+     * the disable and re-enable cycle that a version change performs.
+     */
+    @OnEnabled
+    public void onEnabled(final ConfigurationContext context) throws IOException {
+        final StateManager stateManager = getStateManager();
+        final StateMap stateMap = stateManager.getState(Scope.LOCAL);
+        if (stateMap.get(CREATED_KEY) != null) {
+            return;
+        }
+
+        final Map<String, String> state = new HashMap<>();
+        state.put(CREATED_KEY, String.valueOf(System.currentTimeMillis()));
+        state.put(ROW_COUNT_KEY, "0");
+        stateManager.setState(state, Scope.LOCAL);
+    }
+
+    @Override
+    public synchronized void append(final String row) {

Review Comment:
   It was synchronized to ensure we don't have any race condition when incrementing `ROW_COUNT_KEY`.
   Provided this is a test-only CS, and we don't access it concurrently in tests, I removed this modifier. 



##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizerTest.java:
##########
@@ -552,6 +528,284 @@ public void testAddProcessorWithServiceAndMigration() {
         assertEquals(controllerServiceNode.getIdentifier(), migratedProperties.get("cs"));
     }
 
+    @Test
+    public void testUserAddedControllerServiceRemovedWhenAbsentFromProposedFlow() {
+        final ProcessGroup processGroup = createMockProcessGroup();
+
+        final PropertyDescriptor descriptor = new PropertyDescriptor.Builder().name("abc").build();
+        final ControllerServiceNode serviceNode = createMockControllerService();
+        when(serviceNode.getComments()).thenReturn("Added by a user");
+        when(serviceNode.getName()).thenReturn("name");
+        when(serviceNode.getCanonicalClassName()).thenReturn("ControllerServiceImpl");
+        when(serviceNode.getProperties()).thenReturn(Map.of(descriptor, new PropertyConfiguration("123", null, null, null)));
+        when(serviceNode.getRawPropertyValues()).thenReturn(Map.of(descriptor, "123"));
+        when(serviceNode.getVersionedComponentId()).thenReturn(Optional.empty());

Review Comment:
   I replaced literals with the existing constants.
   I also introduced constants in the nested `MigrationCreatedControllerService` test



##########
nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/migration/MigrationCreatedControllerServiceVersioningIT.java:
##########
@@ -0,0 +1,365 @@
+/*
+ * 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.migration.StandardControllerServiceFactory;
+import org.apache.nifi.tests.system.AbstractNarSwapMigrationIT;
+import org.apache.nifi.tests.system.ExceptionalBooleanSupplier;
+import org.apache.nifi.toolkit.client.NiFiClientException;
+import org.apache.nifi.web.api.dto.ComponentStateDTO;
+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 java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * Verifies that a Controller Service created by property migration survives the operations a deployed flow goes
+ * through: a runtime upgrade that introduces the migration, a plain runtime restart, and version changes of the
+ * enclosing versioned Process Group.
+ *
+ * Two independent flow lineages are pre-seeded in the registry. In the first, a later version declares the store
+ * Controller Service itself, as a published flow would once the vendor adds it. In the second, no version ever
+ * declares the service, so the flow relies entirely on property migration to create it, and the later version only
+ * adds an unrelated processor.
+ */
+public class MigrationCreatedControllerServiceVersioningIT extends AbstractNarSwapMigrationIT {
+    private static final String TEST_FLOWS_BUCKET = "test-flows";
+    private static final String SERVICE_DECLARED_FLOW_ID = "11111111-2222-3333-4444-555555555555";
+    private static final String SERVICE_ABSENT_FLOW_ID = "22222222-3333-4444-5555-666666666666";
+    private static final String DECLARED_SERVICE_VERSIONED_ID = "99999999-8888-7777-6666-555555555555";
+    private static final String STORE_SERVICE_PROPERTY = "Store Service";
+    private static final String STORE_SERVICE_TYPE = "org.apache.nifi.cs.tests.system.StateBackedStoreService";
+    private static final String PROCESSOR_TYPE = "org.apache.nifi.processors.tests.system.MigrateToControllerService";
+    private static final String MIGRATING_PROCESSOR_NAME = "MigrateToControllerService";
+    private static final String ADDED_PROCESSOR_NAME = "Added Processor";
+    private static final String VERSIONED_FLOWS_DIRECTORY = "src/test/resources/versioned-flows";
+    private static final String CREATED_STATE_KEY = "created";
+    private static final String ROW_COUNT_STATE_KEY = "rowCount";
+    private static final Duration CONDITION_TIMEOUT = Duration.ofSeconds(30);
+    private static final Duration CONDITION_POLL = Duration.ofMillis(100);
+
+    /**
+     * After the runtime is upgraded, the Controller Service that property migration creates must be present,
+     * enabled and referenced, and the flow that was running before the upgrade must be running again, with no manual action.
+     */
+    @Test
+    public void testRuntimeUpgradeCreatesEnabledServiceAndKeepsFlowRunning() throws NiFiClientException, IOException, InterruptedException {
+        final MigratedFlow flow = importAndUpgradeRuntime(SERVICE_DECLARED_FLOW_ID);
+        final ControllerServiceEntity service = waitForSingleStoreService(flow.groupId());
+        final String serviceId = service.getComponent().getId();
+
+        assertEquals(StandardControllerServiceFactory.MIGRATION_CREATED_COMMENT, service.getComponent().getComments());
+        assertBelongsToLocalFlowOnly(service);
+        getClientUtil().waitForControllerServiceRunStatus(serviceId, "ENABLED");
+        getClientUtil().waitForRunningProcessor(flow.processorId());
+        assertEquals(serviceId, getStoreServiceId(flow.processorId()));
+
+        final Collection<String> validationErrors = getNifiClient().getProcessorClient().getProcessor(flow.processorId()).getComponent().getValidationErrors();
+        final boolean processorValid = validationErrors == null || validationErrors.isEmpty();
+        assertTrue(processorValid, "Processor must be valid after the runtime upgrade");
+
+        waitForCondition(() -> countRows(serviceId) > 0, "store row count > 0");
+
+        final String versionedFlowState = getClientUtil().getVersionedFlowState(flow.groupId(), "root");
+        assertNotEquals("LOCALLY_MODIFIED", versionedFlowState, "The migration-created Controller Service must not make the flow dirty");
+        assertNotEquals("LOCALLY_MODIFIED_AND_STALE", versionedFlowState, "The migration-created Controller Service must not make the flow dirty");
+
+        final boolean serviceReportedAsLocalModification = getNifiClient().getProcessGroupClient().getLocalModifications(flow.groupId())
+                .getComponentDifferences().stream()
+                .anyMatch(diff -> serviceId.equals(diff.getComponentId()));
+        assertFalse(serviceReportedAsLocalModification);
+    }
+
+    /**
+     * Upgrading to a flow version that declares the store Controller Service must keep using the service that
+     * property migration already created, rather than removing it and substituting the one the published version declares.
+     */
+    @Test
+    public void testFlowUpgradePreservesMigrationCreatedControllerService() throws NiFiClientException, IOException, InterruptedException {
+        final MigratedFlow flow = importAndUpgradeRuntime(SERVICE_DECLARED_FLOW_ID);
+        final MigratedStore store = awaitPopulatedStoreService(flow);
+        assertBelongsToLocalFlowOnly(waitForSingleStoreService(flow.groupId()));
+
+        final VersionedFlowUpdateRequestEntity upgradeRequest = getClientUtil().changeFlowVersion(flow.groupId(), "2", false);
+
+        assertStorePreserved(flow, store, "flow upgrade");
+        getClientUtil().assertFlowUpToDate(flow.groupId());
+
+        final ControllerServiceEntity serviceAfterUpgrade = waitForSingleStoreService(flow.groupId());
+        assertEquals(DECLARED_SERVICE_VERSIONED_ID, serviceAfterUpgrade.getComponent().getVersionedComponentId(),

Review Comment:
   I cleaned up the assertions descriptions. I still left some of them to provide more details on what a failed assertions means



##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java:
##########
@@ -1169,9 +1169,225 @@ private void removeMissingRpg(final ProcessGroup group, final VersionedProcessGr
         removeMissingComponents(group, proposed, rpgsByVersionedId, VersionedProcessGroup::getRemoteProcessGroups, ProcessGroup::removeRemoteProcessGroup);
     }
 
+    /**
+     * Assigns a proposed versioned id to a Controller Service created by property migration.
+     * The service must have exactly one referencer.
+     * The proposed counterpart of that referencer must point at a Controller Service of the same type.
+     * If those do not hold, the service stays unversioned.
+     * A proposed id is assigned to at most one local service.
+     */
+    private void assignVersionedIdsToMigrationCreatedControllerServices(final ProcessGroup group, final VersionedProcessGroup proposed) {
+        final Collection<ControllerServiceNode> groupServices = group.getControllerServices(false);
+        if (groupServices == null || groupServices.isEmpty()) {
+            return;
+        }
+
+        final List<ControllerServiceNode> migrationCreatedServices = new ArrayList<>();
+        for (final ControllerServiceNode localService : groupServices) {
+            if (localService.getVersionedComponentId().isEmpty() && isMigrationCreated(localService)) {
+                migrationCreatedServices.add(localService);
+            }
+        }
+
+        if (migrationCreatedServices.isEmpty()) {
+            return;
+        }
+
+        final Set<String> claimedVersionedIds = HashSet.newHashSet(groupServices.size());
+        for (final ControllerServiceNode localService : groupServices) {
+            localService.getVersionedComponentId().ifPresent(claimedVersionedIds::add);
+        }
+
+        final Map<String, VersionedConfigurableExtension> proposedComponentsByVersionedId = indexByVersionedId(proposed.getControllerServices(), proposed.getProcessors());

Review Comment:
   I extracted longer parts into separate methods. Listing `migrationCreatedServices`, `claimedVersionedIds` as well as version id assignment for `migrationCreatedServices`. Looks more readable now



##########
nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/migration/MigrationCreatedControllerServiceVersioningIT.java:
##########
@@ -0,0 +1,365 @@
+/*
+ * 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.migration.StandardControllerServiceFactory;
+import org.apache.nifi.tests.system.AbstractNarSwapMigrationIT;
+import org.apache.nifi.tests.system.ExceptionalBooleanSupplier;
+import org.apache.nifi.toolkit.client.NiFiClientException;
+import org.apache.nifi.web.api.dto.ComponentStateDTO;
+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 java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * Verifies that a Controller Service created by property migration survives the operations a deployed flow goes
+ * through: a runtime upgrade that introduces the migration, a plain runtime restart, and version changes of the
+ * enclosing versioned Process Group.
+ *
+ * Two independent flow lineages are pre-seeded in the registry. In the first, a later version declares the store
+ * Controller Service itself, as a published flow would once the vendor adds it. In the second, no version ever
+ * declares the service, so the flow relies entirely on property migration to create it, and the later version only
+ * adds an unrelated processor.
+ */
+public class MigrationCreatedControllerServiceVersioningIT extends AbstractNarSwapMigrationIT {
+    private static final String TEST_FLOWS_BUCKET = "test-flows";
+    private static final String SERVICE_DECLARED_FLOW_ID = "11111111-2222-3333-4444-555555555555";
+    private static final String SERVICE_ABSENT_FLOW_ID = "22222222-3333-4444-5555-666666666666";
+    private static final String DECLARED_SERVICE_VERSIONED_ID = "99999999-8888-7777-6666-555555555555";
+    private static final String STORE_SERVICE_PROPERTY = "Store Service";
+    private static final String STORE_SERVICE_TYPE = "org.apache.nifi.cs.tests.system.StateBackedStoreService";
+    private static final String PROCESSOR_TYPE = "org.apache.nifi.processors.tests.system.MigrateToControllerService";
+    private static final String MIGRATING_PROCESSOR_NAME = "MigrateToControllerService";
+    private static final String ADDED_PROCESSOR_NAME = "Added Processor";
+    private static final String VERSIONED_FLOWS_DIRECTORY = "src/test/resources/versioned-flows";
+    private static final String CREATED_STATE_KEY = "created";
+    private static final String ROW_COUNT_STATE_KEY = "rowCount";
+    private static final Duration CONDITION_TIMEOUT = Duration.ofSeconds(30);
+    private static final Duration CONDITION_POLL = Duration.ofMillis(100);
+
+    /**
+     * After the runtime is upgraded, the Controller Service that property migration creates must be present,
+     * enabled and referenced, and the flow that was running before the upgrade must be running again, with no manual action.
+     */
+    @Test
+    public void testRuntimeUpgradeCreatesEnabledServiceAndKeepsFlowRunning() throws NiFiClientException, IOException, InterruptedException {
+        final MigratedFlow flow = importAndUpgradeRuntime(SERVICE_DECLARED_FLOW_ID);
+        final ControllerServiceEntity service = waitForSingleStoreService(flow.groupId());
+        final String serviceId = service.getComponent().getId();
+
+        assertEquals(StandardControllerServiceFactory.MIGRATION_CREATED_COMMENT, service.getComponent().getComments());
+        assertBelongsToLocalFlowOnly(service);
+        getClientUtil().waitForControllerServiceRunStatus(serviceId, "ENABLED");
+        getClientUtil().waitForRunningProcessor(flow.processorId());
+        assertEquals(serviceId, getStoreServiceId(flow.processorId()));
+
+        final Collection<String> validationErrors = getNifiClient().getProcessorClient().getProcessor(flow.processorId()).getComponent().getValidationErrors();
+        final boolean processorValid = validationErrors == null || validationErrors.isEmpty();
+        assertTrue(processorValid, "Processor must be valid after the runtime upgrade");
+
+        waitForCondition(() -> countRows(serviceId) > 0, "store row count > 0");
+
+        final String versionedFlowState = getClientUtil().getVersionedFlowState(flow.groupId(), "root");
+        assertNotEquals("LOCALLY_MODIFIED", versionedFlowState, "The migration-created Controller Service must not make the flow dirty");
+        assertNotEquals("LOCALLY_MODIFIED_AND_STALE", versionedFlowState, "The migration-created Controller Service must not make the flow dirty");
+
+        final boolean serviceReportedAsLocalModification = getNifiClient().getProcessGroupClient().getLocalModifications(flow.groupId())
+                .getComponentDifferences().stream()
+                .anyMatch(diff -> serviceId.equals(diff.getComponentId()));
+        assertFalse(serviceReportedAsLocalModification);
+    }
+
+    /**
+     * Upgrading to a flow version that declares the store Controller Service must keep using the service that
+     * property migration already created, rather than removing it and substituting the one the published version declares.
+     */
+    @Test
+    public void testFlowUpgradePreservesMigrationCreatedControllerService() throws NiFiClientException, IOException, InterruptedException {
+        final MigratedFlow flow = importAndUpgradeRuntime(SERVICE_DECLARED_FLOW_ID);
+        final MigratedStore store = awaitPopulatedStoreService(flow);
+        assertBelongsToLocalFlowOnly(waitForSingleStoreService(flow.groupId()));
+
+        final VersionedFlowUpdateRequestEntity upgradeRequest = getClientUtil().changeFlowVersion(flow.groupId(), "2", false);
+
+        assertStorePreserved(flow, store, "flow upgrade");
+        getClientUtil().assertFlowUpToDate(flow.groupId());
+
+        final ControllerServiceEntity serviceAfterUpgrade = waitForSingleStoreService(flow.groupId());
+        assertEquals(DECLARED_SERVICE_VERSIONED_ID, serviceAfterUpgrade.getComponent().getVersionedComponentId(),
+                "The service must track the Controller Service that version 2 declares, instead of belonging to the local flow only");
+        assertEquals("", serviceAfterUpgrade.getComponent().getComments(),
+                "Comments must come from version 2 once the service tracks the declared Controller Service");
+        assertNull(upgradeRequest.getRequest().getFailureReason(),
+                "The migration-created Controller Service must not prevent the versioned flow from upgrading");
+    }
+
+    /**
+     * A flow whose definition never declares the store Controller Service relies entirely on property migration to
+     * create it. Upgrading such a flow to a version that only adds an unrelated processor, leaving the migrating
+     * processor untouched, must not disturb the service: the version change has nothing to say about it.
+     */
+    @Test
+    public void testFlowUpgradeAddingUnrelatedProcessorPreservesMigrationCreatedControllerService() throws NiFiClientException, IOException, InterruptedException {
+        final MigratedFlow flow = importAndUpgradeRuntime(SERVICE_ABSENT_FLOW_ID);
+        final MigratedStore store = awaitPopulatedStoreService(flow);
+
+        final VersionedFlowUpdateRequestEntity upgradeRequest = getClientUtil().changeFlowVersion(flow.groupId(), "2", false);
+
+        assertTrue(hasProcessorNamed(flow.groupId(), ADDED_PROCESSOR_NAME), "The upgrade must have added the unrelated processor");
+
+        assertStorePreserved(flow, store, "flow upgrade");
+        getClientUtil().assertFlowUpToDate(flow.groupId());
+
+        final ControllerServiceEntity serviceAfterUpgrade = waitForSingleStoreService(flow.groupId());
+        assertBelongsToLocalFlowOnly(serviceAfterUpgrade);
+        assertEquals(StandardControllerServiceFactory.MIGRATION_CREATED_COMMENT, serviceAfterUpgrade.getComponent().getComments());
+        assertNull(upgradeRequest.getRequest().getFailureReason(),
+                "The migration-created Controller Service must not prevent the versioned flow from upgrading");
+    }
+
+    /**
+     * Restarting the runtime with no NAR change must reuse the Controller Service that property migration
+     * created previously rather than creating a second one.
+     */
+    @Test
+    public void testRuntimeRestartDoesNotRecreateMigrationCreatedControllerService() throws NiFiClientException, IOException, InterruptedException {
+        final MigratedFlow flow = importAndUpgradeRuntime(SERVICE_DECLARED_FLOW_ID);
+        final MigratedStore store = awaitPopulatedStoreService(flow);
+
+        getNiFiInstance().stop();
+        getNiFiInstance().start(true);
+
+        assertStorePreserved(flow, store, "runtime restart");
+        assertBelongsToLocalFlowOnly(waitForSingleStoreService(flow.groupId()));
+    }
+
+    /**
+     * Asserts that the given Controller Service has no counterpart in the flow definition. Such a service either
+     * carries no versioned component id at all, or carries the placeholder that the framework derives from its own
+     * instance id while mapping the group. A service that tracks a Controller Service declared by the flow definition
+     * carries the identifier from the definition instead.
+     */
+    private void assertBelongsToLocalFlowOnly(final ControllerServiceEntity service) {
+        final String versionedComponentId = service.getComponent().getVersionedComponentId();
+        if (versionedComponentId == null) {
+            return;
+        }
+
+        final String derivedFromInstanceId = UUID.nameUUIDFromBytes(service.getComponent().getId().getBytes(StandardCharsets.UTF_8)).toString();
+        assertEquals(derivedFromInstanceId, versionedComponentId,
+                "The service tracks a Controller Service declared by the flow definition, so it no longer belongs to the local flow only");
+    }
+
+    /**
+     * Imports version 1 of the given pre-seeded flow, waits for its processor to be running, and then simulates a
+     * runtime upgrade by stopping NiFi, swapping in the alternate-config extensions and starting NiFi again. Nothing
+     * in the flow is stopped or started by hand, so the assertions that follow observe what the upgrade does on its own.
+     */
+    private MigratedFlow importAndUpgradeRuntime(final String flowId) throws NiFiClientException, IOException, InterruptedException {
+        final FlowRegistryClientEntity registryClient = registerClient(new File(VERSIONED_FLOWS_DIRECTORY));
+        final ProcessGroupEntity group = getClientUtil().importFlowFromRegistry("root", registryClient.getId(), TEST_FLOWS_BUCKET, flowId, "1");
+        final ProcessorEntity processor = findProcessor(group.getId(), MIGRATING_PROCESSOR_NAME);
+        getClientUtil().waitForRunningProcessor(processor.getId());
+
+        getNiFiInstance().stop();
+        switchOutNars();
+        getNiFiInstance().start(true);
+
+        return new MigratedFlow(group.getId(), processor.getId());
+    }
+
+    /**
+     * Waits until the migration-created Controller Service is enabled and its store has accumulated at least one row,
+     * so that the flow is known to be working before the operation under test runs, and returns its identifier.
+     */
+    private MigratedStore awaitPopulatedStoreService(final MigratedFlow flow) throws NiFiClientException, IOException, InterruptedException {
+        final ControllerServiceEntity service = waitForSingleStoreService(flow.groupId());
+        final String serviceId = service.getComponent().getId();
+        getClientUtil().waitForControllerServiceRunStatus(serviceId, "ENABLED");
+        waitForCondition(() -> countRows(serviceId) > 0, "store row count > 0");
+
+        return new MigratedStore(serviceId, readCreatedTimestamp(flow, serviceId));
+    }
+
+    /**
+     * Asserts that the given operation left the migration-created Controller Service in place and the flow working.
+     *
+     * Two independent properties are checked. An unchanged creation timestamp proves the service was never torn down,
+     * since removing a Controller Service clears its state and a replacement records a new timestamp. A row count that
+     * keeps climbing afterwards proves the flow is doing work again, which a RUNNING state alone does not show: a
+     * processor whose Controller Service reference is broken still reports RUNNING while writing nothing.
+     *
+     * The store is read through the service the processor references now rather than the one recorded earlier, so
+     * that a service substituted by the operation is detected rather than silently passing.
+     */
+    private void assertStorePreserved(final MigratedFlow flow, final MigratedStore store, final String operation)
+            throws NiFiClientException, IOException, InterruptedException {
+
+        waitForCondition(() -> !findStoreServices(flow.groupId()).isEmpty(), "store Controller Service found");
+
+        final List<ControllerServiceEntity> servicesAfter = findControllerServicesInGroup(flow.groupId());
+        assertEquals(1, servicesAfter.size(),
+                "The " + operation + " left " + servicesAfter.size() + " Controller Services in the group, so one was created alongside the migration-created one");
+
+        final ControllerServiceEntity serviceAfter = servicesAfter.getFirst();
+        assertEquals(store.serviceId(), serviceAfter.getComponent().getId(),
+                "The " + operation + " replaced the migration-created Controller Service with a different one");
+        assertEquals(store.serviceId(), getStoreServiceId(flow.processorId()));
+
+        getClientUtil().waitForControllerServiceRunStatus(store.serviceId(), "ENABLED");
+        getClientUtil().waitForRunningProcessor(flow.processorId());
+
+        assertEquals(store.created(), readCreatedTimestamp(flow, store.serviceId()),
+                "The migration-created Controller Service was removed during the " + operation + ": its store was created again, so the state it held is gone");
+
+        final long rowsAfterOperation = countRows(getStoreServiceId(flow.processorId()));
+        waitForCondition(() -> countRows(getStoreServiceId(flow.processorId())) > rowsAfterOperation, "store row count increase");
+    }
+
+
+    private ProcessorEntity findProcessor(final String groupId, final String name) throws NiFiClientException, IOException {
+        final List<ProcessorEntity> matching = getNifiClient().getFlowClient().getProcessGroup(groupId).getProcessGroupFlow().getFlow().getProcessors().stream()
+                .filter(processor -> PROCESSOR_TYPE.equals(processor.getComponent().getType()))
+                .filter(processor -> name.equals(processor.getComponent().getName()))
+                .toList();
+
+        if (matching.size() != 1) {
+            throw new AssertionError("Expected exactly one processor named " + name + " in group " + groupId + " but found " + matching.size());
+        }
+
+        return matching.getFirst();
+    }
+
+    private boolean hasProcessorNamed(final String groupId, final String name) throws NiFiClientException, IOException {
+        return getNifiClient().getFlowClient().getProcessGroup(groupId).getProcessGroupFlow().getFlow().getProcessors().stream()
+                .anyMatch(processor -> name.equals(processor.getComponent().getName()));
+    }
+
+    private List<ControllerServiceEntity> findControllerServicesInGroup(final String groupId) throws NiFiClientException, IOException {
+        return getNifiClient().getFlowClient().getControllerServices(groupId).getControllerServices().stream()
+                .filter(service -> groupId.equals(service.getComponent().getParentGroupId()))
+                .toList();
+    }
+
+    private List<ControllerServiceEntity> findStoreServices(final String groupId) throws NiFiClientException, IOException {
+        return findControllerServicesInGroup(groupId).stream()
+                .filter(service -> STORE_SERVICE_TYPE.equals(service.getComponent().getType()))
+                .toList();
+    }
+
+    private ControllerServiceEntity waitForSingleStoreService(final String groupId) throws NiFiClientException, IOException, InterruptedException {
+        waitForCondition(() -> findStoreServices(groupId).size() == 1, "exactly one store Controller Service");
+        return findStoreServices(groupId).getFirst();
+    }
+
+    private String getStoreServiceId(final String processorId) throws NiFiClientException, IOException {
+        final Map<String, String> properties = getNifiClient().getProcessorClient().getProcessor(processorId).getComponent().getConfig().getProperties();
+        return properties.get(STORE_SERVICE_PROPERTY);
+    }
+
+    /**
+     * Reads the store's local component state. Removing a Controller Service clears its state, so an empty map means
+     * either that the service has not been enabled yet or that it was torn down.
+     */
+    private Map<String, String> readStoreState(final String serviceId) throws NiFiClientException, IOException {
+        final ComponentStateEntity stateEntity = getNifiClient().getControllerServicesClient().getControllerServiceState(serviceId);
+        final ComponentStateDTO componentState = stateEntity.getComponentState();
+        if (componentState == null || componentState.getLocalState() == null || componentState.getLocalState().getState() == null) {
+            return Map.of();
+        }
+
+        final Map<String, String> state = new HashMap<>();
+        for (final StateEntryDTO entry : componentState.getLocalState().getState()) {
+            state.put(entry.getKey(), entry.getValue());
+        }
+
+        return state;
+    }
+
+    /**
+     * Returns the timestamp recorded when the store was established, or null if the service is no longer in the group.
+     * A service that has been removed outright has no state to read, and reporting that as a missing timestamp lets the
+     * caller fail on the comparison rather than on a lookup error.
+     */
+    private String readCreatedTimestamp(final MigratedFlow flow, final String serviceId) throws NiFiClientException, IOException {
+        final boolean stillPresent = findStoreServices(flow.groupId()).stream()
+                .anyMatch(service -> serviceId.equals(service.getComponent().getId()));
+        if (!stillPresent) {
+            return null;
+        }
+
+        return readStoreState(serviceId).get(CREATED_STATE_KEY);
+    }
+
+    private long countRows(final String serviceId) throws NiFiClientException, IOException {
+        final String rowCount = readStoreState(serviceId).get(ROW_COUNT_STATE_KEY);
+        return rowCount == null ? 0 : Long.parseLong(rowCount);
+    }
+
+    /**
+     * Polls the given condition until it holds, failing the test once the timeout elapses. Conditions query a NiFi that
+     * may still be starting up or replacing components, so a failing query is treated as the condition not holding yet.
+     */
+    private void waitForCondition(final ExceptionalBooleanSupplier condition, final String conditionDescription) throws InterruptedException {

Review Comment:
   Replaced `waitForCondition` with `waitFor`.
   The difference is that we'll be waiting for up to 5 minutes for a condition instead of 30 seconds, but I don't think this is a concern.



##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java:
##########
@@ -1169,9 +1169,225 @@ private void removeMissingRpg(final ProcessGroup group, final VersionedProcessGr
         removeMissingComponents(group, proposed, rpgsByVersionedId, VersionedProcessGroup::getRemoteProcessGroups, ProcessGroup::removeRemoteProcessGroup);
     }
 
+    /**
+     * Assigns a proposed versioned id to a Controller Service created by property migration.
+     * The service must have exactly one referencer.
+     * The proposed counterpart of that referencer must point at a Controller Service of the same type.
+     * If those do not hold, the service stays unversioned.
+     * A proposed id is assigned to at most one local service.
+     */
+    private void assignVersionedIdsToMigrationCreatedControllerServices(final ProcessGroup group, final VersionedProcessGroup proposed) {
+        final Collection<ControllerServiceNode> groupServices = group.getControllerServices(false);
+        if (groupServices == null || groupServices.isEmpty()) {

Review Comment:
   Removed the null check



-- 
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: issues-unsubscribe@nifi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] NIFI-16310 Preserve Controller Services created by migrateProperties during versioned flow synchronization [nifi]

["awelless (via GitHub)" <gi...@apache.org>]

awelless closed pull request #11670: NIFI-16310 Preserve Controller Services created by migrateProperties during versioned flow synchronization
URL: https://github.com/apache/nifi/pull/11670


-- 
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: issues-unsubscribe@nifi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org