Fix Incorrect Transport Response Handler Type (#38264)

* Fix Incorrect Transport Response Handler Type
* The response type here is not empty and was always wrong but this only became visible now that 0a604e3b24 was introduced
   * As a result of 0a604e3b24 we started actually handling the response
of this request and logging/handling exceptions before that we simply dropped the classcast exception here quietly using the empty response handler
* fix busy assert not handling `Exception`
* Closes #38226
* Closes #38256

server/src/main/java/org/elasticsearch/snapshots/SnapshotShardsService.java

@@ -67,10 +67,9 @@ import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.repositories.IndexId;
 import org.elasticsearch.repositories.Repository;
 import org.elasticsearch.threadpool.ThreadPool;
-import org.elasticsearch.transport.EmptyTransportResponseHandler;
 import org.elasticsearch.transport.TransportException;
 import org.elasticsearch.transport.TransportRequestDeduplicator;
-import org.elasticsearch.transport.TransportResponse;
+import org.elasticsearch.transport.TransportResponseHandler;
 import org.elasticsearch.transport.TransportService;
 
 import java.io.IOException;
@@ -508,9 +507,16 @@ public class SnapshotShardsService extends AbstractLifecycleComponent implements
                 }
             },
             (req, reqListener) -> transportService.sendRequest(transportService.getLocalNode(), UPDATE_SNAPSHOT_STATUS_ACTION_NAME, req,
-                new EmptyTransportResponseHandler(ThreadPool.Names.SAME) {
+                new TransportResponseHandler<UpdateIndexShardSnapshotStatusResponse>() {
                     @Override
-                    public void handleResponse(TransportResponse.Empty response) {
+                    public UpdateIndexShardSnapshotStatusResponse read(StreamInput in) throws IOException {
+                        final UpdateIndexShardSnapshotStatusResponse response = new UpdateIndexShardSnapshotStatusResponse();
+                        response.readFrom(in);
+                        return response;
+                    }
+
+                    @Override
+                    public void handleResponse(UpdateIndexShardSnapshotStatusResponse response) {
                         reqListener.onResponse(null);
                     }
 
@@ -518,6 +524,11 @@ public class SnapshotShardsService extends AbstractLifecycleComponent implements
                     public void handleException(TransportException exp) {
                         reqListener.onFailure(exp);
                     }
+
+                    @Override
+                    public String executor() {
+                        return ThreadPool.Names.SAME;
+                    }
                 })
         );
     }

server/src/test/java/org/elasticsearch/snapshots/DedicatedClusterSnapshotRestoreIT.java

@@ -988,7 +988,6 @@ public class DedicatedClusterSnapshotRestoreIT extends AbstractSnapshotIntegTest
      * can be restored when the node the shrunken index was created on is no longer part of
      * the cluster.
      */
-    @AwaitsFix(bugUrl="https://github.com/elastic/elasticsearch/issues/38256")
     public void testRestoreShrinkIndex() throws Exception {
         logger.info("-->  starting a master node and a data node");
         internalCluster().startMasterOnlyNode();

server/src/test/java/org/elasticsearch/snapshots/SharedClusterSnapshotRestoreIT.java

@@ -3637,7 +3637,6 @@ public class SharedClusterSnapshotRestoreIT extends AbstractSnapshotIntegTestCas
     }
 
     @TestLogging("org.elasticsearch.snapshots:TRACE")
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/38226")
     public void testAbortedSnapshotDuringInitDoesNotStart() throws Exception {
         final Client client = client();
 
@@ -3684,8 +3683,14 @@ public class SharedClusterSnapshotRestoreIT extends AbstractSnapshotIntegTestCas
 
             // The deletion must set the snapshot in the ABORTED state
             assertBusy(() -> {
-                SnapshotsStatusResponse status = client.admin().cluster().prepareSnapshotStatus("repository").setSnapshots("snap").get();
+                try {
-                assertThat(status.getSnapshots().iterator().next().getState(), equalTo(State.ABORTED));
+                    SnapshotsStatusResponse status =
+                        client.admin().cluster().prepareSnapshotStatus("repository").setSnapshots("snap").get();
+                    assertThat(status.getSnapshots().iterator().next().getState(), equalTo(State.ABORTED));
+                } catch (Exception e) {
+                    // Force assertBusy to retry on every exception
+                    throw new AssertionError(e);
+                }
             });
 
             // Now unblock the repository