Simplify response of /handoff API, fix test flakiness

docs/api-reference/supervisor-api.md

@@ -3662,9 +3662,8 @@ Content-Type: application/json
 <details>
   <summary>202 Accepted</summary>
 
-```json
-{}
-```
+*Empty response*
+
 </details>
 
 ### Shut down a supervisor

embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaFaultToleranceTest.java

@@ -19,7 +19,6 @@
 
 package org.apache.druid.testing.embedded.indexing;
 
-import com.fasterxml.jackson.core.type.TypeReference;
 import org.apache.druid.indexer.TaskStatusPlus;
 import org.apache.druid.indexing.overlord.supervisor.SupervisorSpec;
 import org.apache.druid.java.util.common.StringUtils;
@@ -65,7 +64,7 @@ public void verifyAndTearDown()
     verifyRowCount(totalRecords);
   }
 
-  @ParameterizedTest
+  @ParameterizedTest(name = "useTransactions={0}")
   @ValueSource(booleans = {true, false})
   public void test_supervisorRecovers_afterOverlordRestart(boolean useTransactions) throws Exception
   {
@@ -107,7 +106,7 @@ public void test_supervisorRecovers_afterHistoricalRestart() throws Exception
     totalRecords += publish1kRecords(topic, useTransactions);
   }
 
-  @ParameterizedTest
+  @ParameterizedTest(name = "useTransactions={0}")
   @ValueSource(booleans = {true, false})
   public void test_supervisorRecovers_afterSuspendResume(boolean useTransactions)
   {
@@ -121,7 +120,7 @@ public void test_supervisorRecovers_afterSuspendResume(boolean useTransactions)
     totalRecords += publish1kRecords(topic, useTransactions);
   }
 
-  @ParameterizedTest
+  @ParameterizedTest(name = "useTransactions={0}")
   @ValueSource(booleans = {true, false})
   public void test_supervisorRecovers_afterChangeInTopicPartitions(boolean useTransactions)
   {
@@ -149,7 +148,7 @@ public void test_supervisorLaunchesNewTask_ifEarlyHandoff()
     cluster.callApi().serviceClient().onLeaderOverlord(
         mapper -> new RequestBuilder(HttpMethod.POST, path)
             .jsonContent(mapper, Map.of("taskGroupIds", List.of(0, 1))),
-        new TypeReference<>() {}
+        null
     );
 
     // Wait for the handoff notice to be processed

embedded-tests/src/test/java/org/apache/druid/testing/embedded/server/FaultyClusterTest.java

@@ -87,6 +87,7 @@ public void test_overlord_skipsCleanupOfPendingSegments()
     cluster.callApi().postSupervisor(supervisorSpec.createSuspendedSpec());
     kafkaServer.deleteTopic(topic);
 
+    cluster.callApi().waitForAllSegmentsToBeAvailable(dataSource, coordinator, broker);
     verifyRowCount(recordCount);
 
     // Verify that pending segments are not cleaned up

indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResource.java

@@ -432,9 +432,7 @@ public Response handoffTaskGroups(
         manager -> {
           try {
             if (manager.handoffTaskGroupsEarly(id, taskGroupIds)) {
-              return Response.status(Response.Status.ACCEPTED)
-                             .entity(Map.of()) // empty json object to allow deserialization by the client
-                             .build();
+              return Response.status(Response.Status.ACCEPTED).build();
             } else {
               return Response.status(Response.Status.NOT_FOUND)
                              .entity(ImmutableMap.of("error", StringUtils.format("Supervisor was not found [%s]", id)))

indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResourceTest.java

@@ -1374,6 +1374,28 @@ public void testSpecPostMergeFallsBackToProvidedTaskCountMin()
     EasyMock.verify(newSpec.getIoConfig());
   }
 
+  @Test
+  public void test_handoffTaskGroups_returnsAccepted()
+  {
+    final SupervisorSpec spec = createTestSpec(1, 1);
+    final SupervisorResource.HandoffTaskGroupsRequest handoffRequest
+        = new SupervisorResource.HandoffTaskGroupsRequest(List.of(0));
+
+    EasyMock.expect(taskMaster.getSupervisorManager())
+            .andReturn(Optional.of(supervisorManager))
+            .times(1);
+    EasyMock.expect(supervisorManager.handoffTaskGroupsEarly(spec.getId(), handoffRequest.getTaskGroupIds()))
+            .andReturn(true)
+            .times(1);
+    replayAll();
+
+    final Response response = supervisorResource.handoffTaskGroups(spec.getId(), handoffRequest);
+    Assert.assertEquals(202, response.getStatus());
+    Assert.assertNull(response.getEntity());
+
+    verifyAll();
+  }
+
   private TestSeekableStreamSupervisorSpec createTestSpec(Integer taskCount, int taskCountMin)
   {
     HashMap<String, Object> autoScalerConfig = new HashMap<>();