sklaha commented on code in PR #275:
URL: https://github.com/apache/cassandra-sidecar/pull/275#discussion_r2619914367
##########
server/src/main/java/org/apache/cassandra/sidecar/modules/CassandraOperationsModule.java:
##########
@@ -180,6 +181,29 @@ VertxRoute cassandraNodeDrainRoute(RouteBuilder.Factory
factory,
return factory.buildRouteWithHandler(nodeDrainHandler);
}
+ @PUT
+ @Path(ApiEndpointsV1.NODE_MOVE_ROUTE)
+ @Operation(summary = "Move node to new token",
+ description = "Moves the Cassandra node to a new token in the
ring")
+ @APIResponse(description = "Node move operation completed successfully",
+ responseCode = "200",
+ content = @Content(mediaType = "application/json",
+ schema = @Schema(implementation =
OperationalJobResponse.class)))
+ @APIResponse(description = "Node move operation initiated successfully",
Review Comment:
done
##########
integration-tests/src/integrationTest/org/apache/cassandra/sidecar/routes/CassandraNodeOperationsIntegrationTest.java:
##########
@@ -88,17 +100,160 @@ void testNodeDrainOperationSuccess()
// Validate the operational job status using the OperationalJobHandler
String jobId = responseBody.getString("jobId");
- validateOperationalJobStatus(jobId, "drain");
+ validateOperationalJobStatus(jobId, "drain",
OperationalJobStatus.SUCCEEDED);
+ }
+
+
+ @Test
+ void testNodeMoveOperationSuccess()
+ {
+ // Use a test token - this is a valid token for Murmur3Partitioner
+ String testToken = "123456789";
+ String requestBody = "{\"newToken\":\"" + testToken + "\"}";
+
+ // Validate that the node owns a different token than testToken
+ String currentToken = getCurrentTokenForNode("localhost");
+ assertThat(currentToken).isNotEqualTo(testToken);
+
+ // Initiate move operation
+ HttpResponse<Buffer> moveResponse = getBlocking(
+ trustedClient().put(serverWrapper.serverPort, "localhost",
ApiEndpointsV1.NODE_MOVE_ROUTE)
+ .putHeader("content-type", "application/json")
+ .sendBuffer(Buffer.buffer(requestBody)));
+
+ assertThat(moveResponse.statusCode()).isIn(OK.code(), ACCEPTED.code());
+
+ JsonObject responseBody = moveResponse.bodyAsJsonObject();
+ assertThat(responseBody).isNotNull();
+ assertThat(responseBody.getString("jobId")).isNotNull();
+ assertThat(responseBody.getString("operation")).isEqualTo("move");
+ assertThat(responseBody.getString("jobStatus")).isIn(
+ OperationalJobStatus.CREATED.name(),
+ OperationalJobStatus.RUNNING.name(),
+ OperationalJobStatus.SUCCEEDED.name()
+ );
+
+ // Verify the job eventually completes (or at least gets processed)
+ loopAssert(30, 500, () -> {
+ HttpResponse<Buffer> streamStatsResponse = getBlocking(
+ trustedClient().get(serverWrapper.serverPort, "localhost",
ApiEndpointsV1.STREAM_STATS_ROUTE)
+ .send());
+
+ assertThat(streamStatsResponse.statusCode()).isEqualTo(OK.code());
+
+ JsonObject streamStats = streamStatsResponse.bodyAsJsonObject();
+ assertThat(streamStats).isNotNull();
+ // The operationMode should be either NORMAL (completed) or MOVING
(in progress)
+ assertThat(streamStats.getString("operationMode")).isIn("NORMAL",
"MOVING");
+ });
+
+ // Validate the operational job status using the OperationalJobHandler
+ String jobId = responseBody.getString("jobId");
+ validateOperationalJobStatus(jobId, "move",
OperationalJobStatus.SUCCEEDED);
+
+ // Validate that the node actually owns the new token
+ currentToken = getCurrentTokenForNode("localhost");
+ assertThat(currentToken).isEqualTo(testToken);
+ }
+
+ /**
+ * Tests the failure case of node move operation when attempting to move
to a token
+ * already owned by another node in the cluster.
+ * <p>
+ * This test validates that:
+ * - The system properly rejects invalid move operations that would create
token conflicts
+ * - The move operation fails with OperationalJobStatus.FAILED when
targeting an existing token
+ * - The original node retains its initial token after the failed move
attempt
+ * <p>
+ * Token conflicts must be prevented to maintain cluster integrity, as
having multiple
+ * nodes own the same token would break the consistent hashing ring and
cause data
+ * distribution issues.
+ */
+ @Test
+ void testNodeMoveOperationFailure()
+ {
+ // Get a token already owned by a node
+ String testToken = getCurrentTokenForNode("localhost2");
+ String requestBody = "{\"newToken\":\"" + testToken + "\"}";
+
+ // Validate that the node owns a different token than testToken
+ String initialToken = getCurrentTokenForNode("localhost");
+ assertThat(initialToken).isNotEqualTo(testToken);
+
+ // Initiate move operation
+ HttpResponse<Buffer> moveResponse = getBlocking(
+ trustedClient().put(serverWrapper.serverPort, "localhost",
ApiEndpointsV1.NODE_MOVE_ROUTE)
+ .putHeader("content-type", "application/json")
+ .sendBuffer(Buffer.buffer(requestBody)));
+
+ assertThat(moveResponse.statusCode()).isIn(OK.code(), ACCEPTED.code());
+
+ JsonObject responseBody = moveResponse.bodyAsJsonObject();
+ assertThat(responseBody).isNotNull();
+ assertThat(responseBody.getString("jobId")).isNotNull();
+ assertThat(responseBody.getString("operation")).isEqualTo("move");
+ assertThat(responseBody.getString("jobStatus")).isIn(
+ OperationalJobStatus.CREATED.name(),
+ OperationalJobStatus.RUNNING.name(),
+ OperationalJobStatus.FAILED.name()
+ );
+
+ // Verify the job eventually completes (or at least gets processed)
+ loopAssert(30, 500, () -> {
+ HttpResponse<Buffer> streamStatsResponse = getBlocking(
+ trustedClient().get(serverWrapper.serverPort, "localhost",
ApiEndpointsV1.STREAM_STATS_ROUTE)
+ .send());
+
+ assertThat(streamStatsResponse.statusCode()).isEqualTo(OK.code());
+
+ JsonObject streamStats = streamStatsResponse.bodyAsJsonObject();
+ assertThat(streamStats).isNotNull();
+ // The operationMode should be either NORMAL (completed) or MOVING
(in progress)
+ assertThat(streamStats.getString("operationMode")).isIn("NORMAL",
"MOVING");
+ });
+
+ // Validate the operational job status using the OperationalJobHandler
+ String jobId = responseBody.getString("jobId");
+ validateOperationalJobStatus(jobId, "move",
OperationalJobStatus.FAILED);
+
+ // Validate that the node didn't move
+ String currentToken = getCurrentTokenForNode("localhost");
+ assertThat(currentToken).isEqualTo(initialToken);
+ assertThat(currentToken).isNotEqualTo(testToken);
+ }
+
+ /**
+ * Gets the current token for the specified node by querying the ring
endpoint.
+ *
+ * @param node the node hostname to get the token for
+ * @return the token currently owned by the specified node
+ */
+ private String getCurrentTokenForNode(String node)
+ {
+ HttpResponse<Buffer> ringResponse = getBlocking(
+ trustedClient().get(serverWrapper.serverPort, node,
ApiEndpointsV1.RING_ROUTE)
+ .send());
+
+ assertThat(ringResponse.statusCode()).isEqualTo(OK.code());
+
+ RingResponse ring = ringResponse.bodyAsJson(RingResponse.class);
+ assertThat(ring).isNotNull();
+
+ RingEntry ringEntry = ring.stream()
+ .filter(entry -> entry.fqdn().equals(node))
Review Comment:
I didn't enable vnodes for this test and it succeeds.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
