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 0887d8e6717 NIFI-15918 Stabilized flaky system tests around Processor
lifecycle (#11223)
0887d8e6717 is described below
commit 0887d8e67175894680b4568d7c07657f99258715
Author: Mark Payne <[email protected]>
AuthorDate: Thu May 14 12:42:29 2026 -0400
NIFI-15918 Stabilized flaky system tests around Processor lifecycle (#11223)
- Defensive copy in LifecycleState.getFutures() to prevent
ConcurrentModificationException.
- Ensure StandardProcessorNode.stop() decrements active threads and
completes the
stop action even if the stop task throws.
- Promote silent processor start abort paths to INFO logs to aid diagnosis.
- Null-safe shutdown in KubernetesConfigMapStateProvider when
initialization was skipped.
- NiFiClientUtil.runProcessorOnce waits for validation completion and fails
fast on
INVALID; system tests updated to use it where appropriate.
- ContentClaimTruncationIT: keep one small FlowFile queued so its resource
claim is
pinned during replay, preventing archive-cleanup-induced truncation
flakes.
Signed-off-by: David Handermann <[email protected]>
---
.cursor/rules/code-style.mdc | 8 +++
.../rules/system-test-troubleshooting-archive.mdc | 77 ++++++++++++++++++++++
.../provider/KubernetesConfigMapStateProvider.java | 10 ++-
.../KubernetesConfigMapStateProviderTest.java | 5 ++
.../nifi/controller/StandardProcessorNode.java | 31 +++++++--
.../nifi/controller/scheduling/LifecycleState.java | 3 +-
.../apache/nifi/tests/system/NiFiClientUtil.java | 36 ++++++++++
.../system/metrics/ComponentMetricReporterIT.java | 2 +-
.../system/nar/NarProviderAndAutoLoaderIT.java | 2 +-
.../nifi/tests/system/processor/RetryIT.java | 2 +-
.../nifi/tests/system/processor/RunOnceIT.java | 4 +-
.../tests/system/python/PythonProcessorIT.java | 2 +-
.../ContentClaimTruncationAfterRestartIT.java | 2 +-
.../repositories/ContentClaimTruncationIT.java | 35 ++++++----
.../ContentClaimTruncationWithSwappingIT.java | 8 +--
.../OffloadContentClaimTruncationIT.java | 2 +-
.../tests/system/state/AbstractStateKeyDropIT.java | 2 +-
17 files changed, 195 insertions(+), 36 deletions(-)
diff --git a/.cursor/rules/code-style.mdc b/.cursor/rules/code-style.mdc
index ca7e9cfed5a..f1d22d36391 100644
--- a/.cursor/rules/code-style.mdc
+++ b/.cursor/rules/code-style.mdc
@@ -72,3 +72,11 @@ final List<String> result = myList.stream()
when the logic is not simple and straightforward. The stream API is
powerful but can be difficult to
read when overused or used in complex scenarios. Functional style is best
used when the logic is simple
and chains together no more than 3-4 operations.
+16. Never use single-element arrays such as `final int[] counter = new
int[]{0}` or
+ `final boolean[] flag = new boolean[]{false}` as a workaround for
capturing a mutable primitive in a
+ lambda or anonymous class. This pattern is unidiomatic and confusing. When
a lambda needs to mutate a
+ captured primitive, use the appropriate `java.util.concurrent.atomic` type
(`AtomicInteger`,
+ `AtomicLong`, `AtomicBoolean`, `AtomicReference`, etc.). When a lambda
needs to mutate a captured
+ object, hold the object in a final reference and mutate the object's
state. The same applies to
+ `final Object[] holder = new Object[1]` for capturing a single reference;
use `AtomicReference`
+ instead.
diff --git a/.cursor/rules/system-test-troubleshooting-archive.mdc
b/.cursor/rules/system-test-troubleshooting-archive.mdc
new file mode 100644
index 00000000000..c0ded81a03d
--- /dev/null
+++ b/.cursor/rules/system-test-troubleshooting-archive.mdc
@@ -0,0 +1,77 @@
+---
+description: Explains that every CI run of NiFi system tests uploads a
per-test troubleshooting archive containing each cluster node's logs/ and conf/
directories, and how to download it when investigating a system test failure.
+alwaysApply: true
+---
+
+# System Test Troubleshooting Archive
+
+When a NiFi system test (anything under `nifi-system-tests/`) fails in CI, the
+`system-tests` workflow uploads a `*-troubleshooting-logs` artifact for each
+OS/JDK combination that ran. This archive is far more useful than the surefire/
+failsafe console output alone and **must** be retrieved before drawing
+conclusions about a system test failure.
+
+## What the archive contains
+
+For every failed test in that job, the archive includes:
+
+```
+troubleshooting/
+ <TestClass>-<testMethod>/
+ node-1/
+ logs/ full nifi-app.log, nifi-bootstrap.log, etc. for that node
+ conf/ nifi.properties, flow definition, state-management.xml, etc.
+ node-2/
+ logs/
+ conf/
+ ...
+```
+
+That is, for each failed test, the running NiFi instance(s) used by the test
+are captured with the same `logs/` and `conf/` you would inspect on a real
+deployment. For clustered tests, every node is captured separately. The
+archive also typically contains the matching `failsafe-reports/`.
+
+This is critical because a system test failure surfaces in CI as a single
+exception message, but the actual root cause is almost always visible only in
+`nifi-app.log` of one of the cluster nodes (cluster join issues, replication
+errors, repository corruption, processor scheduling issues, etc.).
+
+## How to fetch the archive
+
+1. Identify the failing GitHub Actions run id. For a PR, list its checks:
+ ```
+ gh pr checks <PR_NUMBER> --repo apache/nifi
+ gh run list --repo apache/nifi --limit 30
+ ```
+2. List the artifacts attached to that run to see which OS/JDK combos are
+ available and confirm none have expired:
+ ```
+ gh api repos/apache/nifi/actions/runs/<RUN_ID>/artifacts \
+ --jq '.artifacts[] | "\(.id)\t\(.name)\t\(.size_in_bytes)\t\(.expired)"'
+ ```
+3. Download into a directory:
+ ```
+ gh run download <RUN_ID> --repo apache/nifi --dir <local-dir>
+ ```
+ Each `*-troubleshooting-logs` artifact becomes a subdirectory under
+ `<local-dir>`.
+4. Inspect the per-node logs for the specific failing test:
+ ```
+
<local-dir>/<os-jdk>-troubleshooting-logs/troubleshooting/<TestClass>-<testMethod>/node-1/logs/nifi-app.log
+ ```
+
+GitHub Actions artifacts expire (currently 7 days for these workflows), so
+fetch them as soon as a failure is observed; once expired they cannot be
+recovered.
+
+## When this rule applies
+
+- Investigating any failed test under `nifi-system-tests/` in CI.
+- Triaging suspected flaky system tests across multiple PRs.
+- Reviewing a PR whose `system-tests` check has failed.
+
+It does **not** apply to unit tests (surefire) or integration tests
+(failsafe outside `nifi-system-tests/`); those workflows do not produce a
+troubleshooting archive, and only the surefire/failsafe console output is
+available via `gh run view <RUN_ID> --log-failed`.
diff --git
a/nifi-framework-bundle/nifi-framework-extensions/nifi-framework-kubernetes-bundle/nifi-framework-kubernetes-state-provider/src/main/java/org/apache/nifi/kubernetes/state/provider/KubernetesConfigMapStateProvider.java
b/nifi-framework-bundle/nifi-framework-extensions/nifi-framework-kubernetes-bundle/nifi-framework-kubernetes-state-provider/src/main/java/org/apache/nifi/kubernetes/state/provider/KubernetesConfigMapStateProvider.java
index 9bd5f0e00b1..ed52c3432d8 100644
---
a/nifi-framework-bundle/nifi-framework-extensions/nifi-framework-kubernetes-bundle/nifi-framework-kubernetes-state-provider/src/main/java/org/apache/nifi/kubernetes/state/provider/KubernetesConfigMapStateProvider.java
+++
b/nifi-framework-bundle/nifi-framework-extensions/nifi-framework-kubernetes-bundle/nifi-framework-kubernetes-state-provider/src/main/java/org/apache/nifi/kubernetes/state/provider/KubernetesConfigMapStateProvider.java
@@ -137,12 +137,16 @@ public class KubernetesConfigMapStateProvider extends
AbstractConfigurableCompon
}
/**
- * Shutdown Provider
+ * Shutdown Provider. Safe to invoke even when the Provider was never
initialized.
*/
@Override
public void shutdown() {
- kubernetesClient.close();
- logger.info("Provider shutdown");
+ if (kubernetesClient != null) {
+ kubernetesClient.close();
+ }
+ if (logger != null) {
+ logger.info("Provider shutdown");
+ }
}
/**
diff --git
a/nifi-framework-bundle/nifi-framework-extensions/nifi-framework-kubernetes-bundle/nifi-framework-kubernetes-state-provider/src/test/java/org/apache/nifi/kubernetes/state/provider/KubernetesConfigMapStateProviderTest.java
b/nifi-framework-bundle/nifi-framework-extensions/nifi-framework-kubernetes-bundle/nifi-framework-kubernetes-state-provider/src/test/java/org/apache/nifi/kubernetes/state/provider/KubernetesConfigMapStateProviderTest.java
index 7ef21176820..3574a7898f6 100644
---
a/nifi-framework-bundle/nifi-framework-extensions/nifi-framework-kubernetes-bundle/nifi-framework-kubernetes-state-provider/src/test/java/org/apache/nifi/kubernetes/state/provider/KubernetesConfigMapStateProviderTest.java
+++
b/nifi-framework-bundle/nifi-framework-extensions/nifi-framework-kubernetes-bundle/nifi-framework-kubernetes-state-provider/src/test/java/org/apache/nifi/kubernetes/state/provider/KubernetesConfigMapStateProviderTest.java
@@ -131,6 +131,11 @@ class KubernetesConfigMapStateProviderTest {
provider.shutdown();
}
+ @Test
+ void testShutdownWithoutInitialize() {
+ provider.shutdown();
+ }
+
@Test
void testInitializeEnableDisable() {
setContext();
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java
index 11c60f38115..6fcea33e04a 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java
@@ -1522,6 +1522,8 @@ public class StandardProcessorNode extends ProcessorNode
implements Connectable
} else {
final String procName =
processorRef.get().getProcessor().toString();
procLog.warn("Cannot start {} because it is not currently stopped.
Current state is {}", procName, currentState);
+ LOG.info("Cannot start {}: current scheduledState={}, current
desiredState={}, requested scheduledState={}, requested desiredState={}",
+ this, currentState, getDesiredState(), scheduledState,
desiredState);
}
}
@@ -1651,7 +1653,8 @@ public class StandardProcessorNode extends ProcessorNode
implements Connectable
final Callable<Void> startupTask = () -> {
final ScheduledState currentScheduleState = scheduledState.get();
if (currentScheduleState == ScheduledState.STOPPING ||
currentScheduleState == ScheduledState.STOPPED || getDesiredState() ==
ScheduledState.STOPPED) {
- LOG.debug("{} is stopped. Will not call @OnScheduled lifecycle
methods or begin trigger onTrigger() method", StandardProcessorNode.this);
+ LOG.info("Aborting start of {}: scheduledState={},
desiredState={}, validationStatus={}",
+ StandardProcessorNode.this, currentScheduleState,
getDesiredState(), getValidationStatus());
schedulingAgentCallback.onTaskComplete();
completeStopAction();
return null;
@@ -1668,13 +1671,17 @@ public class StandardProcessorNode extends
ProcessorNode implements Connectable
return null;
}
- LOG.debug("Cannot start {} because Processor is currently not
valid; will try again after 5 seconds", StandardProcessorNode.this);
-
final long attempt = startupAttemptCount.getAndIncrement();
- if (attempt % 7200 == 0) {
+ if (attempt == 0) {
+ final ValidationState validationState =
getValidationState();
+ LOG.info("Cannot start {} because Processor is currently
not valid (Validation State is {}: {}). Will continue trying to start.",
+ StandardProcessorNode.this, validationState,
validationState.getValidationErrors());
+ } else if (attempt % 7200 == 0) {
final ValidationState validationState =
getValidationState();
procLog.warn("Encountering difficulty starting.
(Validation State is {}: {}). Will continue trying to start.",
validationState,
validationState.getValidationErrors());
+ } else {
+ LOG.debug("Cannot start {} because Processor is currently
not valid; will try again after 500 ms", StandardProcessorNode.this);
}
// re-initiate the entire process
@@ -1858,6 +1865,7 @@ public class StandardProcessorNode extends ProcessorNode
implements Connectable
executor.execute(new Runnable() {
@Override
public void run() {
+ boolean cleanupHandled = false;
try {
if (lifecycleState.isScheduled()) {
schedulingAgent.unschedule(StandardProcessorNode.this, lifecycleState);
@@ -1883,6 +1891,7 @@ public class StandardProcessorNode extends ProcessorNode
implements Connectable
LOG.debug("Will not trigger @OnStopped
methods of {} because ProcessorStopLifecycleMethods.isTriggerOnStopped() =
false", this);
}
} finally {
+ cleanupHandled = true;
lifecycleState.decrementActiveThreadCount();
completeStopAction();
@@ -1912,13 +1921,27 @@ public class StandardProcessorNode extends
ProcessorNode implements Connectable
// stop action and exit. completeStopAction() is
idempotent if procNode.terminate() already
// invoked it.
LOG.debug("Stop sequence for {} aborted because
LifecycleState was terminated", this);
+ cleanupHandled = true;
completeStopAction();
} else {
// Not all of the active threads have finished.
Try again in 100 milliseconds.
executor.schedule(this, 100,
TimeUnit.MILLISECONDS);
+ cleanupHandled = true;
}
} catch (final Exception e) {
LOG.warn("Failed while shutting down processor {}",
processor, e);
+
+ // If an exception escaped before the normal
completion path ran (for example because
+ // schedulingAgent.unschedule or an @OnUnscheduled
method threw), the active thread count
+ // increment performed at the top of stop() must still
be reversed and the stop future must
+ // still be completed. Otherwise the processor remains
permanently in STOPPING.
+ if (!cleanupHandled) {
+ try {
+ lifecycleState.decrementActiveThreadCount();
+ } finally {
+ completeStopAction();
+ }
+ }
}
}
});
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/scheduling/LifecycleState.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/scheduling/LifecycleState.java
index a3095a86ea1..3a30b189c71 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/scheduling/LifecycleState.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/scheduling/LifecycleState.java
@@ -23,7 +23,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Collection;
-import java.util.Collections;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
@@ -151,7 +150,7 @@ public class LifecycleState {
}
public synchronized Set<ScheduledFuture<?>> getFutures() {
- return Collections.unmodifiableSet(futures);
+ return Set.copyOf(futures);
}
public synchronized void terminate() {
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiClientUtil.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiClientUtil.java
index 2de5c0f915e..8960e202ce3 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiClientUtil.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiClientUtil.java
@@ -187,6 +187,30 @@ public class NiFiClientUtil {
return getProcessorClient().startProcessor(currentEntity);
}
+ /**
+ * Runs the given Processor exactly one time. Waits for the Processor's
validation to settle before issuing the
+ * run request so that recent modifications to the Processor, its inbound
or outbound Connections, or any
+ * referenced Controller Service are reflected in the Processor's
validation status. If validation settles to
+ * INVALID, an {@link IllegalStateException} is thrown immediately rather
than waiting indefinitely; the toolkit
+ * client's run-once endpoint silently does nothing for an invalid
Processor, so failing fast prevents flaky
+ * downstream assertions about FlowFiles that were never produced.
+ *
+ * @param currentEntity the Processor to run once
+ * @return the updated Processor entity returned by the run-once API call
+ */
+ public ProcessorEntity runProcessorOnce(final ProcessorEntity
currentEntity) throws NiFiClientException, IOException, InterruptedException {
+ waitForValidationCompleted(currentEntity);
+
+ final ProcessorEntity refreshed =
getProcessorClient().getProcessor(currentEntity.getId());
+ final String validationStatus =
refreshed.getComponent().getValidationStatus();
+ if (ProcessorDTO.INVALID.equalsIgnoreCase(validationStatus)) {
+ throw new IllegalStateException(String.format("Processor %s is
INVALID and cannot be run once. Validation errors: %s",
+ currentEntity.getId(),
refreshed.getComponent().getValidationErrors()));
+ }
+
+ return getProcessorClient().runProcessorOnce(currentEntity);
+ }
+
public void stopProcessor(final ProcessorEntity currentEntity) throws
NiFiClientException, IOException, InterruptedException {
currentEntity.setDisconnectedNodeAcknowledged(true);
getProcessorClient().stopProcessor(currentEntity);
@@ -1154,9 +1178,15 @@ public class NiFiClientUtil {
final long maxTimestamp = System.currentTimeMillis() +
TimeUnit.MINUTES.toMillis(2);
logger.info("Waiting for Processor {} to reach state {}", processorId,
expectedState);
+ String lastObservedState = null;
+ String lastObservedPhysicalState = null;
+ Integer lastObservedActiveThreadCount = null;
+ Integer lastObservedTerminatedThreadCount = null;
+
while (System.currentTimeMillis() < maxTimestamp) {
final ProcessorEntity entity =
getProcessorClient().getProcessor(processorId);
final String state = entity.getComponent().getState();
+ lastObservedState = state;
// We've reached the desired state if the state equal the expected
state, OR if we expect stopped and the state is disabled (because disabled
implies stopped)
final boolean desiredStateReached = expectedState.equals(state) ||
("STOPPED".equalsIgnoreCase(expectedState) &&
"DISABLED".equalsIgnoreCase(state));
@@ -1169,6 +1199,8 @@ public class NiFiClientUtil {
final ProcessorStatusSnapshotDTO snapshotDto =
entity.getStatus().getAggregateSnapshot();
final Integer activeThreadCount =
snapshotDto.getActiveThreadCount();
final Integer terminatedThreadCount =
snapshotDto.getTerminatedThreadCount();
+ lastObservedActiveThreadCount = activeThreadCount;
+ lastObservedTerminatedThreadCount = terminatedThreadCount;
if ("RUNNING".equals(expectedState) || (activeThreadCount == 0 &&
terminatedThreadCount == 0)) {
// The logical state masks the framework's physical STOPPING
state as STOPPED. The framework's
@@ -1179,6 +1211,7 @@ public class NiFiClientUtil {
// it is not stopped. Current state is STOPPING".
if ("STOPPED".equalsIgnoreCase(expectedState)) {
final String physicalState =
entity.getComponent().getPhysicalState();
+ lastObservedPhysicalState = physicalState;
final boolean physicalStateSettled = physicalState == null
|| "STOPPED".equalsIgnoreCase(physicalState)
|| "DISABLED".equalsIgnoreCase(physicalState);
@@ -1195,6 +1228,9 @@ public class NiFiClientUtil {
Thread.sleep(10L);
}
+
+ throw new IOException(String.format("Timed out waiting for Processor
%s to reach state of %s. Last observed state=%s, physicalState=%s,
activeThreadCount=%s, terminatedThreadCount=%s",
+ processorId, expectedState, lastObservedState,
lastObservedPhysicalState, lastObservedActiveThreadCount,
lastObservedTerminatedThreadCount));
}
public ReportingTaskEntity waitForReportingTaskState(final String
reportingTaskId, final String expectedState) throws NiFiClientException,
IOException, InterruptedException {
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/metrics/ComponentMetricReporterIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/metrics/ComponentMetricReporterIT.java
index fa13bbbeac6..ca06c2f6e4c 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/metrics/ComponentMetricReporterIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/metrics/ComponentMetricReporterIT.java
@@ -48,7 +48,7 @@ public class ComponentMetricReporterIT extends NiFiSystemIT {
@Test
void testUpdateMetricReported() throws NiFiClientException, IOException,
InterruptedException {
final ProcessorEntity updateMetric =
getClientUtil().createProcessor("UpdateMetric");
- getNifiClient().getProcessorClient().runProcessorOnce(updateMetric);
+ getClientUtil().runProcessorOnce(updateMetric);
getClientUtil().waitForStoppedProcessor(updateMetric.getId());
final String componentId = updateMetric.getId();
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/nar/NarProviderAndAutoLoaderIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/nar/NarProviderAndAutoLoaderIT.java
index 05ad540095e..95d6fda0f99 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/nar/NarProviderAndAutoLoaderIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/nar/NarProviderAndAutoLoaderIT.java
@@ -71,7 +71,7 @@ public class NarProviderAndAutoLoaderIT extends NiFiSystemIT {
final ProcessorEntity terminateFlowFile =
getClientUtil().createProcessor("TerminateFlowFile");
final ConnectionEntity connection =
getClientUtil().createConnection(updatedGetClassLoaderInfo, terminateFlowFile,
"success");
-
getNifiClient().getProcessorClient().runProcessorOnce(updatedGetClassLoaderInfo);
+ getClientUtil().runProcessorOnce(updatedGetClassLoaderInfo);
waitForQueueCount(connection.getId(), 1);
final String flowFileContent =
getClientUtil().getFlowFileContentAsUtf8(connection.getId(), 0);
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/processor/RetryIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/processor/RetryIT.java
index f4a488e15e0..25feb45a4f5 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/processor/RetryIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/processor/RetryIT.java
@@ -160,7 +160,7 @@ public class RetryIT extends NiFiSystemIT {
}
private void runProcessorOnce(final ProcessorEntity processorEntity)
throws NiFiClientException, IOException, InterruptedException {
- getNifiClient().getProcessorClient().runProcessorOnce(processorEntity);
+ getClientUtil().runProcessorOnce(processorEntity);
getClientUtil().waitForStoppedProcessor(processorEntity.getId());
}
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/processor/RunOnceIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/processor/RunOnceIT.java
index 0409c4a416b..4fdc111e40f 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/processor/RunOnceIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/processor/RunOnceIT.java
@@ -34,7 +34,7 @@ public class RunOnceIT extends NiFiSystemIT {
final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile");
final ConnectionEntity generateToTerminate =
getClientUtil().createConnection(generate, terminate, "success");
- getNifiClient().getProcessorClient().runProcessorOnce(generate);
+ getClientUtil().runProcessorOnce(generate);
waitForQueueCount(generateToTerminate.getId(), 1);
getClientUtil().waitForStoppedProcessor(generate.getId());
@@ -44,7 +44,7 @@ public class RunOnceIT extends NiFiSystemIT {
getClientUtil().updateProcessorSchedulingStrategy(generate,
"CRON_DRIVEN");
getClientUtil().updateProcessorSchedulingPeriod(generate, "* * * * *
?");
- getNifiClient().getProcessorClient().runProcessorOnce(generate);
+ getClientUtil().runProcessorOnce(generate);
waitForQueueCount(generateToTerminate.getId(), 2);
getClientUtil().waitForStoppedProcessor(generate.getId());
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/python/PythonProcessorIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/python/PythonProcessorIT.java
index d4ea775a49d..520b1bc7a0c 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/python/PythonProcessorIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/python/PythonProcessorIT.java
@@ -287,7 +287,7 @@ public class PythonProcessorIT extends NiFiSystemIT {
}
private void runProcessorOnce(final ProcessorEntity processorEntity)
throws NiFiClientException, IOException, InterruptedException {
- getNifiClient().getProcessorClient().runProcessorOnce(processorEntity);
+ getClientUtil().runProcessorOnce(processorEntity);
getClientUtil().waitForStoppedProcessor(processorEntity.getId());
}
}
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/repositories/ContentClaimTruncationAfterRestartIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/repositories/ContentClaimTruncationAfterRestartIT.java
index 012739ca872..c236e604c05 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/repositories/ContentClaimTruncationAfterRestartIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/repositories/ContentClaimTruncationAfterRestartIT.java
@@ -113,7 +113,7 @@ public class ContentClaimTruncationAfterRestartIT extends
NiFiSystemIT {
// FlowFiles (priority=1) are dequeued first.
for (int i = 0; i < 10; i++) {
final ProcessorEntity terminateAfterRestart =
getNifiClient().getProcessorClient().getProcessor(terminate.getId());
-
getNifiClient().getProcessorClient().runProcessorOnce(terminateAfterRestart);
+ getClientUtil().runProcessorOnce(terminateAfterRestart);
getClientUtil().waitForStoppedProcessor(terminateAfterRestart.getId());
}
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/repositories/ContentClaimTruncationIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/repositories/ContentClaimTruncationIT.java
index 0aac8752434..66b666c36c0 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/repositories/ContentClaimTruncationIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/repositories/ContentClaimTruncationIT.java
@@ -127,7 +127,7 @@ public class ContentClaimTruncationIT extends NiFiSystemIT {
}
for (int i = 0; i < BATCH_COUNT; i++) {
- terminateFlowFile =
getNifiClient().getProcessorClient().runProcessorOnce(terminateFlowFile);
+ terminateFlowFile =
getClientUtil().runProcessorOnce(terminateFlowFile);
getClientUtil().waitForStoppedProcessor(terminateFlowFile.getId());
}
waitForQueueCount(connection.getId(), BATCH_COUNT *
SMALL_FILES_PER_BATCH);
@@ -244,7 +244,7 @@ public class ContentClaimTruncationIT extends NiFiSystemIT {
ProcessorEntity secondTerminate = secondTerminateFlowFile;
for (int i = 0; i < BATCH_COUNT; i++) {
- secondTerminate =
getNifiClient().getProcessorClient().runProcessorOnce(secondTerminate);
+ secondTerminate =
getClientUtil().runProcessorOnce(secondTerminate);
getClientUtil().waitForStoppedProcessor(secondTerminate.getId());
}
waitForQueueCount(secondConnection.getId(), BATCH_COUNT *
SMALL_FILES_PER_BATCH);
@@ -290,7 +290,7 @@ public class ContentClaimTruncationIT extends NiFiSystemIT {
ProcessorEntity currentUpdateContent = updateContent;
for (int i = 0; i < BATCH_COUNT; i++) {
- currentUpdateContent =
getNifiClient().getProcessorClient().runProcessorOnce(currentUpdateContent);
+ currentUpdateContent =
getClientUtil().runProcessorOnce(currentUpdateContent);
getClientUtil().waitForStoppedProcessor(currentUpdateContent.getId());
}
waitForQueueCount(generatorToUpdate.getId(), BATCH_COUNT *
SMALL_FILES_PER_BATCH);
@@ -313,11 +313,12 @@ public class ContentClaimTruncationIT extends
NiFiSystemIT {
final ProcessorEntity generator =
getClientUtil().createProcessor("GenerateTruncatableFlowFiles");
final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile");
+ final int smallFilesPerBatch = 9;
final Map<String, String> generateProps = Map.of(
"Batch Count", "1",
"Small File Size", "1 KB",
"Large File Size", "10 MB",
- "Small Files Per Batch", "9");
+ "Small Files Per Batch", String.valueOf(smallFilesPerBatch));
getClientUtil().updateProcessorProperties(generator, generateProps);
getClientUtil().updateProcessorSchedulingPeriod(generator, "0 sec");
@@ -326,33 +327,39 @@ public class ContentClaimTruncationIT extends
NiFiSystemIT {
connection = getClientUtil().updateConnectionBackpressure(connection,
10000, BACKPRESSURE_BYTES);
getClientUtil().startProcessor(generator);
- waitForQueueCount(connection.getId(), 10);
+ waitForQueueCount(connection.getId(), smallFilesPerBatch + 1);
getClientUtil().stopProcessor(generator);
getClientUtil().waitForStoppedProcessor(generator.getId());
+ // Keep one small FlowFile in the queue. All small FlowFiles share a
single resource claim (they fit
+ // within nifi.content.claim.max.appendable.size), and that is also
the claim referenced by the last
+ // DROP event. The remaining FlowFile pins the claim so the aggressive
archive/truncation cleanup
+ // cannot delete the replayed content before it is read back.
ProcessorEntity currentTerminate = terminate;
- while (getConnectionQueueSize(connection.getId()) > 0) {
- currentTerminate =
getNifiClient().getProcessorClient().runProcessorOnce(currentTerminate);
+ while (getConnectionQueueSize(connection.getId()) > 1) {
+ currentTerminate =
getClientUtil().runProcessorOnce(currentTerminate);
getClientUtil().waitForStoppedProcessor(currentTerminate.getId());
}
- waitForQueueCount(connection.getId(), 0);
+ waitForQueueCount(connection.getId(), 1);
+
+ final String pinnedFlowFileUuid =
getClientUtil().getQueueFlowFile(connection.getId(), 0).getFlowFile().getUuid();
final ReplayLastEventResponseEntity replayResponse =
getNifiClient().getProvenanceClient().replayLastEvent(currentTerminate.getId(),
ReplayEventNodes.PRIMARY);
assertNull(replayResponse.getAggregateSnapshot().getFailureExplanation());
assertNotNull(replayResponse.getAggregateSnapshot().getEventsReplayed());
- waitForQueueCount(connection.getId(), 1);
+ waitForQueueCount(connection.getId(), 2);
- final byte[] replayedContent =
getClientUtil().getFlowFileContentAsByteArray(connection.getId(), 0);
- assertNotNull(replayedContent);
- assertTrue(replayedContent.length > 0,
- "Replayed FlowFile content should not be empty — truncation
must not have destroyed the content");
+ final String firstQueueFlowFileUuid =
getClientUtil().getQueueFlowFile(connection.getId(), 0).getFlowFile().getUuid();
+ final int replayedFlowFileIndex =
pinnedFlowFileUuid.equals(firstQueueFlowFileUuid) ? 1 : 0;
+ final byte[] replayedContent =
getClientUtil().getFlowFileContentAsByteArray(connection.getId(),
replayedFlowFileIndex);
+ assertEquals(SMALL_FILE_SIZE_BYTES, replayedContent.length);
}
private void drainTerminateQueue(final ProcessorEntity
terminateFlowFileProcessor, final String connectionId) throws
NiFiClientException, IOException, InterruptedException {
ProcessorEntity currentProcessor = terminateFlowFileProcessor;
while (getConnectionQueueSize(connectionId) > 0) {
- currentProcessor =
getNifiClient().getProcessorClient().runProcessorOnce(currentProcessor);
+ currentProcessor =
getClientUtil().runProcessorOnce(currentProcessor);
getClientUtil().waitForStoppedProcessor(currentProcessor.getId());
}
}
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/repositories/ContentClaimTruncationWithSwappingIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/repositories/ContentClaimTruncationWithSwappingIT.java
index 797c127aabe..407f12d3ccf 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/repositories/ContentClaimTruncationWithSwappingIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/repositories/ContentClaimTruncationWithSwappingIT.java
@@ -101,7 +101,7 @@ public class ContentClaimTruncationWithSwappingIT extends
NiFiSystemIT {
connection = getClientUtil().updateConnectionBackpressure(connection,
TOTAL_FLOWFILES_PER_CONNECTION + 1000, BACKPRESSURE_BYTES);
ProcessorEntity currentGenerator = generator;
- currentGenerator =
getNifiClient().getProcessorClient().runProcessorOnce(currentGenerator);
+ currentGenerator = getClientUtil().runProcessorOnce(currentGenerator);
getClientUtil().waitForStoppedProcessor(currentGenerator.getId());
waitForQueueCount(connection.getId(), TOTAL_FLOWFILES_PER_CONNECTION);
@@ -114,7 +114,7 @@ public class ContentClaimTruncationWithSwappingIT extends
NiFiSystemIT {
ProcessorEntity currentTerminate = terminateFlowFile;
for (int i = 0; i < BATCH_COUNT; i++) {
- currentTerminate =
getNifiClient().getProcessorClient().runProcessorOnce(currentTerminate);
+ currentTerminate =
getClientUtil().runProcessorOnce(currentTerminate);
getClientUtil().waitForStoppedProcessor(currentTerminate.getId());
}
waitForQueueCount(connection.getId(), TOTAL_FLOWFILES_PER_CONNECTION -
BATCH_COUNT);
@@ -145,7 +145,7 @@ public class ContentClaimTruncationWithSwappingIT extends
NiFiSystemIT {
secondConnection =
getClientUtil().updateConnectionBackpressure(secondConnection,
TOTAL_FLOWFILES_PER_CONNECTION + 1000, BACKPRESSURE_BYTES);
ProcessorEntity currentGenerator = generator;
- currentGenerator =
getNifiClient().getProcessorClient().runProcessorOnce(currentGenerator);
+ currentGenerator = getClientUtil().runProcessorOnce(currentGenerator);
getClientUtil().waitForStoppedProcessor(currentGenerator.getId());
waitForQueueCount(firstConnection.getId(),
TOTAL_FLOWFILES_PER_CONNECTION);
waitForQueueCount(secondConnection.getId(),
TOTAL_FLOWFILES_PER_CONNECTION);
@@ -198,7 +198,7 @@ public class ContentClaimTruncationWithSwappingIT extends
NiFiSystemIT {
secondConnection =
getClientUtil().updateConnectionBackpressure(secondConnection,
TOTAL_FLOWFILES_PER_CONNECTION + 1000, BACKPRESSURE_BYTES);
ProcessorEntity currentGenerator = generator;
- currentGenerator =
getNifiClient().getProcessorClient().runProcessorOnce(currentGenerator);
+ currentGenerator = getClientUtil().runProcessorOnce(currentGenerator);
getClientUtil().waitForStoppedProcessor(currentGenerator.getId());
waitForQueueCount(firstConnection.getId(),
TOTAL_FLOWFILES_PER_CONNECTION);
waitForQueueCount(secondConnection.getId(),
TOTAL_FLOWFILES_PER_CONNECTION);
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/repositories/OffloadContentClaimTruncationIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/repositories/OffloadContentClaimTruncationIT.java
index a4b084ee1fb..d6f9b1e14da 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/repositories/OffloadContentClaimTruncationIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/repositories/OffloadContentClaimTruncationIT.java
@@ -169,7 +169,7 @@ public class OffloadContentClaimTruncationIT extends
NiFiSystemIT {
// DELETE is sufficient for updateContentClaims() to queue the shared
claim for truncation
// even though FLOW_FILE_COUNT - 1 sibling FlowFiles still point
inside it.
terminate =
getNifiClient().getProcessorClient().getProcessor(terminate.getId());
- terminate =
getNifiClient().getProcessorClient().runProcessorOnce(terminate);
+ terminate = getClientUtil().runProcessorOnce(terminate);
getClientUtil().waitForStoppedProcessor(terminate.getId());
waitForQueueCount(dataConnection.getId(), FLOW_FILE_COUNT - 1);
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 e9a468d57fc..112f19603df 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
@@ -127,7 +127,7 @@ public abstract class AbstractStateKeyDropIT extends
NiFiSystemIT {
while (true) {
try {
-
getNifiClient().getProcessorClient().runProcessorOnce(processor);
+ getClientUtil().runProcessorOnce(processor);
break;
} catch (final NiFiClientException e) {
if (!isTransientClusterError(e) || System.currentTimeMillis()
> maxTime) {
