Relax Overly Strict Assertion in TransportShardBulkAction (#40940)

* Remove Overly Strict Assertion in TransportShardBulkAction

* In #39793 this assertion was added under the assumption that no exceptions would be thrown in this method, which turned out not to be correct and at the very least `org.elasticsearch.index.shard.IndexShardClosedException` can be thrown by `org.elasticsearch.index.shard.IndexShard.sync`
* Closes #40933

Author: original-brownbear

server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java

@@ -147,7 +147,7 @@ public static void performOnPrimary(
         Consumer<ActionListener<Void>> waitForMappingUpdate,
         ActionListener<PrimaryResult<BulkShardRequest, BulkShardResponse>> listener,
         ThreadPool threadPool) {
-        new ActionRunnable<PrimaryResult<BulkShardRequest, BulkShardResponse>>(listener) {
+        new ActionRunnable<>(listener) {
 
             private final Executor executor = threadPool.executor(ThreadPool.Names.WRITE);
 
@@ -189,11 +189,11 @@ public void onRejection(Exception e) {
             }
 
             private void finishRequest() {
-                listener.onResponse(
-                    new WritePrimaryResult<>(context.getBulkShardRequest(), context.buildShardResponse(), context.getLocationToSync(),
-                        null, context.getPrimary(), logger));
+                ActionListener.completeWith(listener,
+                    () -> new WritePrimaryResult<>(
+                        context.getBulkShardRequest(), context.buildShardResponse(), context.getLocationToSync(), null,
+                        context.getPrimary(), logger));
             }
-
         }.run();
     }
 

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/integration/AnnotationIndexIT.java

@@ -51,7 +51,6 @@ public void testNotCreatedWhenNoOtherMlIndices() {
         }
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/40933")
     public void testCreatedWhenAfterOtherMlIndex() throws Exception {
 
         Auditor auditor = new Auditor(client(), "node_1");