diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java
index 26411942f..4c5bb4ff9 100644
--- a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java
+++ b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java
@@ -61,6 +61,7 @@
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -97,6 +98,7 @@ final class StreamingSubscriberConnection extends AbstractApiService implements
   private final String subscription;
   private final SubscriptionName subscriptionNameObject;
   private final ScheduledExecutorService systemExecutor;
+  private final ExecutorService eodAckCallbackExecutor;
   private final MessageDispatcher messageDispatcher;
 
   private final FlowControlSettings flowControlSettings;
@@ -128,6 +130,7 @@ private StreamingSubscriberConnection(Builder builder) {
     subscription = builder.subscription;
     subscriptionNameObject = SubscriptionName.parse(builder.subscription);
     systemExecutor = builder.systemExecutor;
+    eodAckCallbackExecutor = builder.eodAckCallbackExecutor;
 
     // We need to set the default stream ack deadline on the initial request, this will be
     // updated by modack requests in the message dispatcher
@@ -455,7 +458,14 @@ private void sendAckOperations(
                       .setSubscription(subscription)
                       .addAllAckIds(ackIdsInRequest)
                       .build());
-      ApiFutures.addCallback(ackFuture, callback, directExecutor());
+      if (getExactlyOnceDeliveryEnabled()) {
+        // When exatly-once delivery is enabled, a lock is acquired on this callback which can cause
+        // deadlock when using MoreExecutors.directExecutor(). We use a separate executor to avoid
+        // this.
+        ApiFutures.addCallback(ackFuture, callback, eodAckCallbackExecutor);
+      } else {
+        ApiFutures.addCallback(ackFuture, callback, directExecutor());
+      }
       pendingOperations++;
     }
     ackOperationsWaiter.incrementPendingCount(pendingOperations);
@@ -504,7 +514,14 @@ private void sendModackOperations(
                         .addAllAckIds(ackIdsInRequest)
                         .setAckDeadlineSeconds(modackRequestData.getDeadlineExtensionSeconds())
                         .build());
-        ApiFutures.addCallback(modackFuture, callback, directExecutor());
+        if (getExactlyOnceDeliveryEnabled()) {
+          // When exatly-once delivery is enabled, a lock is acquired on this callback which can
+          // cause deadlock when using MoreExecutors.directExecutor(). We use a separate executor to
+          // avoid this.
+          ApiFutures.addCallback(modackFuture, callback, eodAckCallbackExecutor);
+        } else {
+          ApiFutures.addCallback(modackFuture, callback, directExecutor());
+        }
         pendingOperations++;
       }
     }
@@ -736,6 +753,7 @@ public static final class Builder {
     private boolean useLegacyFlowControl;
     private ScheduledExecutorService executor;
     private ScheduledExecutorService systemExecutor;
+    private ExecutorService eodAckCallbackExecutor;
     private ApiClock clock;
 
     private boolean enableOpenTelemetryTracing;
@@ -826,6 +844,11 @@ public Builder setSystemExecutor(ScheduledExecutorService systemExecutor) {
       return this;
     }
 
+    public Builder setEodAckCallbackExecutor(ExecutorService eodAckCallbackExecutor) {
+      this.eodAckCallbackExecutor = eodAckCallbackExecutor;
+      return this;
+    }
+
     public Builder setClock(ApiClock clock) {
       this.clock = clock;
       return this;
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Subscriber.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Subscriber.java
index 36ec5dc36..27e9a5067 100644
--- a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Subscriber.java
+++ b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Subscriber.java
@@ -51,6 +51,7 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
@@ -150,6 +151,7 @@ public class Subscriber extends AbstractApiService implements SubscriberInterfac
   // An instantiation of the SystemExecutorProvider used for processing acks
   // and other system actions.
   @Nullable private final ScheduledExecutorService alarmsExecutor;
+  private final ExecutorService eodAckCallbackExecutor;
   private final Distribution ackLatencyDistribution =
       new Distribution(Math.toIntExact(MAX_STREAM_ACK_DEADLINE.getSeconds()) + 1);
 
@@ -200,6 +202,13 @@ private Subscriber(Builder builder) {
       backgroundResources.add(new ExecutorAsBackgroundResource((alarmsExecutor)));
     }
 
+    // When exactly-once delivery is enabled, we use a different executor from the system executor
+    // for handling ack and modack responses to prevent deadlock.
+    ThreadFactory eodAckCallbackThreadFactory = new ThreadFactoryBuilder().setDaemon(true).build();
+    int eodAckCallbackThreadCount = Math.max(6, 2 * numPullers);
+    eodAckCallbackExecutor =
+        Executors.newFixedThreadPool(eodAckCallbackThreadCount, eodAckCallbackThreadFactory);
+
     TransportChannelProvider channelProvider = builder.channelProvider;
     if (channelProvider.acceptsPoolSize()) {
       channelProvider = channelProvider.withPoolSize(numPullers);
@@ -416,6 +425,7 @@ private void startStreamingConnections() {
                 .setUseLegacyFlowControl(useLegacyFlowControl)
                 .setExecutor(executor)
                 .setSystemExecutor(alarmsExecutor)
+                .setEodAckCallbackExecutor(eodAckCallbackExecutor)
                 .setClock(clock)
                 .setEnableOpenTelemetryTracing(enableOpenTelemetryTracing)
                 .setTracer(tracer)
diff --git a/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnectionTest.java b/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnectionTest.java
index 335ccbdc3..559b72822 100644
--- a/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnectionTest.java
+++ b/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnectionTest.java
@@ -587,6 +587,7 @@ private StreamingSubscriberConnection getStreamingSubscriberConnectionFromBuilde
         .setFlowController(mock(FlowController.class))
         .setExecutor(executor)
         .setSystemExecutor(systemExecutor)
+        .setEodAckCallbackExecutor(systemExecutor)
         .setClock(clock)
         .setMinDurationPerAckExtension(Subscriber.DEFAULT_MIN_ACK_DEADLINE_EXTENSION)
         .setMinDurationPerAckExtensionDefaultUsed(true)