DBZ-7904 Enhancing the threads utility class for broader use

Forced Shutdown for errored pubsub managed channel Forced Shutdown for errored pubsub managed channel DBZ-7904 Enhancing the threads utility class for broader use
debezium · Jun 3, 2024 · ccbcc14 · ccbcc14
1 parent bfe425b
commit ccbcc14
Showing 1 changed file with 28 additions and 5 deletions.
diff --git a/debezium-server-pubsub/src/main/java/io/debezium/server/pubsub/PubSubChangeConsumer.java b/debezium-server-pubsub/src/main/java/io/debezium/server/pubsub/PubSubChangeConsumer.java
@@ -11,10 +11,9 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
+import java.util.concurrent.*;
 
+import io.debezium.util.Threads;
 import jakarta.annotation.PostConstruct;
 import jakarta.annotation.PreDestroy;
 import jakarta.enterprise.context.Dependent;
@@ -131,6 +130,9 @@ public interface PublisherBuilder {
     @ConfigProperty(name = PROP_PREFIX + "wait.message.delivery.timeout.ms", defaultValue = "30000")
     Integer waitMessageDeliveryTimeout;
 
+    @ConfigProperty(name = PROP_PREFIX + "channel.shutdown.timeout.ms", defaultValue = "30000")
+    Integer channelShutdownTimeout;
+
     @ConfigProperty(name = PROP_PREFIX + "address")
     Optional<String> address;
 
@@ -213,12 +215,33 @@ void close() {
                 publisher.shutdown();
             }
             catch (Exception e) {
-                LOGGER.warn("Exception while closing publisher: {}", e);
+                LOGGER.warn("Exception while closing publisher: {}", e.getMessage(), e);
             }
         });
+        shutdownChannel(channel);
+    }
+
+    void shutdownChannel(ManagedChannel channel) {
 
         if (channel != null && !channel.isShutdown()) {
-            channel.shutdown();
+            ExecutorService executor = Threads.newSingleThreadExecutor(this.getClass(), projectId, "managed-channel-shutdown-coordinator");
+
+            Future<?> future = executor.submit(() -> {
+                channel.shutdown();
+                try {
+                    channel.awaitTermination(channelShutdownTimeout, TimeUnit.MILLISECONDS);
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                }
+            });
+
+            try {
+                future.get(channelShutdownTimeout, TimeUnit.MILLISECONDS);
+            } catch (Exception e) {
+                LOGGER.warn("Exception while shutting down the managed channel {}", e.getMessage(), e);
+            } finally {
+                executor.shutdownNow();
+            }
         }
     }