fix: Fix errors caused by client library upgrade

googleapis · Jul 30, 2021 · 874f811 · 874f811
1 parent 56ce037
commit 874f811
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 5 deletions.
diff --git a/src/main/java/com/google/cloud/pubsublite/flink/PubsubLiteSourceSettings.java b/src/main/java/com/google/cloud/pubsublite/flink/PubsubLiteSourceSettings.java
@@ -104,7 +104,7 @@ private static SubscriberServiceClient newSubscriberServiceClient(
               RoutingMetadata.of(path, partition),
               settingsBuilder);
       return SubscriberServiceClient.create(
-          addDefaultSettings(path.location().region(), settingsBuilder));
+          addDefaultSettings(path.location().extractRegion(), settingsBuilder));
     } catch (Throwable t) {
       throw toCanonical(t).underlying;
     }
@@ -127,7 +127,9 @@ AdminClient getAdminClient() {
       return adminClientSupplier().get();
     }
     return AdminClient.create(
-        AdminClientSettings.newBuilder().setRegion(subscriptionPath().location().region()).build());
+        AdminClientSettings.newBuilder()
+            .setRegion(subscriptionPath().location().extractRegion())
+            .build());
   }
 
   CursorClient getCursorClient() {
@@ -136,7 +138,7 @@ CursorClient getCursorClient() {
     }
     return CursorClient.create(
         CursorClientSettings.newBuilder()
-            .setRegion(subscriptionPath().location().region())
+            .setRegion(subscriptionPath().location().extractRegion())
             .build());
   }
 

diff --git a/src/main/java/com/google/cloud/pubsublite/flink/internal/sink/PerServerPublisherCache.java b/src/main/java/com/google/cloud/pubsublite/flink/internal/sink/PerServerPublisherCache.java
@@ -48,7 +48,7 @@ private static PublisherServiceClient newServiceClient(TopicPath topic, Partitio
             settingsBuilder);
     try {
       return PublisherServiceClient.create(
-          addDefaultSettings(topic.location().region(), settingsBuilder));
+          addDefaultSettings(topic.location().extractRegion(), settingsBuilder));
     } catch (Throwable t) {
       throw toCanonical(t).underlying;
     }
@@ -69,7 +69,7 @@ private static Publisher<MessageMetadata> newPublisher(PublisherOptions options)
                     .setServiceClient(newServiceClient(options.topicPath(), partition))
                     .setBatchingSettings(PublisherSettings.DEFAULT_BATCHING_SETTINGS)
                     .build())
-        .setAdminClient(getAdminClient(options.topicPath().location().region()))
+        .setAdminClient(getAdminClient(options.topicPath().location().extractRegion()))
         .build()
         .instantiate();
   }

diff --git a/src/main/java/com/google/cloud/pubsublite/flink/internal/sink/PublisherCache.java b/src/main/java/com/google/cloud/pubsublite/flink/internal/sink/PublisherCache.java
@@ -67,6 +67,14 @@ public void failed(State s, Throwable t) {
   @VisibleForTesting
   public synchronized void set(T options, Publisher<MessageMetadata> toCache) {
     livePublishers.put(options, toCache);
+    toCache.addListener(
+        new Listener() {
+          @Override
+          public void failed(State s, Throwable t) {
+            evict(options);
+          }
+        },
+        SystemExecutors.getAlarmExecutor());
   }
 
   @Override