This is an automated email from the ASF dual-hosted git repository.
kevdoran pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 6e7b31fa194 NIFI-16085 - Flow synchronization fails when a Parameter
Provider-backed Parameter Context contains a parameter not flagged as provided
(#11403)
6e7b31fa194 is described below
commit 6e7b31fa194280512a3866fc83c3d97ad74ba692
Author: Pierre Villard <[email protected]>
AuthorDate: Wed Jul 8 21:08:22 2026 +0200
NIFI-16085 - Flow synchronization fails when a Parameter Provider-backed
Parameter Context contains a parameter not flagged as provided (#11403)
Signed-off-by: Kevin Doran <[email protected]>
---
.../StandardVersionedComponentSynchronizer.java | 17 +-
.../parameter/TestStandardParameterContext.java | 39 ++++
.../serialization/VersionedFlowSynchronizer.java | 11 +-
.../VersionedFlowSynchronizerTest.java | 91 +++++++++
.../ClusteredProviderParamFlowSyncIT.java | 205 +++++++++++++++++++++
5 files changed, 353 insertions(+), 10 deletions(-)
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 deace1e7831..61ae3a19a2c 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 class StandardVersionedComponentSynchronizer
implements VersionedComponen
contextManager.removeParameterContext(parameterContext.getIdentifier());
LOG.info("Successfully synchronized {} by removing it from
the flow", parameterContext);
} else {
- final Map<String, Parameter> updatedParameters =
createParameterMap(proposed.getParameters());
+ final Map<String, Parameter> 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 @@ public class StandardVersionedComponentSynchronizer
implements VersionedComponen
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 @@ public class StandardVersionedComponentSynchronizer
implements VersionedComponen
final Map<String,
VersionedParameterContext> versionedParameterContexts,
final Map<String,
ParameterProviderReference> parameterProviderReferences, final
ComponentIdGenerator componentIdGenerator) {
- final Map<String, Parameter> parameters =
createParameterMap(versionedParameterContext.getParameters());
+ final Map<String, Parameter> parameters =
createParameterMap(versionedParameterContext.getParameters(),
versionedParameterContext.getParameterProvider() != null);
final List<String> parameterContextRefs = new ArrayList<>();
if (versionedParameterContext.getInheritedParameterContexts() != null)
{
@@ -2361,10 +2362,10 @@ public class StandardVersionedComponentSynchronizer
implements VersionedComponen
return contextReference.get();
}
- private Map<String, Parameter> createParameterMap(final
Collection<VersionedParameter> versionedParameters) {
+ private Map<String, Parameter> createParameterMap(final
Collection<VersionedParameter> versionedParameters, final boolean
providerBacked) {
final Map<String, Parameter> 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 @@ public class StandardVersionedComponentSynchronizer
implements VersionedComponen
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 @@ public class StandardVersionedComponentSynchronizer
implements VersionedComponen
}
}
- private Parameter createParameter(final String contextId, final
VersionedParameter versionedParameter) {
+ private Parameter createParameter(final String contextId, final
VersionedParameter versionedParameter, final boolean providerBacked) {
final List<VersionedAsset> referencedAssets =
versionedParameter.getReferencedAssets();
final List<Asset> assets;
@@ -2498,7 +2499,7 @@ public class StandardVersionedComponentSynchronizer
implements VersionedComponen
.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 992be669bab..d6d2cd32a04 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 class TestStandardParameterContext {
}
}
+ @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 29310984eb8..11a1de113df 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 @@ public class VersionedFlowSynchronizer implements
FlowSynchronizer {
) {
final Map<String, Parameter> 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<String, Parameter> 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 @@ public class VersionedFlowSynchronizer implements
FlowSynchronizer {
.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 96a1432301f..7f9fe249bd2 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,12 +23,15 @@ import
org.apache.nifi.components.connector.ConnectorRepository;
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;
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 +40,21 @@ import org.apache.nifi.flow.ScheduledState;
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.ParameterGroup;
+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 +76,16 @@ import java.nio.file.Path;
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 +292,81 @@ class VersionedFlowSynchronizerTest {
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()));
+
+ // 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);
+
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<Parameter> 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() {
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 00000000000..e6d987fccc2
--- /dev/null
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/parameters/ClusteredProviderParamFlowSyncIT.java
@@ -0,0 +1,205 @@
+/*
+ * 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.assertEquals;
+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<String, String> 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<String, ParameterSensitivity> 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<NiFiInstance> 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 {
+ 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<NiFiInstance> 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);
+ }
+ }
+ }
+}
