fix: re-use the same subscriber service client for all BlockingPullSu…

…bscribers created (#513)

* fix: respect max messages per batch option

* fix formatting

* fix: use single subscriber service client

* fix public method

* fix: mark the service client transient

* fix: guard by this

src/main/java/com/google/cloud/pubsublite/spark/PslReadDataSourceOptions.java

@@ -58,11 +58,15 @@
 import java.util.Map;
 import java.util.Optional;
 import javax.annotation.Nullable;
+import javax.annotation.concurrent.GuardedBy;
 
 @AutoValue
 public abstract class PslReadDataSourceOptions implements Serializable {
   private static final long serialVersionUID = 2680059304693561607L;
 
+  @GuardedBy("this")
+  private transient SubscriberServiceClient subscriberServiceClient = null;
+
   @Nullable
   public abstract String credentialsKey();
 
@@ -127,36 +131,41 @@ MultiPartitionCommitter newMultiPartitionCommitter() {
 
   @SuppressWarnings("CheckReturnValue")
   PartitionSubscriberFactory getSubscriberFactory() {
+    SubscriberServiceClient serviceClient = getSubscriberServiceClient();
     return (partition, offset, consumer) -> {
+      return SubscriberBuilder.newBuilder()
+          .setSubscriptionPath(this.subscriptionPath())
+          .setPartition(partition)
+          .setMessageConsumer(consumer)
+          .setStreamFactory(
+              responseStream -> {
+                ApiCallContext context =
+                    getCallContext(
+                        PubsubContext.of(Constants.FRAMEWORK),
+                        RoutingMetadata.of(subscriptionPath(), partition));
+                return serviceClient.subscribeCallable().splitCall(responseStream, context);
+              })
+          .setInitialLocation(
+              SeekRequest.newBuilder()
+                  .setCursor(Cursor.newBuilder().setOffset(offset.value()))
+                  .build())
+          .build();
+    };
+  }
+
+  private synchronized SubscriberServiceClient getSubscriberServiceClient() {
+    if (subscriberServiceClient != null) return subscriberServiceClient;
+    try {
       SubscriberServiceSettings.Builder settingsBuilder =
           SubscriberServiceSettings.newBuilder()
               .setCredentialsProvider(new PslCredentialsProvider(credentialsKey()));
-      try {
-        SubscriberServiceClient serviceClient =
-            SubscriberServiceClient.create(
-                addDefaultSettings(
-                    this.subscriptionPath().location().extractRegion(), settingsBuilder));
-        return SubscriberBuilder.newBuilder()
-            .setSubscriptionPath(this.subscriptionPath())
-            .setPartition(partition)
-            .setMessageConsumer(consumer)
-            .setStreamFactory(
-                responseStream -> {
-                  ApiCallContext context =
-                      getCallContext(
-                          PubsubContext.of(Constants.FRAMEWORK),
-                          RoutingMetadata.of(subscriptionPath(), partition));
-                  return serviceClient.subscribeCallable().splitCall(responseStream, context);
-                })
-            .setInitialLocation(
-                SeekRequest.newBuilder()
-                    .setCursor(Cursor.newBuilder().setOffset(offset.value()))
-                    .build())
-            .build();
-      } catch (IOException e) {
-        throw new IllegalStateException("Failed to create subscriber service.", e);
-      }
-    };
+      subscriberServiceClient =
+          SubscriberServiceClient.create(
+              addDefaultSettings(subscriptionPath().location().extractRegion(), settingsBuilder));
+      return subscriberServiceClient;
+    } catch (IOException e) {
+      throw new IllegalStateException("Unable to create SubscriberServiceClient.");
+    }
   }
 
   // TODO(b/jiangmichael): Make XXXClientSettings accept creds so we could simplify below methods.