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 1cabec6b5e5 NIFI-15913 Fixed Repository Record creation for S2S and
Load-Balanced Connections (#11215)
1cabec6b5e5 is described below
commit 1cabec6b5e55ef4ba782ecb66a3d51e55dcdb753
Author: Mark Payne <[email protected]>
AuthorDate: Wed May 6 14:41:58 2026 -0400
NIFI-15913 Fixed Repository Record creation for S2S and Load-Balanced
Connections (#11215)
- Add system test reproducing premature content claim truncation after
node offload
- Fixed bug in which we create RepositoryRecords with the incorrect 'type'
when receiving data via site-to-site or load-balanced connections; also
addressed the same issue in cases where we have a Stateless Process Group with
an OutputPort that clones to multiple destinations.
Signed-off-by: David Handermann <[email protected]>
---
.../apache/nifi/connectable/ConnectionUtils.java | 3 +-
.../nifi/connectable/TestConnectionUtils.java | 121 ++++++++++++
.../server/StandardLoadBalanceProtocol.java | 5 +-
.../server/TestStandardLoadBalanceProtocol.java | 6 +
.../TestWriteAheadFlowFileRepository.java | 59 ++++++
.../OffloadContentClaimTruncationIT.java | 212 +++++++++++++++++++++
.../StatelessOutputContentClaimTruncationIT.java | 191 +++++++++++++++++++
7 files changed, 595 insertions(+), 2 deletions(-)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/connectable/ConnectionUtils.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/connectable/ConnectionUtils.java
index d6eccf61a34..87792473fdc 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/connectable/ConnectionUtils.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/connectable/ConnectionUtils.java
@@ -101,7 +101,8 @@ public class ConnectionUtils {
}
private static RepositoryRecord createRepositoryRecord(final
FlowFileRecord flowFile, final FlowFileQueue destinationQueue) {
- final StandardRepositoryRecord repoRecord = new
StandardRepositoryRecord(null, flowFile);
+ // The FlowFile is being introduced to this repository for the first
time, so it is tracked as a CREATE record.
+ final StandardRepositoryRecord repoRecord = new
StandardRepositoryRecord(destinationQueue);
repoRecord.setWorking(flowFile, Collections.emptyMap(), false);
repoRecord.setDestination(destinationQueue);
return repoRecord;
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/connectable/TestConnectionUtils.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/connectable/TestConnectionUtils.java
new file mode 100644
index 00000000000..9dcf56843ad
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/connectable/TestConnectionUtils.java
@@ -0,0 +1,121 @@
+/*
+ * 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.connectable;
+
+import org.apache.nifi.connectable.ConnectionUtils.FlowFileCloneResult;
+import org.apache.nifi.controller.queue.FlowFileQueue;
+import org.apache.nifi.controller.repository.ContentRepository;
+import org.apache.nifi.controller.repository.FlowFileRecord;
+import org.apache.nifi.controller.repository.FlowFileRepository;
+import org.apache.nifi.controller.repository.RepositoryRecord;
+import org.apache.nifi.controller.repository.RepositoryRecordType;
+import org.apache.nifi.controller.repository.StandardFlowFileRecord;
+import org.apache.nifi.controller.repository.claim.ContentClaim;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.when;
+
+public class TestConnectionUtils {
+
+ private FlowFileRepository flowFileRepository;
+ private ContentRepository contentRepository;
+
+ @BeforeEach
+ public void setup() {
+ flowFileRepository = Mockito.mock(FlowFileRepository.class);
+ contentRepository = Mockito.mock(ContentRepository.class);
+
+ final AtomicLong nextId = new AtomicLong(1000L);
+
when(flowFileRepository.getNextFlowFileSequence()).thenAnswer(invocation ->
nextId.getAndIncrement());
+ }
+
+ @Test
+ public void testCloneSingleDestinationProducesCreateRecord() {
+ final FlowFileRecord flowFile = new StandardFlowFileRecord.Builder()
+ .id(1L)
+ .addAttribute("uuid", "test-uuid-1")
+ .build();
+
+ final FlowFileQueue destinationQueue =
Mockito.mock(FlowFileQueue.class);
+ final Connection destination = mockConnection(destinationQueue);
+
+ final FlowFileCloneResult result = ConnectionUtils.clone(flowFile,
List.of(destination), flowFileRepository, contentRepository);
+
+ final List<RepositoryRecord> records = result.getRepositoryRecords();
+ assertEquals(1, records.size());
+ final RepositoryRecord record = records.get(0);
+ assertEquals(RepositoryRecordType.CREATE, record.getType());
+ assertEquals(destinationQueue, record.getDestination());
+ assertNotNull(record.getCurrent());
+ assertEquals(flowFile.getId(), record.getCurrent().getId());
+ }
+
+ @Test
+ public void testCloneMultipleDestinationsAllProduceCreateRecords() {
+ final ContentClaim contentClaim = Mockito.mock(ContentClaim.class);
+ final FlowFileRecord flowFile = new StandardFlowFileRecord.Builder()
+ .id(1L)
+ .addAttribute("uuid", "test-uuid-1")
+ .contentClaim(contentClaim)
+ .size(1024L)
+ .build();
+
+ final FlowFileQueue queueOne = Mockito.mock(FlowFileQueue.class);
+ final FlowFileQueue queueTwo = Mockito.mock(FlowFileQueue.class);
+ final FlowFileQueue queueThree = Mockito.mock(FlowFileQueue.class);
+ final List<Connection> destinations = List.of(
+ mockConnection(queueOne),
+ mockConnection(queueTwo),
+ mockConnection(queueThree));
+
+ final FlowFileCloneResult result = ConnectionUtils.clone(flowFile,
destinations, flowFileRepository, contentRepository);
+
+ final List<RepositoryRecord> records = result.getRepositoryRecords();
+ assertEquals(destinations.size(), records.size());
+
+ // Each record produced by clone() represents a FlowFile being
introduced to the repository,
+ // whether it is the original routed FlowFile or a sibling clone.
+ assertTrue(records.stream().allMatch(record -> record.getType() ==
RepositoryRecordType.CREATE));
+
+ // The clones (one per additional destination beyond the first) should
each have triggered a
+ // claimant count increment on the shared Content Claim.
+ Mockito.verify(contentRepository, Mockito.times(destinations.size() -
1)).incrementClaimaintCount(contentClaim);
+
+ final List<FlowFileQueue> destinationQueues = new ArrayList<>();
+ for (final RepositoryRecord record : records) {
+ destinationQueues.add(record.getDestination());
+ }
+ assertTrue(destinationQueues.contains(queueOne));
+ assertTrue(destinationQueues.contains(queueTwo));
+ assertTrue(destinationQueues.contains(queueThree));
+ }
+
+ private Connection mockConnection(final FlowFileQueue queue) {
+ final Connection connection = Mockito.mock(Connection.class);
+ when(connection.getFlowFileQueue()).thenReturn(queue);
+ return connection;
+ }
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/server/StandardLoadBalanceProtocol.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/server/StandardLoadBalanceProtocol.java
index 4d82d5ee86c..b187badce3e 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/server/StandardLoadBalanceProtocol.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/server/StandardLoadBalanceProtocol.java
@@ -56,6 +56,7 @@ import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -421,7 +422,9 @@ public class StandardLoadBalanceProtocol implements
LoadBalanceProtocol {
private void updateFlowFileRepository(final List<RemoteFlowFileRecord>
flowFiles, final FlowFileQueue flowFileQueue) throws IOException {
final List<RepositoryRecord> repoRecords = flowFiles.stream()
.map(remoteFlowFile -> {
- final StandardRepositoryRecord record = new
StandardRepositoryRecord(flowFileQueue, remoteFlowFile.getFlowFile());
+ // A received FlowFile is new to this node's repository,
so track it as a CREATE record.
+ final StandardRepositoryRecord record = new
StandardRepositoryRecord(flowFileQueue);
+ record.setWorking(remoteFlowFile.getFlowFile(),
Collections.emptyMap(), false);
record.setDestination(flowFileQueue);
return record;
})
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/queue/clustered/server/TestStandardLoadBalanceProtocol.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/queue/clustered/server/TestStandardLoadBalanceProtocol.java
index 1df846d1ac6..04a57646648 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/queue/clustered/server/TestStandardLoadBalanceProtocol.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/queue/clustered/server/TestStandardLoadBalanceProtocol.java
@@ -27,6 +27,7 @@ import
org.apache.nifi.controller.repository.ContentRepository;
import org.apache.nifi.controller.repository.FlowFileRecord;
import org.apache.nifi.controller.repository.FlowFileRepository;
import org.apache.nifi.controller.repository.RepositoryRecord;
+import org.apache.nifi.controller.repository.RepositoryRecordType;
import org.apache.nifi.controller.repository.claim.ContentClaim;
import org.apache.nifi.controller.repository.claim.ResourceClaim;
import org.apache.nifi.provenance.ProvenanceEventRecord;
@@ -206,6 +207,10 @@ public class TestStandardLoadBalanceProtocol {
Mockito.verify(provenanceRepo, times(1)).registerEvents(anyList());
Mockito.verify(flowFileQueue, times(0)).putAll(anyCollection());
Mockito.verify(flowFileQueue,
times(1)).receiveFromPeer(anyCollection());
+
+ // Repository records for received FlowFiles must be CREATE so that
the FlowFile Repository
+ // increments the Content Claim's truncation reference count for each
received FlowFile.
+ assertTrue(flowFileRepoUpdateRecords.stream().allMatch(record ->
record.getType() == RepositoryRecordType.CREATE));
}
@Test
@@ -270,6 +275,7 @@ public class TestStandardLoadBalanceProtocol {
assertEquals(4, flowFileQueueReceiveRecords.size());
assertTrue(provRepoUpdateRecords.stream().allMatch(event ->
event.getEventType() == ProvenanceEventType.RECEIVE));
+ assertTrue(flowFileRepoUpdateRecords.stream().allMatch(record ->
record.getType() == RepositoryRecordType.CREATE));
}
@Test
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestWriteAheadFlowFileRepository.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestWriteAheadFlowFileRepository.java
index 7898820435d..af5d92b3956 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestWriteAheadFlowFileRepository.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestWriteAheadFlowFileRepository.java
@@ -1155,6 +1155,65 @@ public class TestWriteAheadFlowFileRepository {
assertTrue(truncated.contains(originalClaim));
}
+ @Test
+ public void
testCreateRecordsForSharedTruncatableClaimPreventPrematureTruncation() throws
IOException {
+ // Multiple CREATE records for the same truncation-eligible Content
Claim should increment
+ // the truncation reference count per FlowFile, so deleting one
sibling does not queue the
+ // shared claim for truncation while others still reference it.
+ final RuntimeRepoContext context = createRuntimeRepoContext();
+
+ final ResourceClaim resourceClaim =
context.claimManager().newResourceClaim("container", "section", "1", false,
false);
+ context.claimManager().incrementClaimantCount(resourceClaim);
+ context.claimManager().incrementClaimantCount(resourceClaim);
+ context.claimManager().incrementClaimantCount(resourceClaim);
+ final StandardContentClaim sharedClaim = createClaim(resourceClaim,
1024L, TRUNCATION_CANDIDATE_LENGTH, true);
+
+ final FlowFileRecord first = new StandardFlowFileRecord.Builder()
+ .id(1L)
+ .addAttribute("uuid", UUID.randomUUID().toString())
+ .contentClaim(sharedClaim)
+ .build();
+ final FlowFileRecord second = new StandardFlowFileRecord.Builder()
+ .id(2L)
+ .addAttribute("uuid", UUID.randomUUID().toString())
+ .contentClaim(sharedClaim)
+ .build();
+ final FlowFileRecord third = new StandardFlowFileRecord.Builder()
+ .id(3L)
+ .addAttribute("uuid", UUID.randomUUID().toString())
+ .contentClaim(sharedClaim)
+ .build();
+
+ try (final WriteAheadFlowFileRepository repo = new
WriteAheadFlowFileRepository(niFiProperties)) {
+ repo.initialize(context.claimManager());
+ repo.loadFlowFiles(context.queueProvider());
+
+ final List<RepositoryRecord> createRecords = new ArrayList<>();
+ for (final FlowFileRecord flowFile : List.of(first, second,
third)) {
+ final StandardRepositoryRecord createRecord = new
StandardRepositoryRecord(context.queue());
+ createRecord.setWorking(flowFile, false);
+ createRecord.setDestination(context.queue());
+ createRecords.add(createRecord);
+ }
+ repo.updateRepository(createRecords);
+
+ assertEquals(3, repo.getContentClaimReferenceCount(sharedClaim),
+ "Each CREATE record for a truncation-eligible Content
Claim must increment the truncation reference count");
+
+ final StandardRepositoryRecord deleteRecord = new
StandardRepositoryRecord(context.queue(), first);
+ deleteRecord.markForDelete();
+ repo.updateRepository(List.of(deleteRecord));
+ repo.checkpoint();
+ }
+
+ assertEquals(2,
context.claimManager().getTruncationReferenceCount(sharedClaim));
+
+ final List<ContentClaim> truncated = new ArrayList<>();
+ context.claimManager().drainTruncatableClaims(truncated, 100);
+ assertFalse(truncated.contains(sharedClaim),
+ "Shared Content Claim must not be queued for truncation while
live siblings still reference it");
+ }
+
//
=========================================================================
// Truncation Feature: Recovery Tests
//
=========================================================================
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
new file mode 100644
index 00000000000..a4b084ee1fb
--- /dev/null
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/repositories/OffloadContentClaimTruncationIT.java
@@ -0,0 +1,212 @@
+/*
+ * 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.repositories;
+
+import org.apache.nifi.cluster.coordination.node.ClusterRoles;
+import org.apache.nifi.scheduling.ExecutionNode;
+import org.apache.nifi.tests.system.InstanceConfiguration;
+import org.apache.nifi.tests.system.NiFiInstanceFactory;
+import org.apache.nifi.tests.system.NiFiSystemIT;
+import org.apache.nifi.tests.system.SpawnedClusterNiFiInstanceFactory;
+import org.apache.nifi.web.api.dto.NodeDTO;
+import org.apache.nifi.web.api.entity.ConnectionEntity;
+import org.apache.nifi.web.api.entity.ProcessorEntity;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+/**
+ * System test that exercises the interaction between Node Offload and Content
Claim Truncation.
+ *
+ * <p>When a node is offloaded, every FlowFile in its local partitions is
shipped to peer nodes via
+ * the load-balance protocol. On the peer, {@code StandardLoadBalanceProtocol}
packs every received
+ * FlowFile in a transaction into a single {@code ContentClaim} produced by
+ * {@code ContentRepository.create()}. If the peer's {@code
writableClaimQueue} contains a
+ * partially-filled Resource Claim at the time of receive, that {@code
ContentClaim} ends up with
+ * {@code offset > 0} and, given a sufficiently large transaction, with {@code
length} above the
+ * truncation threshold — making the claim a truncation candidate.</p>
+ *
+ * <p>The receive path persists each received FlowFile as an {@code UPDATE}
repository record without
+ * marking the record as content-modified, so the truncation reference count
for the shared Content
+ * Claim is never incremented to reflect the live FlowFiles that point inside
it. Once a single
+ * sibling FlowFile is removed downstream, the shared Content Claim's
reference count is observed as
+ * zero by {@code isTruncationAllowed} and the claim is queued for truncation
while the surviving
+ * FlowFiles still reference offsets inside it.</p>
+ *
+ * <p>This test reproduces that scenario end-to-end. It generates content on
the primary node, offloads
+ * the primary node so the data is shipped to the peer, removes a single
FlowFile to arm truncation,
+ * waits for the periodic truncate task to run, and then asserts that every
remaining FlowFile in the
+ * connection on the peer still returns its original content.</p>
+ */
+@Timeout(value = 5, unit = TimeUnit.MINUTES)
+public class OffloadContentClaimTruncationIT extends NiFiSystemIT {
+
+ private static final int CONTENT_BYTES_PER_FLOW_FILE = 8 * 1024;
+ private static final int FLOW_FILE_COUNT = 30;
+ private static final String CONTENT_TEXT =
"x".repeat(CONTENT_BYTES_PER_FLOW_FILE);
+
+ private static final int CLUSTER_NODE_COUNT = 2;
+
+ @Override
+ public NiFiInstanceFactory getInstanceFactory() {
+ final Map<String, String> propertyOverrides = Map.of(
+ "nifi.flowfile.repository.checkpoint.interval", "1 sec",
+ "nifi.content.repository.archive.cleanup.frequency", "1 sec",
+ "nifi.content.claim.max.appendable.size", "50 KB",
+ "nifi.content.claim.truncation.enabled", "true",
+ "nifi.content.repository.archive.max.usage.percentage", "1%");
+
+ return new SpawnedClusterNiFiInstanceFactory(
+ new InstanceConfiguration.Builder()
+
.bootstrapConfig("src/test/resources/conf/clustered/node1/bootstrap.conf")
+ .instanceDirectory("target/node1")
+ .overrideNifiProperties(propertyOverrides)
+ .build(),
+ new InstanceConfiguration.Builder()
+
.bootstrapConfig("src/test/resources/conf/clustered/node2/bootstrap.conf")
+ .instanceDirectory("target/node2")
+ .overrideNifiProperties(propertyOverrides)
+ .build());
+ }
+
+ @Override
+ protected boolean isAllowFactoryReuse() {
+ return false;
+ }
+
+ @Override
+ protected boolean isDestroyEnvironmentAfterEachTest() {
+ return true;
+ }
+
+ @Test
+ public void testOffloadedFlowFileContentNotPrematurelyTruncated() throws
Exception {
+ // Pre-warm both nodes' writableClaimQueue with a partially-filled
Resource Claim. Each node
+ // generates a single small FlowFile, leaving it queued upstream of an
unscheduled
+ // TerminateFlowFile. The Resource Claim file remains writable and a
few bytes long, so the
+ // next ContentRepository.create() call on each node will reuse this
Resource Claim and the
+ // resulting ContentClaim will have offset > 0.
+ final ProcessorEntity prewarmGenerator =
getClientUtil().createProcessor("GenerateFlowFile");
+ getClientUtil().updateProcessorProperties(prewarmGenerator, Map.of(
+ "Text", "warmup",
+ "Batch Size", "1",
+ "Max FlowFiles", "1"));
+ getClientUtil().updateProcessorSchedulingPeriod(prewarmGenerator, "0
sec");
+
+ final ProcessorEntity prewarmTerminate =
getClientUtil().createProcessor("TerminateFlowFile");
+ final ConnectionEntity prewarmConnection =
getClientUtil().createConnection(prewarmGenerator, prewarmTerminate, "success");
+
+ getClientUtil().startProcessor(prewarmGenerator);
+ waitForQueueCount(prewarmConnection.getId(), CLUSTER_NODE_COUNT);
+ getClientUtil().stopProcessor(prewarmGenerator);
+
+ // Source generator runs only on the primary node and produces
FLOW_FILE_COUNT FlowFiles,
+ // each carrying the same CONTENT_TEXT. The non-primary node generates
nothing in this phase, so
+ // its writableClaimQueue retains the partially-filled Resource Claim
from pre-warm.
+ final ProcessorEntity sourceGenerator =
getClientUtil().createProcessor("GenerateFlowFile");
+ getClientUtil().updateProcessorProperties(sourceGenerator, Map.of(
+ "Text", CONTENT_TEXT,
+ "Batch Size", String.valueOf(FLOW_FILE_COUNT),
+ "Max FlowFiles", String.valueOf(FLOW_FILE_COUNT)));
+ getClientUtil().updateProcessorExecutionNode(sourceGenerator,
ExecutionNode.PRIMARY);
+ getClientUtil().updateProcessorSchedulingPeriod(sourceGenerator, "0
sec");
+
+ ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile");
+ final ConnectionEntity dataConnection =
getClientUtil().createConnection(sourceGenerator, terminate, "success");
+
+ getClientUtil().startProcessor(sourceGenerator);
+ waitForQueueCount(dataConnection.getId(), FLOW_FILE_COUNT);
+ getClientUtil().stopProcessor(sourceGenerator);
+
+ // Identify the primary and non-primary nodes. The primary node is the
one holding the
+ // FlowFiles produced above and is the node that will be offloaded.
The non-primary node is
+ // the receiver and must remain reachable for queue listings and
content fetches after offload.
+ final NodeDTO primaryNode = findNodeByRole(ClusterRoles.PRIMARY_NODE);
+ assertNotNull(primaryNode, "Cluster does not have a Primary Node");
+ final NodeDTO nonPrimaryNode = findOtherNode(primaryNode);
+ assertNotNull(nonPrimaryNode, "Cluster does not have a non-primary
node");
+
+ final int nonPrimaryNodeIndex = nonPrimaryNode.getApiPort() -
CLUSTERED_CLIENT_API_BASE_PORT + 1;
+ switchClientToNode(nonPrimaryNodeIndex);
+
+ // Disconnect and offload the primary node. All FlowFiles in the
primary's local partition are
+ // shipped to the non-primary via the load-balance receive path. On
the receiver, every FlowFile
+ // in the receive transaction is packed into a single ContentClaim
that sits at offset > 0
+ // inside the recycled Resource Claim. With FLOW_FILE_COUNT *
CONTENT_BYTES_PER_FLOW_FILE bytes
+ // exceeding the 50 KB truncation threshold, the shared ContentClaim
is flagged as a truncation
+ // candidate.
+ getClientUtil().disconnectNode(primaryNode.getNodeId());
+ waitForNodeStatus(primaryNode, "DISCONNECTED");
+ getClientUtil().offloadNode(primaryNode.getNodeId());
+ waitForNodeStatus(primaryNode, "OFFLOADED");
+
+ waitForQueueCount(dataConnection.getId(), FLOW_FILE_COUNT);
+
+ // Trigger a single DELETE record that references the shared,
truncatable ContentClaim by
+ // running TerminateFlowFile once on the receiver. Without correct
truncation reference count
+ // tracking on the receive path the ContentClaim's reference count is
zero, so this single
+ // 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);
+ getClientUtil().waitForStoppedProcessor(terminate.getId());
+ waitForQueueCount(dataConnection.getId(), FLOW_FILE_COUNT - 1);
+
+ // Allow time for the periodic TruncateClaims task on the receiver to
drain the truncatable
+ // claim queue and call FileChannel.truncate(claim.getOffset()), which
would otherwise destroy
+ // the on-disk bytes for every remaining live FlowFile that references
this claim.
+ Thread.sleep(5_000L);
+
+ // Read every remaining FlowFile and verify its content is intact.
With the bug present, the
+ // underlying Resource Claim file has been truncated back to
claim.getOffset() and these reads
+ // return zero bytes (or fewer than expected) for each offload
FlowFile.
+ final byte[] expectedContent =
CONTENT_TEXT.getBytes(StandardCharsets.UTF_8);
+ final int remaining = FLOW_FILE_COUNT - 1;
+ for (int index = 0; index < remaining; index++) {
+ final byte[] actualContent =
getClientUtil().getFlowFileContentAsByteArray(dataConnection.getId(), index);
+ assertArrayEquals(expectedContent, actualContent,
+ "FlowFile at index " + index + " has unexpected content;
expected "
+ + expectedContent.length + " bytes, got " +
actualContent.length);
+ }
+ }
+
+ private NodeDTO findNodeByRole(final String role) throws Exception {
+ for (final NodeDTO nodeDto :
getNifiClient().getControllerClient().getNodes().getCluster().getNodes()) {
+ final Set<String> roles = nodeDto.getRoles();
+ if (roles != null && roles.contains(role)) {
+ return nodeDto;
+ }
+ }
+ return null;
+ }
+
+ private NodeDTO findOtherNode(final NodeDTO excludedNode) throws Exception
{
+ for (final NodeDTO nodeDto :
getNifiClient().getControllerClient().getNodes().getCluster().getNodes()) {
+ if (!nodeDto.getNodeId().equals(excludedNode.getNodeId())) {
+ return nodeDto;
+ }
+ }
+ return null;
+ }
+}
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/repositories/StatelessOutputContentClaimTruncationIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/repositories/StatelessOutputContentClaimTruncationIT.java
new file mode 100644
index 00000000000..79d24be824e
--- /dev/null
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/repositories/StatelessOutputContentClaimTruncationIT.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.repositories;
+
+import org.apache.nifi.tests.system.NiFiSystemIT;
+import org.apache.nifi.web.api.entity.ConnectionEntity;
+import org.apache.nifi.web.api.entity.PortEntity;
+import org.apache.nifi.web.api.entity.ProcessGroupEntity;
+import org.apache.nifi.web.api.entity.ProcessorEntity;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+
+/**
+ * System test that exercises the interaction between Content Claim Truncation
and the Stateless
+ * to framework output bridge.
+ *
+ * <p>When a Stateless Process Group hands its output FlowFiles back to the
framework,
+ * {@code StatelessFlowTask.createOutputRecords} delegates to
+ * {@code ConnectionUtils.createRepositoryRecord}, which constructs each output
+ * {@code StandardRepositoryRecord} as an {@code UPDATE} record without
marking the record as
+ * content-modified. As a result, when {@code
WriteAheadFlowFileRepository.updateRepository}
+ * processes the batch, the truncation reference count for the output
FlowFiles' Content Claim is
+ * never incremented. The framework's view of that Content Claim therefore
retains only the count
+ * that was established for the input FlowFile when it was originally created
upstream of the
+ * Stateless group.</p>
+ *
+ * <p>If the input FlowFile inside the Stateless group is split via {@code
session.clone(offset, size)}
+ * into many children that all share the same Content Claim, and then the
input FlowFile is removed
+ * inside the Stateless group, the framework records the input FlowFile as a
{@code DELETE} alongside
+ * the children's {@code UPDATE}-without-content-modified records. The {@code
DELETE} decrements the
+ * truncation reference count for the shared Content Claim to zero while the
children still reference
+ * offsets inside it. The Content Claim is queued for truncation and its
on-disk Resource Claim file
+ * is shrunk back to {@code claim.getOffset()} by the periodic truncate task,
destroying every byte
+ * the surviving children depend on.</p>
+ *
+ * <p>This test reproduces that scenario end-to-end. A pre-warm phase ensures
the Resource Claim
+ * reused for the input FlowFile begins at {@code offset > 0}. A multi-line
input FlowFile larger than
+ * the truncation threshold flows into a Stateless group that splits it via
{@code session.clone}. The
+ * children land in a downstream queue and the test then asserts that every
child still returns its
+ * original line bytes after the truncate task has had time to run.</p>
+ */
+@Timeout(value = 5, unit = TimeUnit.MINUTES)
+public class StatelessOutputContentClaimTruncationIT extends NiFiSystemIT {
+
+ private static final int LINE_BYTES = 2 * 1024;
+ private static final int LINE_COUNT = 30;
+ private static final String LINE_TEXT = "x".repeat(LINE_BYTES);
+
+ @Override
+ protected Map<String, String> getNifiPropertiesOverrides() {
+ return Map.of(
+ "nifi.flowfile.repository.checkpoint.interval", "1 sec",
+ "nifi.content.repository.archive.cleanup.frequency", "1 sec",
+ "nifi.content.claim.max.appendable.size", "50 KB",
+ "nifi.content.claim.truncation.enabled", "true",
+ "nifi.content.repository.archive.max.usage.percentage", "1%");
+ }
+
+ @Override
+ protected boolean isAllowFactoryReuse() {
+ return false;
+ }
+
+ @Override
+ protected boolean isDestroyEnvironmentAfterEachTest() {
+ return true;
+ }
+
+ @Test
+ public void testStatelessOutputContentNotPrematurelyTruncated() throws
Exception {
+ // Pre-warm the writableClaimQueue with a partially-filled Resource
Claim. The pre-warm
+ // FlowFile is queued upstream of an unstarted TerminateFlowFile so
the Resource Claim file
+ // remains writable and only a few bytes long. The next
ContentRepository.create() call will
+ // reuse this Resource Claim and the resulting ContentClaim will have
offset > 0.
+ final ProcessorEntity prewarmGenerator =
getClientUtil().createProcessor("GenerateFlowFile");
+ getClientUtil().updateProcessorProperties(prewarmGenerator, Map.of(
+ "Text", "warmup",
+ "Batch Size", "1",
+ "Max FlowFiles", "1"));
+ getClientUtil().updateProcessorSchedulingPeriod(prewarmGenerator, "0
sec");
+
+ final ProcessorEntity prewarmTerminate =
getClientUtil().createProcessor("TerminateFlowFile");
+ final ConnectionEntity prewarmConnection =
getClientUtil().createConnection(prewarmGenerator, prewarmTerminate, "success");
+
+ getClientUtil().startProcessor(prewarmGenerator);
+ waitForQueueCount(prewarmConnection.getId(), 1);
+ getClientUtil().stopProcessor(prewarmGenerator);
+
+ // Build the multi-line content. Each line is LINE_BYTES of 'x'
followed by a unique numeric
+ // suffix and a newline. The unique suffix lets the assertion identify
which line each child
+ // FlowFile carries. The total content size of LINE_COUNT *
(LINE_BYTES + suffix + '\n') is
+ // well above the 50 KB truncation threshold, so the input FlowFile's
ContentClaim becomes a
+ // truncation candidate once the writableClaimQueue is pre-warmed.
+ final StringBuilder contentBuilder = new StringBuilder();
+ final String[] expectedLines = new String[LINE_COUNT];
+ for (int lineIndex = 0; lineIndex < LINE_COUNT; lineIndex++) {
+ final String line = LINE_TEXT + "-" + lineIndex;
+ expectedLines[lineIndex] = line;
+ contentBuilder.append(line).append('\n');
+ }
+ final String inputContent = contentBuilder.toString();
+
+ // Source generator produces a single multi-line FlowFile. Connecting
it to the Stateless
+ // group's input port enrolls the FlowFile in the framework's FlowFile
repository with a
+ // CREATE record that increments the truncation reference count for
the new ContentClaim.
+ final ProcessorEntity sourceGenerator =
getClientUtil().createProcessor("GenerateFlowFile");
+ getClientUtil().updateProcessorProperties(sourceGenerator, Map.of(
+ "Text", inputContent,
+ "Batch Size", "1",
+ "Max FlowFiles", "1"));
+ getClientUtil().updateProcessorSchedulingPeriod(sourceGenerator, "0
sec");
+
+ final ProcessGroupEntity statelessGroup =
getClientUtil().createProcessGroup("Stateless", "root");
+ getClientUtil().markStateless(statelessGroup, "1 min");
+
+ final PortEntity inputPort = getClientUtil().createInputPort("In",
statelessGroup.getId());
+ final PortEntity outputPort = getClientUtil().createOutputPort("Out",
statelessGroup.getId());
+
+ // Inside the Stateless group, SplitByLine with Use Clone = true uses
session.clone(offset,
+ // size) so every child FlowFile shares the input FlowFile's
ContentClaim, with each child's
+ // FlowFile-level offset pointing into a different region of that
shared claim.
+ final ProcessorEntity splitByLine =
getClientUtil().createProcessor("SplitByLine", statelessGroup.getId());
+ getClientUtil().updateProcessorProperties(splitByLine, Map.of("Use
Clone", "true"));
+
+ final ConnectionEntity sourceToInput =
getClientUtil().createConnection(sourceGenerator, inputPort, "success");
+ getClientUtil().createConnection(inputPort, splitByLine,
statelessGroup.getId());
+ getClientUtil().createConnection(splitByLine, outputPort, "success",
statelessGroup.getId());
+
+ // The output queue is connected to an unstarted TerminateFlowFile so
children stay queued
+ // and remain readable for the assertion phase. The act of bridging
the children back to the
+ // framework from the Stateless group is itself sufficient to expose
the bug; no manual
+ // delete on the output side is required.
+ final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile");
+ final ConnectionEntity outputToTerminate =
getClientUtil().createConnection(outputPort, terminate);
+
+ getClientUtil().waitForValidProcessor(sourceGenerator.getId());
+ getClientUtil().waitForValidProcessor(splitByLine.getId());
+
+ getClientUtil().startProcessor(sourceGenerator);
+ waitForQueueCount(sourceToInput.getId(), 1);
+ getClientUtil().stopProcessor(sourceGenerator);
+
+ getClientUtil().startProcessGroupComponents(statelessGroup.getId());
+
+ // Wait for the Stateless invocation to land all of its children in
the framework queue.
+ // Internally this is the moment StatelessFlowTask calls
flowFileRepository.updateRepository
+ // with one DELETE record for the input FlowFile and LINE_COUNT
UPDATE-without-modified
+ // records for the children. With the bug present, the shared
ContentClaim's truncation
+ // reference count drops to zero on this single repository call and
the claim is enqueued
+ // for truncation by the periodic TruncateClaims task.
+ waitForQueueCount(outputToTerminate.getId(), LINE_COUNT);
+ getClientUtil().stopProcessGroupComponents(statelessGroup.getId());
+
+ // Allow the periodic TruncateClaims task to run. With the bug
present, this is when the
+ // shared Resource Claim file is truncated back to the input
FlowFile's ContentClaim offset
+ // and the children's bytes are destroyed.
+ Thread.sleep(5_000L);
+
+ // Read every child and verify its content matches the corresponding
line. With the bug
+ // present, the underlying Resource Claim file has been shrunk to
ContentClaim.getOffset()
+ // and these reads return zero bytes (or fewer than expected) for
every child FlowFile that
+ // lived inside the truncated region of the file.
+ for (int childIndex = 0; childIndex < LINE_COUNT; childIndex++) {
+ final byte[] expectedContent =
expectedLines[childIndex].getBytes(StandardCharsets.UTF_8);
+ final byte[] actualContent =
getClientUtil().getFlowFileContentAsByteArray(outputToTerminate.getId(),
childIndex);
+ assertArrayEquals(expectedContent, actualContent,
+ "Child FlowFile at queue index " + childIndex + " has
unexpected content; expected "
+ + expectedContent.length + " bytes, got " +
actualContent.length);
+ }
+ }
+}
