fix: Use the system executor instead of a separate thread pool for EOD ack/modack callbacks (#2526)

* fix: Use the system executor instead of a separate thread pool for EOD ack/modack callbacks

* fix: Add obsolete method to clirr ignored file

* fix: Update clirr ignore rule

Author: michaelpri10

google-cloud-pubsub/clirr-ignored-differences.xml

@@ -15,4 +15,10 @@
         <method>*(org.threeten.bp.Duration)</method>
         <to>*(java.time.Duration)</to>
     </difference>
+    <difference>
+        <differenceType>7002</differenceType>
+        <!--Ignore changes in this class because it's package private-->
+        <className>com/google/cloud/pubsub/v1/StreamingSubscriberConnection$Builder</className>
+        <method>* setEodAckCallbackExecutor(*)</method>
+    </difference>
 </differences>

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java

@@ -62,7 +62,6 @@
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executor;
-import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -99,7 +98,6 @@ 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;
@@ -131,7 +129,6 @@ 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
@@ -720,11 +717,14 @@ public void run() {
     };
   }
 
+  // If exactly-once is enabled, we hold a lock for the ack/modack response callback, so we want to
+  // avoid using the directExecutor() which runs the callback on the invoking thread. Instead, we
+  // want to schedule the callback to be run on a different thread.
   private Executor getCallbackExecutor() {
     if (!getExactlyOnceDeliveryEnabled()) {
       return directExecutor();
     }
-    return eodAckCallbackExecutor;
+    return systemExecutor;
   }
 
   /** Builder of {@link StreamingSubscriberConnection StreamingSubscriberConnections}. */
@@ -747,7 +747,6 @@ public static final class Builder {
     private boolean useLegacyFlowControl;
     private ScheduledExecutorService executor;
     private ScheduledExecutorService systemExecutor;
-    private ExecutorService eodAckCallbackExecutor;
     private ApiClock clock;
 
     private boolean enableOpenTelemetryTracing;
@@ -838,11 +837,6 @@ 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;

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Subscriber.java

@@ -51,7 +51,6 @@
 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;
@@ -151,8 +150,6 @@ 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;
-  // An executor used for handling ack and modack callbacks when exactly-once delivery is enabled.
-  private final ExecutorService eodAckCallbackExecutor;
 
   private final Distribution ackLatencyDistribution =
       new Distribution(Math.toIntExact(MAX_STREAM_ACK_DEADLINE.getSeconds()) + 1);
@@ -204,15 +201,6 @@ private Subscriber(Builder builder) {
       backgroundResources.add(new ExecutorAsBackgroundResource((alarmsExecutor)));
     }
 
-    // A cached thread pool will create new threads as needed but can reuse previously constructed
-    // threads when available, which helps to improve performance.
-    ThreadFactory eodAckCallbackThreadFactory =
-        new ThreadFactoryBuilder()
-            .setDaemon(true)
-            .setNameFormat("Subscriber-EOD-CallbackExecutor-%d")
-            .build();
-    eodAckCallbackExecutor = Executors.newCachedThreadPool(eodAckCallbackThreadFactory);
-
     TransportChannelProvider channelProvider = builder.channelProvider;
     if (channelProvider.acceptsPoolSize()) {
       channelProvider = channelProvider.withPoolSize(numPullers);
@@ -429,7 +417,6 @@ private void startStreamingConnections() {
                 .setUseLegacyFlowControl(useLegacyFlowControl)
                 .setExecutor(executor)
                 .setSystemExecutor(alarmsExecutor)
-                .setEodAckCallbackExecutor(eodAckCallbackExecutor)
                 .setClock(clock)
                 .setEnableOpenTelemetryTracing(enableOpenTelemetryTracing)
                 .setTracer(tracer)

google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnectionTest.java

@@ -587,7 +587,6 @@ private StreamingSubscriberConnection getStreamingSubscriberConnectionFromBuilde
         .setFlowController(mock(FlowController.class))
         .setExecutor(executor)
         .setSystemExecutor(systemExecutor)
-        .setEodAckCallbackExecutor(systemExecutor)
         .setClock(clock)
         .setMinDurationPerAckExtension(Subscriber.DEFAULT_MIN_ACK_DEADLINE_EXTENSION)
         .setMinDurationPerAckExtensionDefaultUsed(true)