From f87a61baaf77fa9b0678cf8e01d1c7eff28cb5c8 Mon Sep 17 00:00:00 2001 From: Pierre Villard Date: Tue, 7 Jul 2026 14:59:48 +0200 Subject: [PATCH 1/2] NIFI-16085 - Flow synchronization fails when a Parameter Provider-backed Parameter Context contains a parameter not flagged as provided --- ...tandardVersionedComponentSynchronizer.java | 17 +- .../TestStandardParameterContext.java | 39 ++++ .../VersionedFlowSynchronizer.java | 11 +- .../VersionedFlowSynchronizerTest.java | 78 +++++++ .../ClusteredProviderParamFlowSyncIT.java | 191 ++++++++++++++++++ 5 files changed, 326 insertions(+), 10 deletions(-) create mode 100644 nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/parameters/ClusteredProviderParamFlowSyncIT.java 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..2d8fc7cdec91 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 @@ -1842,7 +1842,7 @@ public void synchronize(final ParameterContext parameterContext, final Versioned contextManager.removeParameterContext(parameterContext.getIdentifier()); LOG.info("Successfully synchronized {} by removing it from the flow", parameterContext); } else { - final Map updatedParameters = createParameterMap(proposed.getParameters()); + final Map updatedParameters = createParameterMap(proposed.getParameters(), proposed.getParameterProvider() != null); // If any parameters are removed, need to add a null value to the map in order to make sure that the parameter is removed. for (final ParameterDescriptor existingParameterDescriptor : parameterContext.getParameters().keySet()) { @@ -2324,7 +2324,8 @@ private ParameterContext createParameterContextWithoutReferences(final Versioned continue; } - final Parameter parameter = createParameter(null, versionedParameter); + // Created without a Parameter Provider configuration, so parameters keep their serialized provided flag. + final Parameter parameter = createParameter(null, versionedParameter, false); parameters.put(versionedParameter.getName(), parameter); } @@ -2341,7 +2342,7 @@ private ParameterContext createParameterContext(final VersionedParameterContext final Map versionedParameterContexts, final Map parameterProviderReferences, final ComponentIdGenerator componentIdGenerator) { - final Map parameters = createParameterMap(versionedParameterContext.getParameters()); + final Map parameters = createParameterMap(versionedParameterContext.getParameters(), versionedParameterContext.getParameterProvider() != null); final List parameterContextRefs = new ArrayList<>(); if (versionedParameterContext.getInheritedParameterContexts() != null) { @@ -2361,10 +2362,10 @@ private ParameterContext createParameterContext(final VersionedParameterContext return contextReference.get(); } - private Map createParameterMap(final Collection versionedParameters) { + private Map createParameterMap(final Collection versionedParameters, final boolean providerBacked) { final Map parameters = new HashMap<>(); for (final VersionedParameter versionedParameter : versionedParameters) { - final Parameter parameter = createParameter(null, versionedParameter); + final Parameter parameter = createParameter(null, versionedParameter, providerBacked); parameters.put(versionedParameter.getName(), parameter); } @@ -2426,7 +2427,7 @@ private void addMissingConfiguration(final VersionedParameterContext versionedPa continue; } - final Parameter parameter = createParameter(currentParameterContext.getIdentifier(), versionedParameter); + final Parameter parameter = createParameter(currentParameterContext.getIdentifier(), versionedParameter, versionedParameterContext.getParameterProvider() != null); parameters.put(versionedParameter.getName(), parameter); } @@ -2476,7 +2477,7 @@ private void addMissingConfiguration(final VersionedParameterContext versionedPa } } - private Parameter createParameter(final String contextId, final VersionedParameter versionedParameter) { + private Parameter createParameter(final String contextId, final VersionedParameter versionedParameter, final boolean providerBacked) { final List referencedAssets = versionedParameter.getReferencedAssets(); final List assets; @@ -2498,7 +2499,7 @@ private Parameter createParameter(final String contextId, final VersionedParamet .sensitive(versionedParameter.isSensitive()) .value(versionedParameter.getValue()) .referencedAssets(assets) - .provided(versionedParameter.isProvided()) + .provided(providerBacked || versionedParameter.isProvided()) .parameterContextId(contextId) .build(); } diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/parameter/TestStandardParameterContext.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/parameter/TestStandardParameterContext.java index 992be669bab0..d6d2cd32a045 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/parameter/TestStandardParameterContext.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/parameter/TestStandardParameterContext.java @@ -17,8 +17,10 @@ package org.apache.nifi.parameter; import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.controller.ParameterProviderNode; import org.apache.nifi.controller.ProcessorNode; import org.apache.nifi.controller.PropertyConfiguration; +import org.apache.nifi.controller.parameter.ParameterProviderLookup; import org.apache.nifi.controller.service.ControllerServiceNode; import org.apache.nifi.controller.service.ControllerServiceState; import org.apache.nifi.groups.ProcessGroup; @@ -649,6 +651,43 @@ public void testChangingParameterForEnabledControllerService() { } } + @Test + public void testSetParametersRejectedWhenProviderBackedAndParameterNotProvided() { + final String providerId = "provider-1"; + + final ParameterProvider parameterProvider = mock(ParameterProvider.class); + when(parameterProvider.getIdentifier()).thenReturn(providerId); + + final ParameterProviderNode parameterProviderNode = mock(ParameterProviderNode.class); + when(parameterProviderNode.getParameterProvider()).thenReturn(parameterProvider); + + final ParameterProviderLookup parameterProviderLookup = mock(ParameterProviderLookup.class); + when(parameterProviderLookup.getParameterProvider(providerId)).thenReturn(parameterProviderNode); + + // A Parameter Context whose parameters are supplied by a Parameter Provider. + final ParameterContext context = new StandardParameterContext.Builder() + .id("provider-backed-context") + .name("Provider Backed Context") + .parameterReferenceManager(new HashMapParameterReferenceManager()) + .parameterProviderLookup(parameterProviderLookup) + .parameterProviderConfiguration(new StandardParameterProviderConfiguration(providerId, "Group", true)) + .build(); + + // Setting a user-entered (provided=false) parameter on a provider-backed context must be rejected. + // This is the guard that, during cluster flow synchronization, surfaces as a FlowSynchronizationException + // when the proposed (cluster) flow carries a non-provided parameter for a provider-backed context. + final ParameterDescriptor descriptor = new ParameterDescriptor.Builder().name("db.host").build(); + final Parameter userEnteredParameter = new Parameter.Builder() + .descriptor(descriptor) + .value("localhost") + .provided(false) + .build(); + + final IllegalArgumentException e = assertThrows(IllegalArgumentException.class, + () -> context.setParameters(Collections.singletonMap("db.host", userEnteredParameter))); + assertTrue(e.getMessage().contains("cannot be manually updated because they are provided by Parameter Provider")); + } + private ParameterContext createStandardParameterContext(final ParameterReferenceManager referenceManager) { return new StandardParameterContext.Builder() .id("unit-test-context") diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizer.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizer.java index 29310984eb88..11a1de113df4 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizer.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizer.java @@ -944,13 +944,20 @@ private Map createParameterMap( ) { final Map providedParameters = getProvidedParameters(flowManager, versionedParameterContext); + // A Parameter Context bound to a Parameter Provider owns all of its Parameters through that Provider; the + // context-level binding is authoritative, not the per-Parameter provided flag in the serialized flow. Treat + // every Parameter of such a context as provided so reconciliation does not attempt a manual (user-entered) + // update, which the Parameter Context rejects. + final boolean providerBacked = versionedParameterContext.getParameterProvider() != null; + final Map parameters = new HashMap<>(); for (final VersionedParameter versioned : versionedParameterContext.getParameters()) { + final boolean provided = providerBacked || versioned.isProvided(); final String parameterValue; final String rawValue = versioned.getValue(); if (rawValue == null) { parameterValue = null; - } else if (versioned.isProvided()) { + } else if (provided) { final String name = versioned.getName(); final Parameter providedParameter = providedParameters.get(name); if (providedParameter == null) { @@ -973,7 +980,7 @@ private Map createParameterMap( .sensitive(versioned.isSensitive()) .value(referencedAssets.isEmpty() ? parameterValue : null) .referencedAssets(referencedAssets) - .provided(versioned.isProvided()) + .provided(provided) .build(); parameters.put(versioned.getName(), parameter); diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizerTest.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizerTest.java index 96a1432301f6..7affc9df6600 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizerTest.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizerTest.java @@ -24,11 +24,13 @@ import org.apache.nifi.components.connector.ConnectorSyncMode; import org.apache.nifi.components.connector.ConnectorSyncResult; import org.apache.nifi.controller.FlowController; +import org.apache.nifi.controller.ParameterProviderNode; import org.apache.nifi.controller.ReportingTaskNode; import org.apache.nifi.controller.SnippetManager; import org.apache.nifi.controller.UninheritableFlowException; import org.apache.nifi.controller.flow.FlowManager; import org.apache.nifi.controller.flow.VersionedDataflow; +import org.apache.nifi.controller.parameter.ParameterProviderLookup; import org.apache.nifi.controller.service.ControllerServiceNode; import org.apache.nifi.controller.service.ControllerServiceProvider; import org.apache.nifi.encrypt.PropertyEncryptor; @@ -37,11 +39,20 @@ import org.apache.nifi.flow.VersionedConnector; import org.apache.nifi.flow.VersionedConnectorState; import org.apache.nifi.flow.VersionedControllerService; +import org.apache.nifi.flow.VersionedParameter; +import org.apache.nifi.flow.VersionedParameterContext; import org.apache.nifi.flow.VersionedProcessGroup; import org.apache.nifi.flow.VersionedReportingTask; import org.apache.nifi.groups.BundleUpdateStrategy; import org.apache.nifi.groups.ProcessGroup; import org.apache.nifi.nar.ExtensionManager; +import org.apache.nifi.parameter.Parameter; +import org.apache.nifi.parameter.ParameterDescriptor; +import org.apache.nifi.parameter.ParameterProvider; +import org.apache.nifi.parameter.ParameterReferenceManager; +import org.apache.nifi.parameter.StandardParameterContext; +import org.apache.nifi.parameter.StandardParameterContextManager; +import org.apache.nifi.parameter.StandardParameterProviderConfiguration; import org.apache.nifi.persistence.FlowConfigurationArchiveManager; import org.apache.nifi.registry.flow.mapping.VersionedComponentStateLookup; import org.apache.nifi.services.FlowService; @@ -63,13 +74,16 @@ import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.UUID; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -276,6 +290,70 @@ private VersionedControllerService getVersionedControllerService() { return controllerService; } + @Test + void testSyncReconcilesProviderBackedContextWithNonProvidedParameter() { + setRootGroup(); + setFlowController(); + + final String providerId = "provider-1"; + final String contextName = "openflow-rds-ingest"; + final String paramName = "db.host"; + + // Parameter Provider infrastructure + final ParameterProvider parameterProvider = mock(ParameterProvider.class); + when(parameterProvider.getIdentifier()).thenReturn(providerId); + final ParameterProviderNode parameterProviderNode = mock(ParameterProviderNode.class); + when(parameterProviderNode.getParameterProvider()).thenReturn(parameterProvider); + final ParameterProviderLookup parameterProviderLookup = mock(ParameterProviderLookup.class); + when(parameterProviderLookup.getParameterProvider(providerId)).thenReturn(parameterProviderNode); + + // The node's existing, self-consistent flow: a provider-backed context whose parameter is provider-supplied. + final StandardParameterContext existingContext = new StandardParameterContext.Builder() + .id("provider-backed-context") + .name(contextName) + .parameterReferenceManager(ParameterReferenceManager.EMPTY) + .parameterProviderLookup(parameterProviderLookup) + .parameterProviderConfiguration(new StandardParameterProviderConfiguration(providerId, "Group", true)) + .build(); + final ParameterDescriptor descriptor = new ParameterDescriptor.Builder().name(paramName).build(); + existingContext.setParameters(Collections.singletonMap(paramName, + new Parameter.Builder().descriptor(descriptor).value("localhost").provided(true).build())); + + final StandardParameterContextManager contextManager = new StandardParameterContextManager(); + contextManager.addParameterContext(existingContext); + when(flowManager.getParameterContextManager()).thenReturn(contextManager); + when(flowManager.getParameterProvider(providerId)).thenReturn(parameterProviderNode); + doAnswer(invocation -> { + invocation.getArgument(0, Runnable.class).run(); + return null; + }).when(flowManager).withParameterContextResolution(any()); + + // The proposed (elected cluster) flow: same provider-backed context, but the parameter is flagged + // provided=false with a divergent value. This is the serialized-flow contradiction seen in production. + final VersionedParameter versionedParameter = new VersionedParameter(); + versionedParameter.setName(paramName); + versionedParameter.setValue("different-host"); + versionedParameter.setSensitive(false); + versionedParameter.setProvided(false); + + final VersionedParameterContext versionedParameterContext = new VersionedParameterContext(); + versionedParameterContext.setName(contextName); + versionedParameterContext.setParameterProvider(providerId); + versionedParameterContext.setParameterGroupName("Group"); + versionedParameterContext.setParameters(Collections.singleton(versionedParameter)); + when(versionedDataflow.getParameterContexts()).thenReturn(List.of(versionedParameterContext)); + + // With the fix, the synchronizer recognizes the context is provider-backed and reconciles it as + // provider-managed (provided=true, value sourced from the provider) rather than attempting a manual + // update, so synchronization succeeds instead of throwing FlowSynchronizationException. + assertDoesNotThrow(() -> + versionedFlowSynchronizer.sync(flowController, dataFlow, flowService, BundleUpdateStrategy.USE_SPECIFIED_OR_GHOST)); + + final Optional reconciled = existingContext.getParameter(paramName); + assertTrue(reconciled.isPresent(), "Parameter should still exist after reconciliation"); + assertTrue(reconciled.get().isProvided(), "Parameter must be reconciled as provider-supplied (provided=true)"); + } + private void setRootGroup() { when(flowController.getFlowManager()).thenReturn(flowManager); when(flowManager.getRootGroup()).thenReturn(rootGroup); diff --git a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/parameters/ClusteredProviderParamFlowSyncIT.java b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/parameters/ClusteredProviderParamFlowSyncIT.java new file mode 100644 index 000000000000..e48eac66b4be --- /dev/null +++ b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/parameters/ClusteredProviderParamFlowSyncIT.java @@ -0,0 +1,191 @@ +/* + * 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.parameters; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.nifi.controller.flow.VersionedDataflow; +import org.apache.nifi.flow.VersionedParameter; +import org.apache.nifi.flow.VersionedParameterContext; +import org.apache.nifi.parameter.ParameterSensitivity; +import org.apache.nifi.parameter.StandardParameterProviderConfiguration; +import org.apache.nifi.stream.io.StreamUtils; +import org.apache.nifi.tests.system.NiFiInstance; +import org.apache.nifi.tests.system.NiFiInstanceFactory; +import org.apache.nifi.tests.system.NiFiSystemIT; +import org.apache.nifi.web.api.entity.ParameterContextEntity; +import org.apache.nifi.web.api.entity.ParameterGroupConfigurationEntity; +import org.apache.nifi.web.api.entity.ParameterProviderApplyParametersRequestEntity; +import org.apache.nifi.web.api.entity.ParameterProviderEntity; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.File; +import java.io.FileInputStream; +import java.io.FileOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.file.Files; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import java.util.zip.GZIPInputStream; +import java.util.zip.GZIPOutputStream; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Reproduces the flow-synchronization failure that occurs when a provider-backed Parameter Context + * contains a parameter flagged as user-entered (provided=false) in the serialized flow. A provider-backed + * context can never hold such a parameter as a live object (it is rejected at every write path), but the + * provided flag on a serialized VersionedParameter is set independently of those guards. When a node loads + * or reconciles a flow whose provider-backed context carries a provided=false parameter, the flow + * synchronizer attempts a manual parameter update, which the Parameter Context rejects, and the node fails + * to bring its flow up. + */ +public class ClusteredProviderParamFlowSyncIT extends NiFiSystemIT { + + private static final String CONTEXT_NAME = "provider-backed-context"; + private static final String GROUP_NAME = "Parameters"; + private static final String PARAM_NAME = "db.host"; + private static final String GUARD_MESSAGE = "cannot be manually updated because they are provided by Parameter Provider"; + + private final ObjectMapper objectMapper = new ObjectMapper(); + + @Override + public NiFiInstanceFactory getInstanceFactory() { + return createTwoNodeInstanceFactory(); + } + + @Override + protected boolean isDestroyEnvironmentAfterEachTest() { + // This test intentionally corrupts the persisted flow, so the environment must not be reused. + return true; + } + + @Test + @Timeout(value = 8, unit = TimeUnit.MINUTES) + public void testProviderBackedContextWithNonProvidedParameterReconcilesOnFlowSync() throws Exception { + // 1. Create a Parameter Provider and a provider-backed Parameter Context, and apply provided parameters + // so that the parameter is provider-supplied (provided=true) on both nodes. + final ParameterProviderEntity provider = getClientUtil().createParameterProvider("PropertiesParameterProvider"); + final Map providerProps = new HashMap<>(); + providerProps.put("parameters", PARAM_NAME + "=localhost"); + getClientUtil().updateParameterProviderProperties(provider, providerProps); + + final ParameterContextEntity contextEntity = getClientUtil().createParameterContextEntity( + CONTEXT_NAME, null, Collections.emptySet(), Collections.emptyList(), + new StandardParameterProviderConfiguration(provider.getId(), GROUP_NAME, true)); + final ParameterContextEntity createdContext = getNifiClient().getParamContextClient().createParamContext(contextEntity); + getClientUtil().setParameterContext("root", createdContext); + + final ParameterGroupConfigurationEntity groupConfig = new ParameterGroupConfigurationEntity(); + groupConfig.setSynchronized(true); + groupConfig.setGroupName(GROUP_NAME); + groupConfig.setParameterContextName(CONTEXT_NAME); + final Map sensitivities = new HashMap<>(); + sensitivities.put(PARAM_NAME, ParameterSensitivity.NON_SENSITIVE); + groupConfig.setParameterSensitivities(sensitivities); + + getClientUtil().fetchParameters(provider); + final ParameterProviderApplyParametersRequestEntity applyRequest = + getClientUtil().applyParameters(provider, List.of(groupConfig)); + getClientUtil().waitForParameterProviderApplicationRequestToComplete( + applyRequest.getRequest().getParameterProvider().getId(), applyRequest.getRequest().getRequestId()); + + waitForAllNodesConnected(); + + // 2. Stop the whole cluster and corrupt the persisted flow on BOTH nodes: flip the provider-backed + // context's parameter to provided=false with a divergent value. Corrupting both nodes ensures the + // elected cluster flow carries the contradiction regardless of which node wins flow election. + final NiFiInstance node1 = getNiFiInstance().getNodeInstance(1); + final NiFiInstance node2 = getNiFiInstance().getNodeInstance(2); + node1.stop(); + node2.stop(); + + corruptProviderBackedParameter(node1); + corruptProviderBackedParameter(node2); + + // 3. Restart both nodes. With the fix, each node recognizes the context is provider-backed and reconciles + // it as provider-managed (re-sourcing values from the Parameter Provider) rather than attempting a + // manual update, so both nodes rejoin the cluster cleanly. + node1.start(true); + node2.start(true); + + // 4. The cluster forms: both nodes connect (this would time out if reconciliation still failed the guard), + // and neither node logs the provider-backed-context guard failure or a flow mismatch. + waitForAllNodesConnected(); + + final List nodes = List.of(node1, node2); + assertNoNodeLogContains(nodes, GUARD_MESSAGE); + assertNoNodeLogContains(nodes, "did not Match Cluster Flow"); + } + + private void corruptProviderBackedParameter(final NiFiInstance node) throws IOException { + final File flowFile = new File(new File(node.getInstanceDirectory(), "conf"), "flow.json.gz"); + + final byte[] bytes; + try (final InputStream fis = new FileInputStream(flowFile); + final InputStream in = new GZIPInputStream(fis); + final ByteArrayOutputStream baos = new ByteArrayOutputStream()) { + StreamUtils.copy(in, baos); + bytes = baos.toByteArray(); + } + + final VersionedDataflow dataflow; + try (final InputStream in = new ByteArrayInputStream(bytes)) { + dataflow = objectMapper.readValue(in, VersionedDataflow.class); + } + + boolean mutated = false; + if (dataflow.getParameterContexts() != null) { + for (final VersionedParameterContext context : dataflow.getParameterContexts()) { + if (!CONTEXT_NAME.equals(context.getName()) || context.getParameters() == null) { + continue; + } + for (final VersionedParameter parameter : context.getParameters()) { + if (PARAM_NAME.equals(parameter.getName())) { + parameter.setProvided(false); + parameter.setValue("user-entered-divergent-value"); + mutated = true; + } + } + } + } + assertTrue(mutated, "Expected provider-backed parameter [" + PARAM_NAME + "] in the persisted flow to corrupt"); + + try (final OutputStream fos = new FileOutputStream(flowFile); + final OutputStream gzipOut = new GZIPOutputStream(fos)) { + objectMapper.writeValue(gzipOut, dataflow); + } + } + + private void assertNoNodeLogContains(final List nodes, final String unexpected) throws IOException { + for (final NiFiInstance node : nodes) { + final File logFile = new File(new File(node.getInstanceDirectory(), "logs"), "nifi-app.log"); + if (logFile.exists()) { + assertFalse(Files.readString(logFile.toPath()).contains(unexpected), + "Node log unexpectedly contains: " + unexpected); + } + } + } +} From 2faf69ff2e0f6e852c57a866bdb5ae0bd919bc80 Mon Sep 17 00:00:00 2001 From: Pierre Villard Date: Wed, 8 Jul 2026 13:19:29 +0200 Subject: [PATCH 2/2] NIFI-16085 - Assert provider value-sourcing in reconciliation tests Address review feedback: strengthen the unit test and system IT to assert the reconciled parameter value is re-sourced from the Parameter Provider (rather than only asserting no-throw / provided=true), covering the provider value-sourcing branch end to end. --- .../VersionedFlowSynchronizerTest.java | 13 +++++++++++++ .../ClusteredProviderParamFlowSyncIT.java | 14 ++++++++++++++ 2 files changed, 27 insertions(+) diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizerTest.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizerTest.java index 7affc9df6600..7f9fe249bd23 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizerTest.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizerTest.java @@ -23,6 +23,7 @@ import org.apache.nifi.components.connector.ConnectorState; import org.apache.nifi.components.connector.ConnectorSyncMode; import org.apache.nifi.components.connector.ConnectorSyncResult; +import org.apache.nifi.components.validation.ValidationStatus; import org.apache.nifi.controller.FlowController; import org.apache.nifi.controller.ParameterProviderNode; import org.apache.nifi.controller.ReportingTaskNode; @@ -48,6 +49,7 @@ import org.apache.nifi.nar.ExtensionManager; import org.apache.nifi.parameter.Parameter; import org.apache.nifi.parameter.ParameterDescriptor; +import org.apache.nifi.parameter.ParameterGroup; import org.apache.nifi.parameter.ParameterProvider; import org.apache.nifi.parameter.ParameterReferenceManager; import org.apache.nifi.parameter.StandardParameterContext; @@ -319,6 +321,15 @@ void testSyncReconcilesProviderBackedContextWithNonProvidedParameter() { existingContext.setParameters(Collections.singletonMap(paramName, new Parameter.Builder().descriptor(descriptor).value("localhost").provided(true).build())); + // Stub the Parameter Provider as VALID and supplying the parameter, so reconciliation sources the value + // from the Provider (exercising createParameterMap's provider value-sourcing branch) rather than the + // "provided parameter not found" fallback that yields a null value. + when(parameterProviderNode.getIdentifier()).thenReturn(providerId); + when(parameterProviderNode.getValidationStatus()).thenReturn(ValidationStatus.VALID); + final ParameterGroup fetchedGroup = new ParameterGroup("Group", List.of( + new Parameter.Builder().descriptor(descriptor).value("provider-host").provided(true).build())); + when(parameterProviderNode.findFetchedParameterGroup("Group")).thenReturn(Optional.of(fetchedGroup)); + final StandardParameterContextManager contextManager = new StandardParameterContextManager(); contextManager.addParameterContext(existingContext); when(flowManager.getParameterContextManager()).thenReturn(contextManager); @@ -352,6 +363,8 @@ void testSyncReconcilesProviderBackedContextWithNonProvidedParameter() { final Optional reconciled = existingContext.getParameter(paramName); assertTrue(reconciled.isPresent(), "Parameter should still exist after reconciliation"); assertTrue(reconciled.get().isProvided(), "Parameter must be reconciled as provider-supplied (provided=true)"); + assertEquals("provider-host", reconciled.get().getValue(), + "Parameter value must be re-sourced from the Parameter Provider, not the corrupted serialized value or null"); } private void setRootGroup() { diff --git a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/parameters/ClusteredProviderParamFlowSyncIT.java b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/parameters/ClusteredProviderParamFlowSyncIT.java index e48eac66b4be..e6d987fccc2c 100644 --- a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/parameters/ClusteredProviderParamFlowSyncIT.java +++ b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/parameters/ClusteredProviderParamFlowSyncIT.java @@ -50,6 +50,7 @@ import java.util.zip.GZIPInputStream; import java.util.zip.GZIPOutputStream; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -138,6 +139,19 @@ public void testProviderBackedContextWithNonProvidedParameterReconcilesOnFlowSyn final List nodes = List.of(node1, node2); assertNoNodeLogContains(nodes, GUARD_MESSAGE); assertNoNodeLogContains(nodes, "did not Match Cluster Flow"); + + // 5. The reconciled parameter resolves to the provider-supplied value (re-sourced from the Parameter + // Provider), not the corrupted user-entered value or null. This makes the "reconciled as + // provider-managed" guarantee explicit rather than only inferring it from the absent guard message. + final ParameterContextEntity reconciledContext = + getNifiClient().getParamContextClient().getParamContext(createdContext.getId(), false); + final String reconciledValue = reconciledContext.getComponent().getParameters().stream() + .filter(entity -> PARAM_NAME.equals(entity.getParameter().getName())) + .map(entity -> entity.getParameter().getValue()) + .findFirst() + .orElse(null); + assertEquals("localhost", reconciledValue, + "Parameter [" + PARAM_NAME + "] must resolve to the provider-supplied value after reconciliation"); } private void corruptProviderBackedParameter(final NiFiInstance node) throws IOException {