diff --git a/src/main/java/com/googlesource/gerrit/plugins/nats/Configuration.java b/src/main/java/com/googlesource/gerrit/plugins/nats/Configuration.java
index b05ecdd..838f9ab 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/nats/Configuration.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/nats/Configuration.java
@@ -79,7 +79,7 @@
             .maxAckPending(pluginConfig.getLong("maxAckPending", 1000L))
             .deliverPolicy(DeliverPolicy.All)
             .buildPushSubscribeOptions();
-    shutdownTimeoutMs = pluginConfig.getInt("shutdownTimeoutMs", 1000);
+    shutdownTimeoutMs = pluginConfig.getInt("shutdownTimeoutMs", 30000);
 
     logger.atInfo().log(
         "NATS client configuration: sendStreamEvents: %b, streamEventsSubject: %s, natsOptions: %s, publishOptions: %s",
diff --git a/src/main/java/com/googlesource/gerrit/plugins/nats/NatsBrokerLifeCycleManager.java b/src/main/java/com/googlesource/gerrit/plugins/nats/NatsBrokerLifeCycleManager.java
index 7c286e0..bfd8512 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/nats/NatsBrokerLifeCycleManager.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/nats/NatsBrokerLifeCycleManager.java
@@ -20,6 +20,7 @@
 import com.google.gerrit.extensions.events.LifecycleListener;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
+import io.nats.client.Connection;
 import io.nats.client.JetStreamApiException;
 import io.nats.client.JetStreamManagement;
 import io.nats.client.api.StorageType;
@@ -27,8 +28,13 @@
 import io.nats.client.api.StreamInfo;
 import java.io.IOException;
 import java.security.ProviderException;
+import java.time.Duration;
 import java.time.format.DateTimeFormatter;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.logging.Level;
 
 @Singleton
@@ -36,6 +42,7 @@
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 
   private final Configuration config;
+  private final Connection connection;
   private final Set<TopicSubscriber> consumers;
   private final BrokerApi brokerApi;
   private JetStreamManagement jetStreamManagement;
@@ -44,12 +51,14 @@
   public NatsBrokerLifeCycleManager(
       Configuration config,
       JetStreamManagement jetStreamManagement,
-      Set<TopicSubscriber> consumers,
-      BrokerApi brokerApi) {
+      Connection connection,
+      BrokerApi brokerApi,
+      Set<TopicSubscriber> consumers) {
     this.config = config;
     this.jetStreamManagement = jetStreamManagement;
-    this.consumers = consumers;
+    this.connection = connection;
     this.brokerApi = brokerApi;
+    this.consumers = consumers;
   }
 
   @Override
@@ -96,6 +105,29 @@
 
   @Override
   public void stop() {
+    drainConnection();
     brokerApi.disconnect();
+    try {
+      connection.close();
+    } catch (InterruptedException e) {
+      logger.at(Level.SEVERE).withCause(e).log(
+          "NATS broker - stopping connection failed with error");
+    }
+  }
+
+  private void drainConnection() {
+    try {
+      int shutdownTimeoutMs = config.getShutdownTimeoutMs();
+      Duration shutdownTimeout = Duration.ofMillis(shutdownTimeoutMs);
+      logger.atInfo().log(
+          "NATS consumer - Waiting up to '%s' milliseconds to drain connection to stream '%s'",
+          shutdownTimeoutMs, config.getStreamName());
+      CompletableFuture<Boolean> f = connection.drain(shutdownTimeout);
+      f.get(shutdownTimeoutMs, TimeUnit.MILLISECONDS);
+    } catch (TimeoutException e) {
+      logger.at(Level.WARNING).withCause(e).log("NATS broker - graceful shutdown failed with timeout");
+    } catch (InterruptedException | ExecutionException e) {
+      logger.at(Level.SEVERE).withCause(e).log("NATS broker - graceful shutdown failed with error");
+    }
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/nats/NatsConsumer.java b/src/main/java/com/googlesource/gerrit/plugins/nats/NatsConsumer.java
index 52cf721..918a7ac 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/nats/NatsConsumer.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/nats/NatsConsumer.java
@@ -32,12 +32,10 @@
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.security.ProviderException;
-import java.time.Duration;
 import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
+import java.util.logging.Level;
 
 // import software.amazon.kinesis.coordinator.Scheduler;
 
@@ -50,7 +48,7 @@
   private final Configuration config;
 
   private java.util.function.Consumer<Event> messageProcessor;
-  private String streamName;
+  private String subjectName;
   private AtomicBoolean resetOffset = new AtomicBoolean(false);
   private final Connection connection;
   private final JetStream jetStream;
@@ -61,25 +59,26 @@
 
   @Inject
   public NatsConsumer(
+      @EventGson Gson gson,
       Configuration config,
       Connection connection,
       JetStream jetStream,
-      JetStreamManagement jetStreamManagement,
-      @EventGson Gson gson) {
+      JetStreamManagement jetStreamManagement) {
+    this.gson = gson;
     this.config = config;
     this.connection = connection;
-    this.dispatcher = connection.createDispatcher();
     this.jetStream = jetStream;
     this.jetStreamManagement = jetStreamManagement;
-    this.gson = gson;
+    // create one dispatcher per consumer
+    this.dispatcher = connection.createDispatcher();
   }
 
-  public void subscribe(String streamName, java.util.function.Consumer<Event> messageProcessor) {
-    this.streamName = streamName;
+  public void subscribe(String subjectName, java.util.function.Consumer<Event> messageProcessor) {
+    this.subjectName = subjectName;
     this.messageProcessor = messageProcessor;
-    String subject = getOrCreateSubject(streamName);
-    logger.atInfo().log("NATS consumer - subscribe to subject [%s]", streamName);
-    runReceiver(messageProcessor, subject);
+    String subject = getOrCreateSubject(subjectName);
+    subscription = runReceiver(messageProcessor, subject);
+    logger.atInfo().log("NATS consumer - subscribed to subject [%s]", subjectName);
   }
 
   private String getOrCreateSubject(String name) {
@@ -107,49 +106,43 @@
         name, sc.getName(), subjects);
   }
 
-  private void runReceiver(java.util.function.Consumer<Event> messageProcessor, String subject) {
+  private JetStreamSubscription runReceiver(
+      java.util.function.Consumer<Event> messageProcessor, String subject) {
     try {
-      this.subscription =
-          jetStream.subscribe(
-              subject,
-              dispatcher,
-              new MessageHandler() {
+      return jetStream.subscribe(
+          subject,
+          dispatcher,
+          new MessageHandler() {
 
-                @Override
-                public void onMessage(Message msg) throws InterruptedException {
-                  String json = new String(msg.getData(), StandardCharsets.UTF_8);
-                  Event e = gson.fromJson(json, Event.class);
-                  messageProcessor.accept(e);
-                  msg.ack();
-                  logger.atFine().log("NATS consumer - consumed and acked event '%s'", e);
-                }
-              },
-              false);
+            @Override
+            public void onMessage(Message msg) throws InterruptedException {
+              String json = new String(msg.getData(), StandardCharsets.UTF_8);
+              Event e = gson.fromJson(json, Event.class);
+              messageProcessor.accept(e);
+              msg.ack();
+              logger.atFine().log("NATS consumer - consumed and acked event '%s'", e);
+            }
+          },
+          false);
     } catch (IOException | JetStreamApiException e) {
       logger.atSevere().withCause(e).log(
           "NATS consumer - subscribing to subject [%s] failed", subject);
+      return null;
     }
   }
 
   public void shutdown() {
     try {
-      int shutdownTimeoutMs = config.getShutdownTimeoutMs();
-      Duration shutdownTimeout = Duration.ofMillis(shutdownTimeoutMs);
-      logger.atInfo().log(
-          "NATS consumer - Waiting up to '%s' milliseconds to complete shutdown of consumer of stream '%s'",
-          shutdownTimeoutMs, getStreamName());
+      subscription.unsubscribe();
       connection.closeDispatcher(dispatcher);
-      CompletableFuture<Boolean> f = connection.drain(shutdownTimeout);
-      connection.flush(shutdownTimeout);
-      f.get(shutdownTimeoutMs, TimeUnit.MILLISECONDS);
-      connection.close();
+      logger.at(Level.INFO).log(
+          "NATS consumer - consumer for subject '%s' was shutdown", subjectName);
     } catch (Exception e) {
       logger.atSevere().withCause(e).log(
-          "NATS consumer - error caught when shutting down consumer for stream %s",
+          "NATS consumer - error caught when shutting down consumer for subject %s",
           getStreamName());
     }
-    logger.atInfo().log(
-        "NATS consumer - shutdown consumer of stream %s completed.", getStreamName());
+    logger.atInfo().log("NATS consumer - shutdown consumer of subject %s completed.", subjectName);
   }
 
   public java.util.function.Consumer<Event> getMessageProcessor() {
@@ -157,7 +150,8 @@
   }
 
   public String getStreamName() {
-    return streamName;
+    // we map BrokerApi streams to NATS subjects
+    return subjectName;
   }
 
   public void resetOffset() {