NIFI-16310 Preserve Controller Services created by migrateProperties during versioned flow synchronization - #11670
Conversation
…during versioned flow synchronization
| * can observe whether store contents survive flow and runtime upgrades. | ||
| */ | ||
| @InputRequirement(Requirement.INPUT_FORBIDDEN) | ||
| public class MigrateToControllerService extends AbstractProcessor { |
There was a problem hiding this 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
| LOG.info("Keeping {} in {} because it was created by property migration and is not present in the proposed flow", | ||
| service, group); |
There was a problem hiding this 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.
exceptionfactory
left a comment
exceptionfactory
left a comment
There was a problem hiding this comment.
Thanks for working on this issue @awelless. The Synchronizer changes have a number of nuances, and although the general direction seems sound, I plan on taking another pass through the approach. I noted a handful of recommendations, mostly focusing on the tests.
| /** | ||
| * 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. | ||
| */ |
There was a problem hiding this comment.
It would be better to use this as the Capability Description annotation, instead of the class comment for semantic definition.
| * 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 { |
There was a problem hiding this comment.
For a new test class, the public modifiers are not needed at the class and method level.
| * 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 { |
There was a problem hiding this comment.
This method appears to be duplicative of the waitFor() method in the base class.
| getClientUtil().assertFlowUpToDate(flow.groupId()); | ||
|
|
||
| final ControllerServiceEntity serviceAfterUpgrade = waitForSingleStoreService(flow.groupId()); | ||
| assertEquals(DECLARED_SERVICE_VERSIONED_ID, serviceAfterUpgrade.getComponent().getVersionedComponentId(), |
There was a problem hiding this comment.
Many of these assertion methods are more verbose than necessary and should be shortened.
| } | ||
|
|
||
| @Override | ||
| public synchronized void append(final String row) { |
There was a problem hiding this comment.
Why is this synchronized, is it necessary?
| 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()); |
There was a problem hiding this comment.
Repeated literals in multiple methods should be declared as static final variables and reused.
| */ | ||
| private void assignVersionedIdsToMigrationCreatedControllerServices(final ProcessGroup group, final VersionedProcessGroup proposed) { | ||
| final Collection<ControllerServiceNode> groupServices = group.getControllerServices(false); | ||
| if (groupServices == null || groupServices.isEmpty()) { |
There was a problem hiding this comment.
This method should never return null
| localService.getVersionedComponentId().ifPresent(claimedVersionedIds::add); | ||
| } | ||
|
|
||
| final Map<String, VersionedConfigurableExtension> proposedComponentsByVersionedId = indexByVersionedId(proposed.getControllerServices(), proposed.getProcessors()); |
There was a problem hiding this comment.
This method is on the longer side, it looks an opportunity to break it up into two methods after the initial checks.
2 participants
Summary
NIFI-16310
Controller Services created by
migratePropertiesmethod were always recreated during a versioned process group upgrade, causing the trigger of@OnRemovedhook 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
Additional changes
FileSystemFlowRegistryClientwhich is used in system tests was moved into a separate NAR, to be available before and after NAR swap.Suggested review order
MigrationCreatedControllerServiceVersioningITand its dependencies to see the relevant use cases.StandardVersionedComponentSynchronizerfor actual implementation.StandardVersionedComponentSynchronizerTestto see the desired behavior for edge cases.Tracking
Please complete the following tracking steps prior to pull request creation.
Issue Tracking
Pull Request Tracking
NIFI-00000NIFI-00000VerifiedstatusPull Request Formatting
mainbranchVerification
Please indicate the verification steps performed prior to pull request creation.
Build
./mvnw clean install -P contrib-checkLicensing
LICENSEandNOTICEfilesDocumentation