kevdoran commented on code in PR #11403:
URL: https://github.com/apache/nifi/pull/11403#discussion_r3539832196
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizerTest.java:
##########
@@ -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);
Review Comment:
Minor / non-blocking: this `ParameterProviderNode` mock doesn't stub a
validation status, so `getParameterProviderValidationStatus(...)` returns
`null`, `getProvidedParameters(...)` returns an empty map, and the assertion
path ends up exercising the "provided parameter not found" branch in
`createParameterMap` (the reconciled value becomes `null`) rather than the
provider value-sourcing branch described in the comment above
`assertDoesNotThrow`. The current assertions (no-throw and `provided == true`)
still hold, but the value-sourcing path isn't actually covered. If you'd like
to lock that in, stubbing the provider node to `VALID` with a fetched parameter
group and asserting the reconciled value would exercise it end to end.
##########
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<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");
Review Comment:
Non-blocking: the IT confirms the nodes reconnect and the guard message is
absent, which is the key regression guard. As optional strengthening, asserting
the parameter's value resolves to the provider-supplied value (rather than null
or the corrupted value) after reconnection would make the "reconciled as
provider-managed" guarantee explicit.
--
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]
