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 d1c7bd2ac94 NIFI-16011: Fix system-tests (#11325)
d1c7bd2ac94 is described below
commit d1c7bd2ac94c3fa3e76f36f3a9823dbb5a08d15a
Author: Mark Payne <[email protected]>
AuthorDate: Fri Jun 12 15:43:17 2026 -0400
NIFI-16011: Fix system-tests (#11325)
Make cluster node HTTP protocol version configurable and reduce
LoadBalanceIT request volume
Introduce a new nifi.properties setting,
nifi.cluster.node.protocol.http.version, that configures the HTTP
version that the cluster node web client prefers when replicating
requests to other nodes. Accepts HTTP_2 (the default) or HTTP_1_1;
invalid values log a warning and fall back to HTTP_2.
The setting is wired into FlowControllerConfiguration.webClientService()
so the cluster replication HttpClient honors the configured version.
The shipped conf/nifi.properties template defaults to HTTP_2 so
production traffic is unchanged. The system test resource templates
under nifi-system-test-suite/src/test/resources/conf/* set the value to
HTTP_1_1 to work around intermittent "RST_STREAM received Stream
cancelled" failures the JDK java.net.http.HttpClient produces when it
talks to Jetty 12.1.10 (jetty PR #15087 / issue #15009) under the heavy
disconnect / offload / restart patterns the system tests exercise. The
replicator cannot retry replicated POSTs, so when Jetty sends RST_STREAM
mid-stream the in-flight request is lost.
Also reduces the number of FlowFiles used in LoadBalanceIT from 100 to
20 so each test iterates over the queue with far fewer API calls,
reducing replication pressure and shortening the test.
Signed-off-by: Kevin Doran <[email protected]>
---
.../java/org/apache/nifi/util/NiFiProperties.java | 17 +++++++++++++++++
.../org/apache/nifi/util/NiFiPropertiesTest.java | 20 ++++++++++++++++++++
.../configuration/FlowControllerConfiguration.java | 17 +++++++++++++++++
.../nifi-framework/nifi-resources/pom.xml | 1 +
.../src/main/resources/conf/nifi.properties | 1 +
.../tests/system/loadbalance/LoadBalanceIT.java | 21 +++++++++------------
.../resources/conf/clustered/node1/nifi.properties | 5 +++++
.../resources/conf/clustered/node2/nifi.properties | 5 +++++
.../src/test/resources/conf/default/nifi.properties | 5 +++++
.../test/resources/conf/pythonic/nifi.properties | 5 +++++
10 files changed, 85 insertions(+), 12 deletions(-)
diff --git
a/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
b/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
index 2c42ca01859..67c1b35e863 100644
---
a/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
+++
b/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
@@ -247,6 +247,7 @@ public class NiFiProperties extends ApplicationProperties {
public static final String CLUSTER_NODE_ADDRESS =
"nifi.cluster.node.address";
public static final String CLUSTER_NODE_PROTOCOL_PORT =
"nifi.cluster.node.protocol.port";
public static final String CLUSTER_NODE_PROTOCOL_MAX_THREADS =
"nifi.cluster.node.protocol.max.threads";
+ public static final String CLUSTER_NODE_PROTOCOL_HTTP_VERSION =
"nifi.cluster.node.protocol.http.version";
public static final String CLUSTER_NODE_CONNECTION_TIMEOUT =
"nifi.cluster.node.connection.timeout";
public static final String CLUSTER_NODE_READ_TIMEOUT =
"nifi.cluster.node.read.timeout";
public static final String CLUSTER_NODE_MAX_CONCURRENT_REQUESTS =
"nifi.cluster.node.max.concurrent.requests";
@@ -419,6 +420,7 @@ public class NiFiProperties extends ApplicationProperties {
// cluster node defaults
public static final int DEFAULT_CLUSTER_NODE_PROTOCOL_THREADS = 10;
public static final int DEFAULT_CLUSTER_NODE_PROTOCOL_MAX_THREADS = 50;
+ public static final String DEFAULT_CLUSTER_NODE_PROTOCOL_HTTP_VERSION =
"HTTP_2";
public static final String DEFAULT_FLOW_ELECTION_MAX_WAIT_TIME = "5 mins";
// cluster load balance defaults
@@ -924,6 +926,21 @@ public class NiFiProperties extends ApplicationProperties {
}
}
+ /**
+ * Returns the configured HTTP protocol version that the cluster node web
client should prefer when communicating with other nodes.
+ * Returned as a raw String matching one of the values of {@code
java.net.http.HttpClient.Version} (i.e. {@code HTTP_1_1} or {@code HTTP_2}).
+ * Defaults to {@link #DEFAULT_CLUSTER_NODE_PROTOCOL_HTTP_VERSION} when
the property is missing or blank.
+ *
+ * @return the configured cluster node protocol HTTP version, or {@link
#DEFAULT_CLUSTER_NODE_PROTOCOL_HTTP_VERSION} when the property is missing or
blank
+ */
+ public String getClusterNodeProtocolHttpVersion() {
+ final String configured =
getProperty(CLUSTER_NODE_PROTOCOL_HTTP_VERSION);
+ if (configured == null || configured.isBlank()) {
+ return DEFAULT_CLUSTER_NODE_PROTOCOL_HTTP_VERSION;
+ }
+ return configured.trim();
+ }
+
public boolean isClustered() {
return Boolean.parseBoolean(getProperty(CLUSTER_IS_NODE));
}
diff --git
a/nifi-commons/nifi-properties/src/test/java/org/apache/nifi/util/NiFiPropertiesTest.java
b/nifi-commons/nifi-properties/src/test/java/org/apache/nifi/util/NiFiPropertiesTest.java
index 36524ed0f25..aae62e208eb 100644
---
a/nifi-commons/nifi-properties/src/test/java/org/apache/nifi/util/NiFiPropertiesTest.java
+++
b/nifi-commons/nifi-properties/src/test/java/org/apache/nifi/util/NiFiPropertiesTest.java
@@ -240,6 +240,26 @@ public class NiFiPropertiesTest {
assertThrows(RuntimeException.class, () ->
properties.getClusterNodeProtocolAddress());
}
+ @Test
+ public void testGetClusterNodeProtocolHttpVersionDefault() {
+ final NiFiProperties properties =
NiFiProperties.createBasicNiFiProperties(null, new HashMap<>());
+
assertEquals(NiFiProperties.DEFAULT_CLUSTER_NODE_PROTOCOL_HTTP_VERSION,
properties.getClusterNodeProtocolHttpVersion());
+ }
+
+ @Test
+ public void testGetClusterNodeProtocolHttpVersionOverride() {
+ final NiFiProperties properties =
NiFiProperties.createBasicNiFiProperties(null,
+ Map.of(NiFiProperties.CLUSTER_NODE_PROTOCOL_HTTP_VERSION,
"HTTP_1_1"));
+ assertEquals("HTTP_1_1",
properties.getClusterNodeProtocolHttpVersion());
+ }
+
+ @Test
+ public void testGetClusterNodeProtocolHttpVersionBlankFallsBackToDefault()
{
+ final NiFiProperties properties =
NiFiProperties.createBasicNiFiProperties(null,
+ Map.of(NiFiProperties.CLUSTER_NODE_PROTOCOL_HTTP_VERSION, "
"));
+
assertEquals(NiFiProperties.DEFAULT_CLUSTER_NODE_PROTOCOL_HTTP_VERSION,
properties.getClusterNodeProtocolHttpVersion());
+ }
+
@Test
public void testShouldHaveReasonableMaxContentLengthValues() {
// Arrange with default values:
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/framework/configuration/FlowControllerConfiguration.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/framework/configuration/FlowControllerConfiguration.java
index 63aff7636a6..88a2e25c93d 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/framework/configuration/FlowControllerConfiguration.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/framework/configuration/FlowControllerConfiguration.java
@@ -74,11 +74,14 @@ import org.apache.nifi.web.client.api.WebClientService;
import org.apache.nifi.web.client.redirect.RedirectHandling;
import org.apache.nifi.web.client.ssl.TlsContext;
import org.apache.nifi.web.revision.RevisionManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
+import java.net.http.HttpClient;
import java.time.Duration;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
@@ -92,6 +95,8 @@ import javax.net.ssl.X509TrustManager;
@Configuration
public class FlowControllerConfiguration {
+ private static final Logger logger =
LoggerFactory.getLogger(FlowControllerConfiguration.class);
+
private static final String FLOW_ACTION_REPORTER_IMPLEMENTATION =
"nifi.flow.action.reporter.implementation";
private static final String COMPONENT_METRIC_REPORTER_IMPLEMENTATION =
"nifi.component.metric.reporter.implementation";
@@ -386,6 +391,7 @@ public class FlowControllerConfiguration {
webClientService.setConnectTimeout(timeout);
webClientService.setReadTimeout(timeout);
webClientService.setRedirectHandling(RedirectHandling.FOLLOWED);
+
webClientService.setHttpVersion(resolveClusterNodeProtocolHttpVersion());
if (sslContext != null) {
webClientService.setTlsContext(new TlsContext() {
@@ -409,6 +415,17 @@ public class FlowControllerConfiguration {
return webClientService;
}
+ private HttpClient.Version resolveClusterNodeProtocolHttpVersion() {
+ final String configured =
properties.getClusterNodeProtocolHttpVersion();
+ try {
+ return HttpClient.Version.valueOf(configured);
+ } catch (final IllegalArgumentException e) {
+ logger.warn("Property {} value [{}] is not a valid
HttpClient.Version; falling back to {}",
+ NiFiProperties.CLUSTER_NODE_PROTOCOL_HTTP_VERSION,
configured, NiFiProperties.DEFAULT_CLUSTER_NODE_PROTOCOL_HTTP_VERSION);
+ return
HttpClient.Version.valueOf(NiFiProperties.DEFAULT_CLUSTER_NODE_PROTOCOL_HTTP_VERSION);
+ }
+ }
+
@Bean
public NarPersistenceProviderFactoryBean narPersistenceProvider() {
return new NarPersistenceProviderFactoryBean(properties,
extensionManager);
diff --git a/nifi-framework-bundle/nifi-framework/nifi-resources/pom.xml
b/nifi-framework-bundle/nifi-framework/nifi-resources/pom.xml
index a62a5c702ab..450c4d2d1f2 100644
--- a/nifi-framework-bundle/nifi-framework/nifi-resources/pom.xml
+++ b/nifi-framework-bundle/nifi-framework/nifi-resources/pom.xml
@@ -204,6 +204,7 @@
<nifi.cluster.node.address />
<nifi.cluster.node.protocol.port />
<nifi.cluster.node.protocol.max.threads>50</nifi.cluster.node.protocol.max.threads>
+
<nifi.cluster.node.protocol.http.version>HTTP_2</nifi.cluster.node.protocol.http.version>
<nifi.cluster.node.event.history.size>25</nifi.cluster.node.event.history.size>
<nifi.cluster.node.connection.timeout>5
sec</nifi.cluster.node.connection.timeout>
<nifi.cluster.node.read.timeout>5 sec</nifi.cluster.node.read.timeout>
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/nifi.properties
b/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/nifi.properties
index fba5d1be311..25a3b53173a 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/nifi.properties
+++
b/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/nifi.properties
@@ -268,6 +268,7 @@
nifi.cluster.leader.election.implementation=${nifi.cluster.leader.election.imple
nifi.cluster.node.address=${nifi.cluster.node.address}
nifi.cluster.node.protocol.port=${nifi.cluster.node.protocol.port}
nifi.cluster.node.protocol.max.threads=${nifi.cluster.node.protocol.max.threads}
+nifi.cluster.node.protocol.http.version=${nifi.cluster.node.protocol.http.version}
nifi.cluster.node.event.history.size=${nifi.cluster.node.event.history.size}
nifi.cluster.node.connection.timeout=${nifi.cluster.node.connection.timeout}
nifi.cluster.node.read.timeout=${nifi.cluster.node.read.timeout}
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/loadbalance/LoadBalanceIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/loadbalance/LoadBalanceIT.java
index 1fac92c880c..3a07c105722 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/loadbalance/LoadBalanceIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/loadbalance/LoadBalanceIT.java
@@ -167,25 +167,23 @@ public class LoadBalanceIT extends NiFiSystemIT {
final ConnectionEntity connection =
getClientUtil().createConnection(generate, count, "success");
getClientUtil().setAutoTerminatedRelationships(count, "success");
- // Configure Processor to generate 10 FlowFiles, each 1 MB, on each
node, for a total of 20 FlowFiles.
+ final int batchSize = 2;
+ final int distinctAttributeValues = 10;
+ final int expectedTotalFlowFiles = batchSize * distinctAttributeValues;
+
final Map<String, String> generateProperties = new HashMap<>();
generateProperties.put("File Size", "1 MB");
- generateProperties.put("Batch Size", "10");
+ generateProperties.put("Batch Size", String.valueOf(batchSize));
generateProperties.put("number", "0");
getClientUtil().updateProcessorProperties(generate,
generateProperties);
getClientUtil().updateProcessorExecutionNode(generate,
ExecutionNode.PRIMARY);
- // Round Robin between nodes. This should result in 10 FlowFiles on
each node.
getClientUtil().updateConnectionLoadBalancing(connection,
LoadBalanceStrategy.PARTITION_BY_ATTRIBUTE,
LoadBalanceCompression.DO_NOT_COMPRESS, "number");
- // Queue 100 FlowFiles. 10 with number=0, 10 with number=1, 10 with
number=2, etc. to up 10 with number=9
- for (int i = 1; i <= 10; i++) {
- // Generate the data.
+ for (int i = 1; i <= distinctAttributeValues; i++) {
getClientUtil().startProcessor(generate);
- final int expectedQueueSize = 10 * i;
-
- // Wait until all 10 FlowFiles are queued up.
+ final int expectedQueueSize = batchSize * i;
waitFor(() -> {
final ConnectionStatusEntity statusEntity =
getConnectionStatus(connection.getId());
return
statusEntity.getConnectionStatus().getAggregateSnapshot().getFlowFilesQueued()
== expectedQueueSize;
@@ -198,18 +196,17 @@ public class LoadBalanceIT extends NiFiSystemIT {
getClientUtil().updateProcessorProperties(generate,
generateProperties);
}
- // Wait until load balancing is complete
waitFor(() -> isConnectionDoneLoadBalancing(connection.getId()));
final Map<String, Set<String>> nodesByAttribute = new HashMap<>();
- for (int i = 0; i < 100; i++) {
+ for (int i = 0; i < expectedTotalFlowFiles; i++) {
final FlowFileEntity flowFile =
getClientUtil().getQueueFlowFile(connection.getId(), i);
final String numberValue =
flowFile.getFlowFile().getAttributes().get("number");
final Set<String> nodes =
nodesByAttribute.computeIfAbsent(numberValue, key -> new HashSet<>());
nodes.add(flowFile.getFlowFile().getClusterNodeId());
}
- assertEquals(10, nodesByAttribute.size());
+ assertEquals(distinctAttributeValues, nodesByAttribute.size());
for (final Map.Entry<String, Set<String>> entry :
nodesByAttribute.entrySet()) {
final Set<String> nodes = entry.getValue();
assertEquals(1, nodes.size(), "FlowFile with attribute number=" +
entry.getKey() + " went to nodes " + nodes);
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/nifi.properties
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/nifi.properties
index 9282db83eb4..438eff18a7c 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/nifi.properties
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/nifi.properties
@@ -200,6 +200,11 @@ nifi.cluster.node.address=
nifi.cluster.node.protocol.port=48101
nifi.cluster.node.protocol.threads=10
nifi.cluster.node.protocol.max.threads=50
+# Short-term workaround: force HTTP/1.1 for cluster node-to-node traffic in
system tests to avoid the
+# intermittent "RST_STREAM received Stream cancelled" failures caused by Jetty
12.1.10's HTTP/2 stream
+# reset rewrite (https://github.com/jetty/jetty.project/pull/15087). Remove
this override once the
+# upstream Jetty regression is resolved.
+nifi.cluster.node.protocol.http.version=HTTP_1_1
nifi.cluster.node.event.history.size=25
nifi.cluster.node.connection.timeout=5 sec
nifi.cluster.node.read.timeout=30 sec
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/nifi.properties
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/nifi.properties
index d843563c423..60a837363f6 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/nifi.properties
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/nifi.properties
@@ -200,6 +200,11 @@ nifi.cluster.node.address=
nifi.cluster.node.protocol.port=48102
nifi.cluster.node.protocol.threads=10
nifi.cluster.node.protocol.max.threads=50
+# Short-term workaround: force HTTP/1.1 for cluster node-to-node traffic in
system tests to avoid the
+# intermittent "RST_STREAM received Stream cancelled" failures caused by Jetty
12.1.10's HTTP/2 stream
+# reset rewrite (https://github.com/jetty/jetty.project/pull/15087). Remove
this override once the
+# upstream Jetty regression is resolved.
+nifi.cluster.node.protocol.http.version=HTTP_1_1
nifi.cluster.node.event.history.size=25
nifi.cluster.node.connection.timeout=5 sec
nifi.cluster.node.read.timeout=30 sec
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/nifi.properties
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/nifi.properties
index d55f0a2d4fa..b2032eadbde 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/nifi.properties
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/nifi.properties
@@ -201,6 +201,11 @@ nifi.cluster.node.address=
nifi.cluster.node.protocol.port=
nifi.cluster.node.protocol.threads=10
nifi.cluster.node.protocol.max.threads=50
+# Short-term workaround: force HTTP/1.1 for cluster node-to-node traffic in
system tests to avoid the
+# intermittent "RST_STREAM received Stream cancelled" failures caused by Jetty
12.1.10's HTTP/2 stream
+# reset rewrite (https://github.com/jetty/jetty.project/pull/15087). Remove
this override once the
+# upstream Jetty regression is resolved.
+nifi.cluster.node.protocol.http.version=HTTP_1_1
nifi.cluster.node.event.history.size=25
nifi.cluster.node.connection.timeout=5 sec
nifi.cluster.node.read.timeout=5 sec
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/pythonic/nifi.properties
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/pythonic/nifi.properties
index bdd285d4bfd..d9d0f283c97 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/pythonic/nifi.properties
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/pythonic/nifi.properties
@@ -205,6 +205,11 @@ nifi.cluster.node.address=
nifi.cluster.node.protocol.port=
nifi.cluster.node.protocol.threads=10
nifi.cluster.node.protocol.max.threads=50
+# Short-term workaround: force HTTP/1.1 for cluster node-to-node traffic in
system tests to avoid the
+# intermittent "RST_STREAM received Stream cancelled" failures caused by Jetty
12.1.10's HTTP/2 stream
+# reset rewrite (https://github.com/jetty/jetty.project/pull/15087). Remove
this override once the
+# upstream Jetty regression is resolved.
+nifi.cluster.node.protocol.http.version=HTTP_1_1
nifi.cluster.node.event.history.size=25
nifi.cluster.node.connection.timeout=5 sec
nifi.cluster.node.read.timeout=5 sec
