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 bf7085c3daf NIFI-16248 Added Response Merger for Connector Purge
Requests (#11587)
bf7085c3daf is described below
commit bf7085c3dafd90b216fbcaeaa2eb833cb5a9fce3
Author: skeossei <[email protected]>
AuthorDate: Tue Aug 25 08:00:22 2026 -0700
NIFI-16248 Added Response Merger for Connector Purge Requests (#11587)
Signed-off-by: David Handermann <[email protected]>
---
.../http/StandardHttpResponseMapper.java | 2 +
.../ConnectorPurgeRequestEndpointMerger.java | 86 +++++++++++++
.../http/StandardHttpResponseMapperTest.java | 44 +++++++
.../ConnectorPurgeRequestEndpointMergerTest.java | 134 +++++++++++++++++++++
.../endpoints/DropRequestEndpointMergerTest.java | 103 ++++++++++++++++
5 files changed, 369 insertions(+)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMapper.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMapper.java
index 7c942385b31..8fdd9a498e0 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMapper.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMapper.java
@@ -31,6 +31,7 @@ import
org.apache.nifi.cluster.coordination.http.endpoints.ConnectorEndpointMerg
import
org.apache.nifi.cluster.coordination.http.endpoints.ConnectorFlowEndpointMerger;
import
org.apache.nifi.cluster.coordination.http.endpoints.ConnectorPropertyGroupEndpointMerger;
import
org.apache.nifi.cluster.coordination.http.endpoints.ConnectorPropertyGroupNamesEndpointMerger;
+import
org.apache.nifi.cluster.coordination.http.endpoints.ConnectorPurgeRequestEndpointMerger;
import
org.apache.nifi.cluster.coordination.http.endpoints.ConnectorStatusEndpointMerger;
import
org.apache.nifi.cluster.coordination.http.endpoints.ConnectorsEndpointMerger;
import
org.apache.nifi.cluster.coordination.http.endpoints.ControllerBulletinsEndpointMerger;
@@ -182,6 +183,7 @@ public class StandardHttpResponseMapper implements
HttpResponseMapper {
endpointMergers.add(new RuleViolationEndpointMerger());
endpointMergers.add(new DropRequestEndpointMerger());
endpointMergers.add(new DropAllFlowFilesRequestEndpointMerger());
+ endpointMergers.add(new ConnectorPurgeRequestEndpointMerger());
endpointMergers.add(new ListFlowFilesEndpointMerger());
endpointMergers.add(new ComponentStateEndpointMerger());
endpointMergers.add(new BulletinBoardEndpointMerger());
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ConnectorPurgeRequestEndpointMerger.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ConnectorPurgeRequestEndpointMerger.java
new file mode 100644
index 00000000000..1911c0622d5
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ConnectorPurgeRequestEndpointMerger.java
@@ -0,0 +1,86 @@
+/*
+ * 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.manager.NodeResponse;
+import org.apache.nifi.cluster.protocol.NodeIdentifier;
+import org.apache.nifi.web.api.dto.DropRequestDTO;
+import org.apache.nifi.web.api.entity.DropRequestEntity;
+
+import java.net.URI;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Pattern;
+
+public class ConnectorPurgeRequestEndpointMerger extends
AbstractSingleDTOEndpoint<DropRequestEntity, DropRequestDTO> {
+ private static final int TOTAL_PERCENT_COMPLETED = 100;
+ private static final Pattern CONNECTOR_PURGE_REQUESTS_URI =
Pattern.compile("/nifi-api/connectors/[a-f0-9\\-]{36}/purge-requests");
+ private static final Pattern CONNECTOR_PURGE_REQUEST_URI =
Pattern.compile("/nifi-api/connectors/[a-f0-9\\-]{36}/purge-requests/[a-f0-9\\-]{36}");
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ final String path = uri.getPath();
+ if ("POST".equalsIgnoreCase(method)) {
+ return CONNECTOR_PURGE_REQUESTS_URI.matcher(path).matches();
+ }
+
+ return ("GET".equalsIgnoreCase(method) ||
"DELETE".equalsIgnoreCase(method))
+ && CONNECTOR_PURGE_REQUEST_URI.matcher(path).matches();
+ }
+
+ @Override
+ protected Class<DropRequestEntity> getEntityClass() {
+ return DropRequestEntity.class;
+ }
+
+ @Override
+ protected DropRequestDTO getDto(final DropRequestEntity entity) {
+ return entity.getDropRequest();
+ }
+
+ @Override
+ protected void mergeResponses(final DropRequestDTO clientDto, final
Map<NodeIdentifier, DropRequestDTO> dtoMap,
+ final Set<NodeResponse> successfulResponses,
final Set<NodeResponse> problematicResponses) {
+ final Set<String> failureReasons = new LinkedHashSet<>();
+ boolean allFinished = true;
+ int percentCompleted = TOTAL_PERCENT_COMPLETED;
+
+ for (final DropRequestDTO nodeDto : dtoMap.values()) {
+ if (nodeDto.getFailureReason() != null) {
+ failureReasons.add(nodeDto.getFailureReason());
+ }
+
+ allFinished &= nodeDto.isFinished();
+ percentCompleted = Math.min(percentCompleted,
nodeDto.getPercentCompleted());
+ }
+
+ if (failureReasons.isEmpty()) {
+ percentCompleted = allFinished ? TOTAL_PERCENT_COMPLETED :
percentCompleted;
+ clientDto.setState(allFinished ? "Complete" : "In Progress");
+ } else {
+ final String failureReason = String.join("; ", failureReasons);
+ allFinished = true;
+ percentCompleted = TOTAL_PERCENT_COMPLETED;
+ clientDto.setFailureReason(failureReason);
+ clientDto.setState("Failed: " + failureReason);
+ }
+
+ clientDto.setFinished(allFinished);
+ clientDto.setPercentCompleted(percentCompleted);
+ }
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMapperTest.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMapperTest.java
new file mode 100644
index 00000000000..51552c6b0b4
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMapperTest.java
@@ -0,0 +1,44 @@
+/*
+ * 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;
+
+import org.apache.nifi.util.NiFiProperties;
+import org.junit.jupiter.api.Test;
+
+import java.net.URI;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class StandardHttpResponseMapperTest {
+ private static final String CONNECTOR_ID =
"abcdef00-abcd-abcd-abcd-abcdef000000";
+ private static final String REQUEST_ID =
"00000000-0000-0000-0000-000000000001";
+
+ private final StandardHttpResponseMapper mapper = new
StandardHttpResponseMapper(NiFiProperties.createBasicNiFiProperties((String)
null));
+
+ @Test
+ void testConnectorPurgeRequestsInterpreted() {
+
assertTrue(mapper.isResponseInterpreted(URI.create("/nifi-api/connectors/" +
CONNECTOR_ID + "/purge-requests"), "POST"));
+
assertTrue(mapper.isResponseInterpreted(URI.create("/nifi-api/connectors/" +
CONNECTOR_ID + "/purge-requests/" + REQUEST_ID), "GET"));
+
assertTrue(mapper.isResponseInterpreted(URI.create("/nifi-api/connectors/" +
CONNECTOR_ID + "/purge-requests/" + REQUEST_ID), "DELETE"));
+ }
+
+ @Test
+ void testUnsupportedConnectorPurgeRequestsNotInterpreted() {
+
assertFalse(mapper.isResponseInterpreted(URI.create("/nifi-api/connectors/" +
CONNECTOR_ID + "/purge-requests"), "GET"));
+ }
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/endpoints/ConnectorPurgeRequestEndpointMergerTest.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/endpoints/ConnectorPurgeRequestEndpointMergerTest.java
new file mode 100644
index 00000000000..a74399c0027
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/endpoints/ConnectorPurgeRequestEndpointMergerTest.java
@@ -0,0 +1,134 @@
+/*
+ * 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.dto.DropRequestDTO;
+import org.junit.jupiter.api.Test;
+
+import java.net.URI;
+import java.util.LinkedHashMap;
+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.assertTrue;
+
+class ConnectorPurgeRequestEndpointMergerTest {
+ private static final String HTTP_DELETE = "DELETE";
+ private static final String HTTP_GET = "GET";
+ private static final String HTTP_POST = "POST";
+ private static final String HTTP_PUT = "PUT";
+ private static final String CONNECTOR_ID =
"abcdef00-abcd-abcd-abcd-abcdef000000";
+ private static final String REQUEST_ID =
"00000000-0000-0000-0000-000000000001";
+ private static final String PURGE_REQUESTS_URI = "/nifi-api/connectors/" +
CONNECTOR_ID + "/purge-requests";
+ private static final String PURGE_REQUEST_URI = PURGE_REQUESTS_URI + "/" +
REQUEST_ID;
+ private static final String FAILURE_REASON = "Failed to purge queue";
+ private static final int NO_PERCENT_COMPLETED = 0;
+ private static final int PARTIAL_PERCENT_COMPLETED = 40;
+ private static final int TOTAL_PERCENT_COMPLETED = 100;
+
+ private final ConnectorPurgeRequestEndpointMerger merger = new
ConnectorPurgeRequestEndpointMerger();
+
+ @Test
+ void testCanHandleConnectorPurgeRequests() {
+ assertTrue(merger.canHandle(URI.create(PURGE_REQUESTS_URI),
HTTP_POST));
+ assertTrue(merger.canHandle(URI.create(PURGE_REQUEST_URI), HTTP_GET));
+ assertTrue(merger.canHandle(URI.create(PURGE_REQUEST_URI),
HTTP_DELETE));
+ }
+
+ @Test
+ void testCanHandleRejectsUnrelatedRequests() {
+ assertFalse(merger.canHandle(URI.create(PURGE_REQUESTS_URI),
HTTP_GET));
+ assertFalse(merger.canHandle(URI.create(PURGE_REQUESTS_URI),
HTTP_DELETE));
+ assertFalse(merger.canHandle(URI.create(PURGE_REQUEST_URI),
HTTP_POST));
+ assertFalse(merger.canHandle(URI.create(PURGE_REQUEST_URI), HTTP_PUT));
+
assertFalse(merger.canHandle(URI.create("/nifi-api/connectors/not-a-uuid/purge-requests"),
HTTP_POST));
+ assertFalse(merger.canHandle(URI.create(PURGE_REQUESTS_URI +
"/not-a-uuid"), HTTP_GET));
+ assertFalse(merger.canHandle(URI.create("/nifi-api/connectors/" +
CONNECTOR_ID + "/backlog-requests"), HTTP_POST));
+ assertFalse(merger.canHandle(URI.create("/nifi-api/processors/" +
CONNECTOR_ID + "/purge-requests"), HTTP_POST));
+ }
+
+ @Test
+ void testMergeWaitsForSlowestNode() {
+ final DropRequestDTO clientRequest = request(true,
TOTAL_PERCENT_COMPLETED, null);
+ final Map<NodeIdentifier, DropRequestDTO> requests = new
LinkedHashMap<>();
+ requests.put(nodeIdentifier(1), clientRequest);
+ requests.put(nodeIdentifier(2), request(false,
PARTIAL_PERCENT_COMPLETED, null));
+
+ merger.mergeResponses(clientRequest, requests, null, null);
+
+ assertFalse(clientRequest.isFinished());
+ assertEquals(PARTIAL_PERCENT_COMPLETED,
clientRequest.getPercentCompleted());
+ assertEquals("In Progress", clientRequest.getState());
+ }
+
+ @Test
+ void testMergeCompletesWhenAllNodesFinish() {
+ final DropRequestDTO clientRequest = request(true,
TOTAL_PERCENT_COMPLETED, null);
+ final Map<NodeIdentifier, DropRequestDTO> requests = new
LinkedHashMap<>();
+ requests.put(nodeIdentifier(1), clientRequest);
+ requests.put(nodeIdentifier(2), request(true, TOTAL_PERCENT_COMPLETED,
null));
+
+ merger.mergeResponses(clientRequest, requests, null, null);
+
+ assertTrue(clientRequest.isFinished());
+ assertEquals(TOTAL_PERCENT_COMPLETED,
clientRequest.getPercentCompleted());
+ assertEquals("Complete", clientRequest.getState());
+ }
+
+ @Test
+ void testMergePropagatesFailure() {
+ final DropRequestDTO clientRequest = request(false,
PARTIAL_PERCENT_COMPLETED, null);
+ final Map<NodeIdentifier, DropRequestDTO> requests = new
LinkedHashMap<>();
+ requests.put(nodeIdentifier(1), clientRequest);
+ requests.put(nodeIdentifier(2), request(true, NO_PERCENT_COMPLETED,
FAILURE_REASON));
+
+ merger.mergeResponses(clientRequest, requests, null, null);
+
+ assertTrue(clientRequest.isFinished());
+ assertEquals(TOTAL_PERCENT_COMPLETED,
clientRequest.getPercentCompleted());
+ assertEquals(FAILURE_REASON, clientRequest.getFailureReason());
+ assertEquals("Failed: " + FAILURE_REASON, clientRequest.getState());
+ }
+
+ private static DropRequestDTO request(
+ final boolean finished,
+ final int percentCompleted,
+ final String failureReason
+ ) {
+ final DropRequestDTO request = new DropRequestDTO();
+ request.setFinished(finished);
+ request.setPercentCompleted(percentCompleted);
+ request.setFailureReason(failureReason);
+ return request;
+ }
+
+ private static NodeIdentifier nodeIdentifier(final int index) {
+ return new NodeIdentifier(
+ "node-" + index,
+ "localhost",
+ 8000 + index,
+ "localhost",
+ 8100 + index,
+ "localhost",
+ 8200 + index,
+ 8300 + index,
+ false
+ );
+ }
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/endpoints/DropRequestEndpointMergerTest.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/endpoints/DropRequestEndpointMergerTest.java
new file mode 100644
index 00000000000..53825740fca
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/endpoints/DropRequestEndpointMergerTest.java
@@ -0,0 +1,103 @@
+/*
+ * 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.controller.queue.DropFlowFileState;
+import org.apache.nifi.web.api.dto.DropRequestDTO;
+import org.junit.jupiter.api.Test;
+
+import java.net.URI;
+import java.util.LinkedHashMap;
+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.assertTrue;
+
+class DropRequestEndpointMergerTest {
+ private static final String CONNECTION_ID =
"12345678-1234-1234-1234-123456789012";
+ private static final String REQUEST_ID =
"00000000-0000-0000-0000-000000000001";
+
+ private final DropRequestEndpointMerger merger = new
DropRequestEndpointMerger();
+
+ @Test
+ void testCanHandleFlowFileQueueDropRequests() {
+ assertTrue(merger.canHandle(URI.create("/nifi-api/flowfile-queues/" +
CONNECTION_ID + "/drop-requests"), "POST"));
+ assertTrue(merger.canHandle(URI.create("/nifi-api/flowfile-queues/" +
CONNECTION_ID + "/drop-requests/" + REQUEST_ID), "GET"));
+ assertTrue(merger.canHandle(URI.create("/nifi-api/flowfile-queues/" +
CONNECTION_ID + "/drop-requests/" + REQUEST_ID), "DELETE"));
+ }
+
+ @Test
+ void testCanHandleRejectsUnsupportedRequests() {
+
assertFalse(merger.canHandle(URI.create("/nifi-api/connectors/abcdef00-abcd-abcd-abcd-abcdef000000/purge-requests"),
"POST"));
+
assertFalse(merger.canHandle(URI.create("/nifi-api/flowfile-queues/not-a-uuid/drop-requests"),
"POST"));
+ }
+
+ @Test
+ void testMergeWaitsForAllNodesAndAggregatesCounts() {
+ final DropRequestDTO clientRequest = dropRequest(true, 0, 4, 40L, 6,
60L);
+ final Map<NodeIdentifier, DropRequestDTO> requests = new
LinkedHashMap<>();
+ requests.put(nodeIdentifier(1), clientRequest);
+ requests.put(nodeIdentifier(2), dropRequest(false, 3, 6, 60L, 7, 70L));
+
+ merger.mergeResponses(clientRequest, requests, null, null);
+
+ assertFalse(clientRequest.isFinished());
+ assertEquals(3, clientRequest.getCurrentCount());
+ assertEquals(30L, clientRequest.getCurrentSize());
+ assertEquals(10, clientRequest.getDroppedCount());
+ assertEquals(100L, clientRequest.getDroppedSize());
+ assertEquals(13, clientRequest.getOriginalCount());
+ assertEquals(130L, clientRequest.getOriginalSize());
+ assertEquals(76, clientRequest.getPercentCompleted());
+ }
+
+ private static DropRequestDTO dropRequest(
+ final boolean finished,
+ final int currentCount,
+ final int droppedCount,
+ final long droppedSize,
+ final int originalCount,
+ final long originalSize
+ ) {
+ final DropRequestDTO request = new DropRequestDTO();
+ request.setFinished(finished);
+ request.setCurrentCount(currentCount);
+ request.setCurrentSize(currentCount * 10L);
+ request.setDroppedCount(droppedCount);
+ request.setDroppedSize(droppedSize);
+ request.setOriginalCount(originalCount);
+ request.setOriginalSize(originalSize);
+ request.setState(finished ? DropFlowFileState.COMPLETE.toString() :
DropFlowFileState.DROPPING_FLOWFILES.toString());
+ return request;
+ }
+
+ private static NodeIdentifier nodeIdentifier(final int index) {
+ return new NodeIdentifier(
+ "node-" + index,
+ "localhost",
+ 8000 + index,
+ "localhost",
+ 8100 + index,
+ "localhost",
+ 8200 + index,
+ 8300 + index,
+ false
+ );
+ }
+}
