bbende commented on code in PR #11875:
URL: https://github.com/apache/nifi/pull/11875#discussion_r4445390201
##########
nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/lifecycle/ClusterDecommissionTask.java:
##########
@@ -72,7 +78,15 @@ public synchronized void decommission() throws
InterruptedException {
flowController.setClustered(false, null);
logger.info("Instructed FlowController to stop sending heartbeats to
Cluster Coordinator and take Cluster Disconnect actions");
- disconnectNode();
+ try {
+ disconnectNode();
+ } catch (final IllegalNodeDisconnectionException e) {
Review Comment:
Both offload-phase failures depend on the node leaving DISCONNECTED or
OFFLOADING after `waitForDisconnection()` confirmed it was DISCONNECTED. With
heartbeats stopped, only an explicit user action can cause that. If someone
reconnects the node, the reconnect handshake
(`StandardFlowService.loadFromConnectionResponse`) already calls
`setClustered(true, ...)` and `startHeartbeating()`, so it isn't stranded. If
someone deletes the node, rolling back or requesting reconnection would undo
the user's delete. A failed offload notification is already retried:
`offloadAsynchronously` resets the state to DISCONNECTED and the loop in
`offloadNode()` tries again. With this PR, any of these failures shows up as
`Decommission Status: FAILED` instead of disappearing into the uncaught
exception handler. The PR description overstated this; I've updated it to say
that only the disconnect path rolls back.
##########
nifi-framework-bundle/nifi-framework/nifi-runtime/src/main/java/org/apache/nifi/runtime/HealthClusterHttpHandler.java:
##########
@@ -85,17 +97,40 @@ public void handle(final HttpExchange exchange) throws
IOException {
}
}
- private void startDecommission() {
+ private synchronized void startDecommission() {
+ final DecommissionStatus currentDecommissionStatus =
decommissionStatus;
+ if (currentDecommissionStatus != null &&
currentDecommissionStatus.state() == DecommissionState.IN_PROGRESS) {
+ logger.info("Decommission requested but a Decommission is already
in progress");
+ return;
+ }
+
final DecommissionTask decommissionTask = server.getDecommissionTask();
+ decommissionStatus = IN_PROGRESS;
Thread.ofVirtual().name(DecommissionTask.class.getSimpleName()).start(() -> {
try {
decommissionTask.decommission();
+ decommissionStatus = COMPLETED;
} catch (final InterruptedException e) {
Thread.currentThread().interrupt();
+ decommissionStatus = DecommissionStatus.failed(e);
+ } catch (final RuntimeException e) {
Review Comment:
Done, the decommission thread now catches `Throwable`, logs it, and records
`FAILED`, so later DELETE requests can start a new attempt.
##########
nifi-framework-bundle/nifi-framework/nifi-runtime/src/test/java/org/apache/nifi/runtime/StandardManagementServerTest.java:
##########
@@ -124,8 +136,57 @@ void testGetHealthClusterConnected() throws Exception {
final InetSocketAddress serverAddress =
standardManagementServer.getServerAddress();
assertNotSame(initialBindAddress.getPort(),
serverAddress.getPort());
- assertResponseStatusCode(HEALTH_CLUSTER_URI, serverAddress,
GET_METHOD, HttpURLConnection.HTTP_OK);
+ final String responseBody =
assertResponseStatusCode(HEALTH_CLUSTER_URI, serverAddress, GET_METHOD,
HttpURLConnection.HTTP_OK);
+ assertEquals("Cluster Status: CONNECTED\n", responseBody);
+ } finally {
+ standardManagementServer.stop();
+ }
+ }
+
+ @Test
+ void testDeleteHealthCluster() throws Exception {
+ final InetSocketAddress initialBindAddress = new
InetSocketAddress(LOCALHOST, 0);
+ final StandardManagementServer standardManagementServer = new
StandardManagementServer(initialBindAddress, server);
+
+
when(server.getClusterDetailsFactory()).thenReturn(clusterDetailsFactory);
+
when(clusterDetailsFactory.getConnectionState()).thenReturn(ConnectionState.CONNECTED);
+ when(server.getDecommissionTask()).thenReturn(decommissionTask);
+
+ final CountDownLatch decommissionLatch = new CountDownLatch(1);
+ doAnswer(invocation -> {
+ decommissionLatch.await();
+ throw new IllegalStateException(DECOMMISSION_FAILURE);
+ }).when(decommissionTask).decommission();
+
+ try {
+ standardManagementServer.start();
+
+ final InetSocketAddress serverAddress =
standardManagementServer.getServerAddress();
+
+ final String deleteResponseBody =
assertResponseStatusCode(HEALTH_CLUSTER_URI, serverAddress, DELETE_METHOD,
HttpURLConnection.HTTP_ACCEPTED);
+ assertEquals("Cluster Status: OFFLOADING\nDecommission Status:
IN_PROGRESS\n", deleteResponseBody);
+ verify(decommissionTask,
timeout(TIMEOUT.toMillis())).decommission();
+
+ final String inProgressResponseBody =
assertResponseStatusCode(HEALTH_CLUSTER_URI, serverAddress, GET_METHOD,
HttpURLConnection.HTTP_OK);
+ assertEquals("Cluster Status: CONNECTED\nDecommission Status:
IN_PROGRESS\n", inProgressResponseBody);
+
+ // A request while a decommission is in progress is accepted
without starting another decommission
+ assertResponseStatusCode(HEALTH_CLUSTER_URI, serverAddress,
DELETE_METHOD, HttpURLConnection.HTTP_ACCEPTED);
Review Comment:
Done, the test now asserts the duplicate DELETE returns `Cluster Status:
OFFLOADING` with `Decommission Status: IN_PROGRESS`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
