This is an automated email from the ASF dual-hosted git repository.
exceptionfactory 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 dddb3dc6b2f NIFI-16196 Fixed Flaky System Test
ClusteredProviderParamFlowSyncIT (#11538)
dddb3dc6b2f is described below
commit dddb3dc6b2f42397cc3ffc5c06604136cdd12c36
Author: Pierre Villard <[email protected]>
AuthorDate: Thu Aug 13 18:12:35 2026 +0200
NIFI-16196 Fixed Flaky System Test ClusteredProviderParamFlowSyncIT (#11538)
Signed-off-by: David Handermann <[email protected]>
---
.../ClusteredProviderParamFlowSyncIT.java | 5 ++++
.../tests/system/state/AbstractStateKeyDropIT.java | 18 +++++++++++++++
.../tests/system/state/ClusterStateKeyDropIT.java | 27 ++++++++--------------
3 files changed, 32 insertions(+), 18 deletions(-)
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 c019fdbaf03..035d9a9c2f2 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
@@ -76,6 +76,11 @@ public class ClusteredProviderParamFlowSyncIT extends
NiFiSystemIT {
return createTwoNodeInstanceFactory();
}
+ @Override
+ protected boolean isAllowFactoryReuse() {
+ return false;
+ }
+
@Override
protected boolean isDestroyEnvironmentAfterEachTest() {
// This test intentionally corrupts the persisted flow, so the
environment must not be reused.
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/state/AbstractStateKeyDropIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/state/AbstractStateKeyDropIT.java
index 112f19603df..69407b2eeb5 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/state/AbstractStateKeyDropIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/state/AbstractStateKeyDropIT.java
@@ -34,6 +34,8 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
public abstract class AbstractStateKeyDropIT extends NiFiSystemIT {
private static final Logger logger =
LoggerFactory.getLogger(AbstractStateKeyDropIT.class);
@@ -82,6 +84,22 @@ public abstract class AbstractStateKeyDropIT extends
NiFiSystemIT {
}
}
+ protected void waitForProcessorState(final String processorId, final Scope
scope, final Map<String, String> expectedState)
+ throws NiFiClientException, IOException, InterruptedException {
+ final long maxTime = System.currentTimeMillis() +
TimeUnit.SECONDS.toMillis(RETRY_TIMEOUT_SECONDS);
+ Map<String, String> currentState;
+
+ do {
+ currentState = getProcessorState(processorId, scope);
+ if (expectedState.equals(currentState)) {
+ return;
+ }
+ Thread.sleep(RETRY_DELAY_MILLIS);
+ } while (System.currentTimeMillis() <= maxTime);
+
+ assertEquals(expectedState, currentState);
+ }
+
/**
* Drops the state for a given processor, optionally setting a new state.
*
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/state/ClusterStateKeyDropIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/state/ClusterStateKeyDropIT.java
index e99178a244e..5cb341dbeac 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/state/ClusterStateKeyDropIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/state/ClusterStateKeyDropIT.java
@@ -68,8 +68,7 @@ public class ClusterStateKeyDropIT extends
AbstractStateKeyDropIT {
runProcessorOnce(processor);
- final Map<String, String> currentState =
getProcessorState(processorId, Scope.CLUSTER);
- assertEquals(Map.of("a", "1", "b", "1", "c", "1"), currentState);
+ waitForProcessorState(processorId, Scope.CLUSTER, Map.of("a", "1",
"b", "1", "c", "1"));
// trying to remove key a
final Map<String, String> newState = Map.of("b", "1", "c", "1");
@@ -90,8 +89,7 @@ public class ClusterStateKeyDropIT extends
AbstractStateKeyDropIT {
runProcessorOnce(processor);
- final Map<String, String> currentState =
getProcessorState(processorId, Scope.CLUSTER);
- assertEquals(Map.of("a", "1", "b", "1", "c", "1"), currentState);
+ waitForProcessorState(processorId, Scope.CLUSTER, Map.of("a", "1",
"b", "1", "c", "1"));
// trying to remove key "a" but with wrong value for "b"
assertThrows(NiFiClientException.class, () -> {
@@ -105,8 +103,7 @@ public class ClusterStateKeyDropIT extends
AbstractStateKeyDropIT {
final String processorId = processor.getId();
runProcessorOnce(processor);
- final Map<String, String> currentState =
getProcessorState(processorId, Scope.CLUSTER);
- assertEquals(Map.of("a", "1", "b", "1", "c", "1"), currentState);
+ waitForProcessorState(processorId, Scope.CLUSTER, Map.of("a", "1",
"b", "1", "c", "1"));
// trying to remove two keys
assertThrows(NiFiClientException.class, () -> {
@@ -120,8 +117,7 @@ public class ClusterStateKeyDropIT extends
AbstractStateKeyDropIT {
final String processorId = processor.getId();
runProcessorOnce(processor);
- final Map<String, String> currentState =
getProcessorState(processorId, Scope.CLUSTER);
- assertEquals(Map.of("a", "1", "b", "1", "c", "1"), currentState);
+ waitForProcessorState(processorId, Scope.CLUSTER, Map.of("a", "1",
"b", "1", "c", "1"));
// trying to remove key a
final Map<String, String> newState = Map.of("b", "1", "c", "1");
@@ -131,8 +127,7 @@ public class ClusterStateKeyDropIT extends
AbstractStateKeyDropIT {
response.getComponentState().getClusterState().getState().forEach(entry ->
updatedState.put(entry.getKey(), entry.getValue()));
assertEquals(newState, updatedState);
- final Map<String, String> state = getProcessorState(processorId,
Scope.CLUSTER);
- assertEquals(newState, state);
+ waitForProcessorState(processorId, Scope.CLUSTER, newState);
}
@Test
@@ -141,14 +136,12 @@ public class ClusterStateKeyDropIT extends
AbstractStateKeyDropIT {
final String processorId = processor.getId();
runProcessorOnce(processor);
- final Map<String, String> currentState =
getProcessorState(processorId, Scope.CLUSTER);
- assertEquals(Map.of("a", "1", "b", "1", "c", "1"), currentState);
+ waitForProcessorState(processorId, Scope.CLUSTER, Map.of("a", "1",
"b", "1", "c", "1"));
final ComponentStateEntity response = dropProcessorState(processorId,
null);
assertTrue(response.getComponentState().getClusterState().getState().isEmpty());
- final Map<String, String> state = getProcessorState(processorId,
Scope.CLUSTER);
- assertTrue(state.isEmpty());
+ waitForProcessorState(processorId, Scope.CLUSTER, Map.of());
}
@Test
@@ -157,13 +150,11 @@ public class ClusterStateKeyDropIT extends
AbstractStateKeyDropIT {
final String processorId = processor.getId();
runProcessorOnce(processor);
- final Map<String, String> currentState =
getProcessorState(processorId, Scope.CLUSTER);
- assertEquals(Map.of("a", "1", "b", "1", "c", "1"), currentState);
+ waitForProcessorState(processorId, Scope.CLUSTER, Map.of("a", "1",
"b", "1", "c", "1"));
final ComponentStateEntity response = dropProcessorState(processorId,
Map.of());
assertTrue(response.getComponentState().getClusterState().getState().isEmpty());
- final Map<String, String> state = getProcessorState(processorId,
Scope.CLUSTER);
- assertTrue(state.isEmpty());
+ waitForProcessorState(processorId, Scope.CLUSTER, Map.of());
}
}
