diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java index d8785cd9a2ac..1aa261eb1ffd 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java @@ -97,11 +97,13 @@ import org.apache.nifi.logging.LogLevel; import org.apache.nifi.migration.ControllerServiceFactory; import org.apache.nifi.migration.StandardControllerServiceFactory; +import org.apache.nifi.parameter.ExpressionLanguageAgnosticParameterParser; import org.apache.nifi.parameter.Parameter; import org.apache.nifi.parameter.ParameterContext; import org.apache.nifi.parameter.ParameterContextManager; import org.apache.nifi.parameter.ParameterContextNameUtils; import org.apache.nifi.parameter.ParameterDescriptor; +import org.apache.nifi.parameter.ParameterParser; import org.apache.nifi.parameter.ParameterProviderConfiguration; import org.apache.nifi.parameter.ParameterReferenceManager; import org.apache.nifi.parameter.ParameterReferencedControllerServiceData; @@ -1653,11 +1655,13 @@ private Map populatePropertiesMap(final ComponentNode componentN || (versionedDescriptor != null && versionedDescriptor.getIdentifiesControllerService()); final boolean sensitive = (descriptor != null && descriptor.isSensitive()) || (versionedDescriptor != null && versionedDescriptor.isSensitive()); - + final String proposedValue = proposedProperties.get(propertyName); + final ParameterParser parameterParser = new ExpressionLanguageAgnosticParameterParser(); + final boolean proposedReferencingParameter = parameterParser.parseTokens(proposedValue).iterator().hasNext(); final String value; - if (descriptor != null && referencesService && (proposedProperties.get(propertyName) != null)) { + if (descriptor != null && referencesService && !proposedReferencingParameter && (proposedValue != null)) { // Need to determine if the component's property descriptor for this service is already set to an id - // of an existing service that is outside the current processor group, and if it is we want to leave + // of an existing service that is outside the current processor group, and if it is, we want to leave // the property set to that value String existingExternalServiceId = null; final String componentDescriptorValue = componentNode.getEffectivePropertyValue(descriptor); @@ -1674,9 +1678,8 @@ private Map populatePropertiesMap(final ComponentNode componentN // If the component's property descriptor is not already set to an id of an existing external service, // then we need to take the Versioned Component ID and resolve this to the instance ID of the service if (existingExternalServiceId == null) { - final String serviceVersionedComponentId = proposedProperties.get(propertyName); - String instanceId = getServiceInstanceId(serviceVersionedComponentId, group); - value = (instanceId == null) ? serviceVersionedComponentId : instanceId; + String instanceId = getServiceInstanceId(proposedValue, group); + value = (instanceId == null) ? proposedValue : instanceId; // Find the same property descriptor in the component's CreatedExtension and replace it with the // instance ID of the service @@ -1687,7 +1690,7 @@ private Map populatePropertiesMap(final ComponentNode componentN value = existingExternalServiceId; } } else { - value = proposedProperties.get(propertyName); + value = proposedValue; } // skip any sensitive properties that are not populated so we can retain whatever is currently set. We do this because sensitive properties are not stored in the registry diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizerTest.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizerTest.java index 9ad6c27646e7..651031576423 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizerTest.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizerTest.java @@ -1021,6 +1021,198 @@ public void testExternalControllerServiceReferenceRemoved() throws FlowSynchroni assertNull(properties.get("cs")); } + @Test + @SuppressWarnings("unchecked") + public void testExternalControllerServiceParameterReferencePreserved() throws FlowSynchronizationException, InterruptedException, TimeoutException { + // A controller-service-identifying property is configured with a Parameter reference (#{svc}) that resolves to a + // controller service living outside this process group. The proposed (versioned) flow references the property via + // the SAME parameter reference. Synchronizing must PRESERVE the parameter reference. + // + // Bug: populatePropertiesMap resolves the effective value (#{svc} -> concrete service id), finds that id as an + // existing external controller service, and pins the property to the concrete instance id -- silently dropping the + // parameterization. On the next flow comparison this surfaces as DifferenceType.PROPERTY_PARAMETERIZATION_REMOVED. + // This test asserts the correct behavior and therefore fails against the current implementation. + final String parameterReference = "#{svc}"; + final String externalServiceId = "external-service-id"; + + final PropertyDescriptor descriptorCS = new PropertyDescriptor.Builder().name("cs") + .identifiesControllerService(ControllerService.class).build(); + + final Map rawPropertyValues = new HashMap<>(); + rawPropertyValues.put(descriptorCS, parameterReference); + + final VersionedPropertyDescriptor versionedDescriptorCS = new VersionedPropertyDescriptor(); + versionedDescriptorCS.setName(descriptorCS.getName()); + final Map proposedDescriptors = new HashMap<>(); + proposedDescriptors.put(versionedDescriptorCS.getName(), versionedDescriptorCS); + + final Map propertiesBefore = new HashMap<>(); + propertiesBefore.put(descriptorCS, new PropertyConfiguration(parameterReference, null, null, null)); + + final ProcessorNode processorNode = createMockProcessor(); + when(processorNode.getPropertyDescriptor(eq("cs"))).thenReturn(descriptorCS); + when(processorNode.getProperties()).thenReturn(propertiesBefore); + when(processorNode.getRawPropertyValues()).thenReturn(rawPropertyValues); + // #{svc} resolves to the concrete id of a controller service outside this group. + when(processorNode.getEffectivePropertyValue(eq(descriptorCS))).thenReturn(externalServiceId); + when(processorNode.isReferencingParameter(eq(descriptorCS.getName()))).thenReturn(true); + + final ProcessGroup processGroup = processorNode.getProcessGroup(); + final ProcessGroup processGroupParent = mock(ProcessGroup.class); + final ControllerServiceNode externalService = createMockControllerService(); + when(processGroup.getParent()).thenReturn(processGroupParent); + when(processGroupParent.findControllerService(eq(externalServiceId), eq(false), eq(true))).thenReturn(externalService); + + // Proposed flow references the property via the SAME parameter reference. + final Map proposedProperties = new HashMap<>(); + proposedProperties.put("cs", parameterReference); + final VersionedProcessor versionedProcessor = createMinimalVersionedProcessor(); + versionedProcessor.setPropertyDescriptors(proposedDescriptors); + versionedProcessor.setProperties(proposedProperties); + + final ArgumentCaptor> captorProperties = ArgumentCaptor.forClass(Map.class); + synchronizer.synchronize(processorNode, versionedProcessor, group, synchronizationOptions); + verify(processorNode).setProperties(captorProperties.capture(), anyBoolean(), any()); + final Map properties = captorProperties.getValue(); + + assertEquals(parameterReference, properties.get("cs"), + "Controller-service property configured with a Parameter reference was flattened to the resolved service id"); + } + + @Test + @SuppressWarnings("unchecked") + public void testNonControllerServiceParameterReferencePreserved() throws FlowSynchronizationException, InterruptedException, TimeoutException { + // A regular (non-CS-identifying) property is configured with a Parameter reference (#{my_param}). + // The proposed (versioned) flow references the same parameter. Synchronizing must preserve the + // parameter reference rather than substituting any resolved effective value. + final String parameterReference = "#{my_param}"; + final String effectiveValue = "resolved-value"; + + final PropertyDescriptor descriptorA = new PropertyDescriptor.Builder().name("a").build(); + + final Map rawPropertyValues = new HashMap<>(); + rawPropertyValues.put(descriptorA, parameterReference); + + final VersionedPropertyDescriptor versionedDescriptorA = new VersionedPropertyDescriptor(); + versionedDescriptorA.setName(descriptorA.getName()); + final Map proposedDescriptors = new HashMap<>(); + proposedDescriptors.put(versionedDescriptorA.getName(), versionedDescriptorA); + + final Map propertiesBefore = new HashMap<>(); + propertiesBefore.put(descriptorA, new PropertyConfiguration(parameterReference, null, null, null)); + + final ProcessorNode processorNode = createMockProcessor(); + when(processorNode.getPropertyDescriptor(eq("a"))).thenReturn(descriptorA); + when(processorNode.getProperties()).thenReturn(propertiesBefore); + when(processorNode.getRawPropertyValues()).thenReturn(rawPropertyValues); + when(processorNode.getEffectivePropertyValue(eq(descriptorA))).thenReturn(effectiveValue); + when(processorNode.isReferencingParameter(eq(descriptorA.getName()))).thenReturn(true); + + final Map proposedProperties = new HashMap<>(); + proposedProperties.put("a", parameterReference); + final VersionedProcessor versionedProcessor = createMinimalVersionedProcessor(); + versionedProcessor.setPropertyDescriptors(proposedDescriptors); + versionedProcessor.setProperties(proposedProperties); + + final ArgumentCaptor> captorProperties = ArgumentCaptor.forClass(Map.class); + synchronizer.synchronize(processorNode, versionedProcessor, group, synchronizationOptions); + verify(processorNode).setProperties(captorProperties.capture(), anyBoolean(), any()); + final Map properties = captorProperties.getValue(); + + assertEquals(parameterReference, properties.get("a"), + "Non-CS property configured with a Parameter reference should preserve the reference, not resolve it"); + } + + @Test + @SuppressWarnings("unchecked") + public void testNonControllerServiceParameterChangedToHardCoded() throws FlowSynchronizationException, InterruptedException, TimeoutException { + // When the proposed flow replaces a Parameter reference with a hard-coded value on a non-CS property, + // the synchronizer must use the hard-coded proposed value, not the old parameter reference. + final String parameterReference = "#{my_param}"; + final String newExplicitValue = "new-explicit-value"; + + final PropertyDescriptor descriptorA = new PropertyDescriptor.Builder().name("a").build(); + + final Map rawPropertyValues = new HashMap<>(); + rawPropertyValues.put(descriptorA, parameterReference); + + final VersionedPropertyDescriptor versionedDescriptorA = new VersionedPropertyDescriptor(); + versionedDescriptorA.setName(descriptorA.getName()); + final Map proposedDescriptors = new HashMap<>(); + proposedDescriptors.put(versionedDescriptorA.getName(), versionedDescriptorA); + + final Map propertiesBefore = new HashMap<>(); + propertiesBefore.put(descriptorA, new PropertyConfiguration(parameterReference, null, null, null)); + + final ProcessorNode processorNode = createMockProcessor(); + when(processorNode.getPropertyDescriptor(eq("a"))).thenReturn(descriptorA); + when(processorNode.getProperties()).thenReturn(propertiesBefore); + when(processorNode.getRawPropertyValues()).thenReturn(rawPropertyValues); + when(processorNode.isReferencingParameter(eq(descriptorA.getName()))).thenReturn(true); + + final Map proposedProperties = new HashMap<>(); + proposedProperties.put("a", newExplicitValue); + final VersionedProcessor versionedProcessor = createMinimalVersionedProcessor(); + versionedProcessor.setPropertyDescriptors(proposedDescriptors); + versionedProcessor.setProperties(proposedProperties); + + final ArgumentCaptor> captorProperties = ArgumentCaptor.forClass(Map.class); + synchronizer.synchronize(processorNode, versionedProcessor, group, synchronizationOptions); + verify(processorNode).setProperties(captorProperties.capture(), anyBoolean(), any()); + final Map properties = captorProperties.getValue(); + + assertEquals(newExplicitValue, properties.get("a"), + "Non-CS property previously set via Parameter reference should adopt the hard-coded value from the proposed flow"); + } + + @Test + @SuppressWarnings("unchecked") + public void testControllerServiceParameterChangedToHardCoded() throws FlowSynchronizationException, InterruptedException, TimeoutException { + // When the proposed flow replaces a Parameter reference with a hard-coded versioned component ID on + // a CS-identifying property, the synchronizer must resolve the versioned ID to the local instance ID. + final String parameterReference = "#{svc}"; + final String versionedCsId = "versioned-cs-uuid"; + + final PropertyDescriptor descriptorCS = new PropertyDescriptor.Builder().name("cs") + .identifiesControllerService(ControllerService.class).build(); + + final Map rawPropertyValues = new HashMap<>(); + rawPropertyValues.put(descriptorCS, parameterReference); + + final VersionedPropertyDescriptor versionedDescriptorCS = new VersionedPropertyDescriptor(); + versionedDescriptorCS.setName(descriptorCS.getName()); + final Map proposedDescriptors = new HashMap<>(); + proposedDescriptors.put(versionedDescriptorCS.getName(), versionedDescriptorCS); + + final Map propertiesBefore = new HashMap<>(); + propertiesBefore.put(descriptorCS, new PropertyConfiguration(parameterReference, null, null, null)); + + final ProcessorNode processorNode = createMockProcessor(); + when(processorNode.getPropertyDescriptor(eq("cs"))).thenReturn(descriptorCS); + when(processorNode.getProperties()).thenReturn(propertiesBefore); + when(processorNode.getRawPropertyValues()).thenReturn(rawPropertyValues); + when(processorNode.isReferencingParameter(eq(descriptorCS.getName()))).thenReturn(true); + + // Wire up a local controller service that the versioned ID resolves to. + final ControllerServiceNode localService = createMockControllerService(); + when(localService.getVersionedComponentId()).thenReturn(Optional.of(versionedCsId)); + when(group.getControllerServices(false)).thenReturn(Set.of(localService)); + + final Map proposedProperties = new HashMap<>(); + proposedProperties.put("cs", versionedCsId); + final VersionedProcessor versionedProcessor = createMinimalVersionedProcessor(); + versionedProcessor.setPropertyDescriptors(proposedDescriptors); + versionedProcessor.setProperties(proposedProperties); + + final ArgumentCaptor> captorProperties = ArgumentCaptor.forClass(Map.class); + synchronizer.synchronize(processorNode, versionedProcessor, group, synchronizationOptions); + verify(processorNode).setProperties(captorProperties.capture(), anyBoolean(), any()); + final Map properties = captorProperties.getValue(); + + assertEquals(localService.getIdentifier(), properties.get("cs"), + "CS-identifying property previously set via Parameter reference should be resolved to the local service instance ID"); + } + @Test public void testControllerServiceRemoved() throws FlowSynchronizationException, InterruptedException, TimeoutException { final ControllerServiceNode service = createMockControllerService(); diff --git a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/registry/ExternalControllerServiceVersioningIT.java b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/registry/ExternalControllerServiceVersioningIT.java index d74000f79357..b79bc96bb01d 100644 --- a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/registry/ExternalControllerServiceVersioningIT.java +++ b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/registry/ExternalControllerServiceVersioningIT.java @@ -28,6 +28,7 @@ import org.apache.nifi.web.api.entity.ControllerServiceEntity; import org.apache.nifi.web.api.entity.FlowComparisonEntity; import org.apache.nifi.web.api.entity.FlowRegistryClientEntity; +import org.apache.nifi.web.api.entity.ParameterContextEntity; import org.apache.nifi.web.api.entity.ProcessGroupEntity; import org.apache.nifi.web.api.entity.ProcessGroupFlowEntity; import org.apache.nifi.web.api.entity.ProcessorEntity; @@ -243,6 +244,58 @@ public void testExternalServiceReferenceWithoutModificationShowsUpToDate() throw "Show Local Changes should report no differences for an unmodified flow"); } + /** + * Reproduces NIFI-16114: a processor property that identifies a Controller Service is set via + * a Parameter reference (#{svc}) pointing at a service defined outside the versioned PG. Both + * v1 and v2 of the flow reference the service through the same Parameter. + * + * When the PG is downgraded to v1 and then upgraded back to v2, the synchronizer used to resolve + * "#{svc}" to the external service's concrete instance ID and pin the property to that literal + * value, silently dropping the Parameter reference. This showed up as a false "local + * modification" (Property Value Changed) even though the flow and the versioned snapshot were + * otherwise identical. + */ + @Test + public void testExternalControllerServiceParameterReferencePreservedOnUpgrade() throws NiFiClientException, IOException, InterruptedException { + final FlowRegistryClientEntity registryClient = registerClient(); + final NiFiClientUtil util = getClientUtil(); + + final ControllerServiceEntity service = util.createControllerService(COUNT_SERVICE_TYPE, "root"); + util.enableControllerService(service); + + final ParameterContextEntity paramContext = util.createParameterContext( + "svc-context", Collections.singletonMap("svc", service.getComponent().getId())); + + final ProcessGroupEntity child = util.createProcessGroup("Child", "root"); + util.setParameterContext(child.getId(), paramContext); + + ProcessorEntity counter = util.createProcessor("CountFlowFiles", child.getId()); + util.updateProcessorProperties(counter, Collections.singletonMap("Count Service", "#{svc}")); + final ProcessorEntity terminate = util.createProcessor("TerminateFlowFile", child.getId()); + util.createConnection(counter, terminate, "success"); + + final VersionControlInformationEntity vci = util.startVersionControl(child, registryClient, TEST_FLOWS_BUCKET, "param-ref-external-cs"); + util.assertFlowUpToDate(child.getId()); + + // v2 differs only in scheduling period. "Count Service" remains #{svc} in both versions. + counter = util.updateProcessorSchedulingPeriod(counter, "10 sec"); + util.saveFlowVersion(child, registryClient, vci); + util.assertFlowUpToDate(child.getId()); + + // Downgrade then upgrade to force the synchronizer to re-resolve the property, which is + // the code path affected by NIFI-16114. + util.changeFlowVersion(child.getId(), "1"); + util.changeFlowVersion(child.getId(), "2"); + + final ProcessorEntity refreshed = getNifiClient().getProcessorClient().getProcessor(counter.getId()); + assertEquals("#{svc}", refreshed.getComponent().getConfig().getProperties().get("Count Service"), + "Parameter reference to external Controller Service should survive version upgrade, not be flattened to the service's instance ID"); + + final FlowComparisonEntity localMods = getNifiClient().getProcessGroupClient().getLocalModifications(child.getId()); + assertTrue(localMods.getComponentDifferences().isEmpty(), + "Show Local Changes should report no differences after upgrading between versions that both reference the service via the same Parameter"); + } + /** * Deletes only the connections and processors within a Process Group, without touching * Controller Services (which may be inherited from ancestor groups).