fix: Add admin client in producer settings (#82)

googleapis · Mar 1, 2021 · c1cf1d1 · c1cf1d1
1 parent c6efa5e
commit c1cf1d1
Showing 1 changed file with 7 additions and 6 deletions.
diff --git a/src/main/java/com/google/cloud/pubsublite/kafka/ProducerSettings.java b/src/main/java/com/google/cloud/pubsublite/kafka/ProducerSettings.java
@@ -48,10 +48,16 @@ public abstract static class Builder {
     public abstract ProducerSettings build();
   }
 
+  private AdminClient newAdminClient() {
+    return AdminClient.create(
+        AdminClientSettings.newBuilder().setRegion(topicPath().location().region()).build());
+  }
+
   public Producer<byte[], byte[]> instantiate() throws ApiException {
     PartitionCountWatchingPublisherSettings publisherSettings =
         PartitionCountWatchingPublisherSettings.newBuilder()
             .setTopic(topicPath())
+            .setAdminClient(newAdminClient())
             .setPublisherFactory(
                 partition -> {
                   try {
@@ -72,12 +78,7 @@ public Producer<byte[], byte[]> instantiate() throws ApiException {
                   }
                 })
             .build();
-    SharedBehavior shared =
-        new SharedBehavior(
-            AdminClient.create(
-                AdminClientSettings.newBuilder()
-                    .setRegion(topicPath().location().region())
-                    .build()));
+    SharedBehavior shared = new SharedBehavior(newAdminClient());
     return new PubsubLiteProducer(publisherSettings.instantiate(), shared, topicPath());
   }
 }