minor adjustments

integration test update from rebase yolo switch to per-test topic creation rename the test for magic remove line fix after being broken by client lib more region -> extractRegion consistent flink version one more global state is bad formatting remove line
googleapis · Jul 30, 2021 · 6c930f1 · 6c930f1
1 parent 56ce037
commit 6c930f1
Show file tree

Hide file tree

Showing 5 changed files with 406 additions and 5 deletions.
diff --git a/pom.xml b/pom.xml
@@ -142,5 +142,18 @@
             <version>${flink.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-test-utils_2.11</artifactId>
+            <version>${flink.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-runtime_2.11</artifactId>
+            <version>${flink.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>
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