pvillard31 commented on code in PR #11875:
URL: https://github.com/apache/nifi/pull/11875#discussion_r4445114637
##########
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:
Can we also assert the response body of this second DELETE, so the reported
status for a rejected duplicate request is locked in?
##########
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:
If decommission() throws an Error, the status stays IN_PROGRESS and every
later DELETE is ignored. Should we also set FAILED for Throwable here?
##########
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:
The description says offload failures (IllegalNodeOffloadException) also
restore the node or request reconnection, but only the disconnect path is
handled here. Should offloadNode() and waitForOffloadToFinish() failures also
roll back, so the node is not left DISCONNECTED?
--
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]
