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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
