Backport PR #16671 to 8.16: PipelineBusV2 deadlock proofing

logstash-core/src/main/java/org/logstash/plugins/pipeline/PipelineBusV2.java

@@ -141,16 +141,20 @@ public AddressState.ReadOnly mutate(final String address,
 
                 consumer.accept(addressState);
 
-                // If this addressState has a listener, ensure that any waiting
+                return addressState.isEmpty() ? null : addressState;
+            });
+
+            if (result == null) {
+                return null;
+            } else {
+                // If the resulting addressState had a listener, ensure that any waiting
                 // threads get notified so that they can resume immediately
-                final PipelineInput currentInput = addressState.getInput();
+                final PipelineInput currentInput = result.getInput();
                 if (currentInput != null) {
                     synchronized (currentInput) { currentInput.notifyAll(); }
                 }
-
-                return addressState.isEmpty() ? null : addressState;
-            });
-            return result == null ? null : result.getReadOnlyView();
+                return result.getReadOnlyView();
+            }
         }
 
         private AddressState.ReadOnly get(final String address) {

logstash-core/src/test/java/org/logstash/plugins/pipeline/PipelineBusTest.java

@@ -23,15 +23,24 @@
 import org.junit.Test;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
 
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 import org.logstash.RubyUtil;
 import org.logstash.ext.JrubyEventExtLibrary;
 
 import java.time.Duration;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.LongAdder;
 import java.util.stream.Stream;
@@ -307,6 +316,56 @@ public void whenInBlockingModeInputsShutdownLast() throws InterruptedException {
         assertThat(bus.getAddressState(address)).isNotPresent();
     }
 
+    @Test
+    public void blockingShutdownDeadlock() throws InterruptedException {
+        final ExecutorService executor = Executors.newFixedThreadPool(10);
+        try {
+            for (int i = 0; i < 100; i++) {
+                bus.registerSender(output, addresses);
+                bus.listen(input, address);
+                bus.setBlockOnUnlisten(true);
+
+                // we use a CountDownLatch to increase the likelihood
+                // of simultaneous execution
+                final CountDownLatch startLatch = new CountDownLatch(2);
+                final CompletableFuture<Void> unlistenFuture = CompletableFuture.runAsync(asRunnable(() -> {
+                    startLatch.countDown();
+                    startLatch.await();
+                    bus.unlisten(input, address);
+                }), executor);
+                final CompletableFuture<Void> unregisterFuture = CompletableFuture.runAsync(asRunnable(() -> {
+                    startLatch.countDown();
+                    startLatch.await();
+                    bus.unregisterSender(output, addresses);
+                }), executor);
+
+                // ensure that our tasks all exit successfully, quickly
+                assertThatCode(() -> CompletableFuture.allOf(unlistenFuture, unregisterFuture).get(1, TimeUnit.SECONDS))
+                        .withThreadDumpOnError()
+                        .withFailMessage("Expected unlisten and unregisterSender to not deadlock, but they did not return in a reasonable amount of time in the <%s>th iteration", i)
+                        .doesNotThrowAnyException();
+            }
+        } finally {
+            executor.shutdownNow();
+        }
+    }
+
+    @FunctionalInterface
+    interface ExceptionalRunnable<E extends Throwable> {
+        void run() throws E;
+    }
+
+    private Runnable asRunnable(final ExceptionalRunnable<?> exceptionalRunnable) {
+        return () -> {
+            try {
+                exceptionalRunnable.run();
+            } catch (Throwable e) {
+                throw new RuntimeException(e);
+            }
+        };
+    }
+
+
     @Test
     public void whenInputFailsOutputRetryOnlyNotYetDelivered() throws InterruptedException {
         bus.registerSender(output, addresses);