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 15de342ed30 NIFI-16130 Fixed ReplayLastEventEndpointMerger to count
replays per Node (#11450)
15de342ed30 is described below
commit 15de342ed30f542cf15c3fd0a330ca63bf8cb80f
Author: Pierre Villard <[email protected]>
AuthorDate: Wed Jul 22 17:38:50 2026 +0200
NIFI-16130 Fixed ReplayLastEventEndpointMerger to count replays per Node
(#11450)
- Changed accumulator to List of Long instead of Set to provide correct
count of Replays regardless of matching Event IDs since different nodes can
have the same local Event ID
Signed-off-by: David Handermann <[email protected]>
---
.../endpoints/ReplayLastEventEndpointMerger.java | 3 +-
.../ReplayLastEventEndpointMergerTest.java | 122 +++++++++++++++++++++
2 files changed, 124 insertions(+), 1 deletion(-)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ReplayLastEventEndpointMerger.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ReplayLastEventEndpointMerger.java
index 7911713e2bd..0f0e8db44da 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ReplayLastEventEndpointMerger.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ReplayLastEventEndpointMerger.java
@@ -28,6 +28,7 @@ import java.net.URI;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -49,7 +50,7 @@ public class ReplayLastEventEndpointMerger extends
AbstractSingleEntityEndpoint<
final Set<NodeResponse>
problematicResponses) {
// Move all aggregate snapshots into the node snapshots.
- final Set<Long> replayedEventIds = new HashSet<>();
+ final List<Long> replayedEventIds = new ArrayList<>();
final Set<String> failureExplanations = new HashSet<>();
boolean eventAvailable = false;
for (final Map.Entry<NodeIdentifier, ReplayLastEventResponseEntity>
entry : entityMap.entrySet()) {
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/endpoints/ReplayLastEventEndpointMergerTest.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/endpoints/ReplayLastEventEndpointMergerTest.java
new file mode 100644
index 00000000000..e07da827b23
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/endpoints/ReplayLastEventEndpointMergerTest.java
@@ -0,0 +1,122 @@
+/*
+ * 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.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.protocol.NodeIdentifier;
+import org.apache.nifi.web.api.entity.ReplayLastEventResponseEntity;
+import org.apache.nifi.web.api.entity.ReplayLastEventSnapshotDTO;
+import org.junit.jupiter.api.Test;
+
+import java.net.URI;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class ReplayLastEventEndpointMergerTest {
+
+ private static final NodeIdentifier NODE_1 = new NodeIdentifier("node1",
"host1", 8080, "host1", 8081, "host1", 8082, 8083, false);
+ private static final NodeIdentifier NODE_2 = new NodeIdentifier("node2",
"host2", 8080, "host2", 8081, "host2", 8082, 8083, false);
+
+ private static final long EVENT_ID_1 = 7L;
+ private static final long EVENT_ID_2 = 5L;
+
+ @Test
+ void testCanHandle() {
+ final ReplayLastEventEndpointMerger merger = new
ReplayLastEventEndpointMerger();
+
assertTrue(merger.canHandle(URI.create("/nifi-api/provenance-events/latest/replays"),
"POST"));
+
assertFalse(merger.canHandle(URI.create("/nifi-api/provenance-events/latest/replays"),
"GET"));
+
assertFalse(merger.canHandle(URI.create("/nifi-api/provenance-events"),
"POST"));
+ }
+
+ /**
+ * Verifies that when two nodes each replay their last event and both
happen to have the same
+ * local event ID (which is expected when both nodes have processed the
same number of events),
+ * the merged aggregate correctly reports 2 events replayed rather than 1.
+ * Provenance event IDs are local counters per node, so ID collisions
across nodes are normal.
+ */
+ @Test
+ void testMergeResponsesWithIdenticalEventIds() {
+ final ReplayLastEventEndpointMerger merger = new
ReplayLastEventEndpointMerger();
+
+ final ReplayLastEventResponseEntity clientEntity =
createEntity(EVENT_ID_1, null, true);
+
+ // Both nodes replay their last event; both happen to report the same
local event ID
+ final Map<NodeIdentifier, ReplayLastEventResponseEntity> entityMap =
new HashMap<>();
+ entityMap.put(NODE_1, createEntity(EVENT_ID_1, null, true));
+ entityMap.put(NODE_2, createEntity(EVENT_ID_1, null, true));
+
+ merger.mergeResponses(clientEntity, entityMap, Collections.emptySet(),
Collections.emptySet());
+
+ assertEquals(2,
clientEntity.getAggregateSnapshot().getEventsReplayed().size(),
+ "Both nodes replayed an event; aggregate must count them
independently regardless of matching local event IDs");
+ assertTrue(clientEntity.getAggregateSnapshot().getEventAvailable());
+
assertNull(clientEntity.getAggregateSnapshot().getFailureExplanation());
+ assertEquals(2, clientEntity.getNodeSnapshots().size());
+ }
+
+ @Test
+ void testMergeResponsesWithDistinctEventIds() {
+ final ReplayLastEventEndpointMerger merger = new
ReplayLastEventEndpointMerger();
+
+ final ReplayLastEventResponseEntity clientEntity =
createEntity(EVENT_ID_1, null, true);
+
+ final Map<NodeIdentifier, ReplayLastEventResponseEntity> entityMap =
new HashMap<>();
+ entityMap.put(NODE_1, createEntity(EVENT_ID_1, null, true));
+ entityMap.put(NODE_2, createEntity(EVENT_ID_2, null, true));
+
+ merger.mergeResponses(clientEntity, entityMap, Collections.emptySet(),
Collections.emptySet());
+
+ assertEquals(2,
clientEntity.getAggregateSnapshot().getEventsReplayed().size());
+ assertTrue(clientEntity.getAggregateSnapshot().getEventAvailable());
+
assertNull(clientEntity.getAggregateSnapshot().getFailureExplanation());
+ }
+
+ @Test
+ void testMergeResponsesWithFailure() {
+ final ReplayLastEventEndpointMerger merger = new
ReplayLastEventEndpointMerger();
+
+ final ReplayLastEventResponseEntity clientEntity =
createEntity(EVENT_ID_1, null, true);
+
+ final Map<NodeIdentifier, ReplayLastEventResponseEntity> entityMap =
new HashMap<>();
+ entityMap.put(NODE_1, createEntity(EVENT_ID_1, null, true));
+ entityMap.put(NODE_2, createEntity(null, "Source FlowFile Queue",
false));
+
+ merger.mergeResponses(clientEntity, entityMap, Collections.emptySet(),
Collections.emptySet());
+
+ assertEquals(1,
clientEntity.getAggregateSnapshot().getEventsReplayed().size());
+ assertTrue(clientEntity.getAggregateSnapshot().getEventAvailable());
+
assertTrue(clientEntity.getAggregateSnapshot().getFailureExplanation().contains("Source
FlowFile Queue"));
+ }
+
+ private ReplayLastEventResponseEntity createEntity(final Long eventId,
final String failureExplanation, final boolean eventAvailable) {
+ final ReplayLastEventSnapshotDTO snapshot = new
ReplayLastEventSnapshotDTO();
+ snapshot.setEventAvailable(eventAvailable);
+ snapshot.setFailureExplanation(failureExplanation);
+ if (eventId != null) {
+ snapshot.setEventsReplayed(Collections.singletonList(eventId));
+ }
+
+ final ReplayLastEventResponseEntity entity = new
ReplayLastEventResponseEntity();
+ entity.setAggregateSnapshot(snapshot);
+ return entity;
+ }
+}
