Fork listener#onFailure in PrimaryReplicaSyncer (#70506)

We assert that the snapshot isn't closed on a transport thread, but we
close it without forking off the transport thread in case of a failure.
With this commit we fork on failure too.

Relates #69949
Closes #70407

server/src/main/java/org/elasticsearch/index/shard/PrimaryReplicaSyncer.java

@@ -222,7 +222,24 @@ public class PrimaryReplicaSyncer {
         @Override
         public void onFailure(Exception e) {
             if (closed.compareAndSet(false, true)) {
-                listener.onFailure(e);
+                executor.execute(new AbstractRunnable() {
+                    @Override
+                    public void onFailure(Exception ex) {
+                        e.addSuppressed(ex);
+
+                        // We are on the generic threadpool so shouldn't be rejected, and listener#onFailure shouldn't throw anything,
+                        // so getting here should be impossible.
+                        assert false : e;
+
+                        // Notify the listener on the current thread anyway, just in case.
+                        listener.onFailure(e);
+                    }
+
+                    @Override
+                    protected void doRun() {
+                        listener.onFailure(e);
+                    }
+                });
             }
         }