deps: update dependency com.google.cloud:google-cloud-pubsublite-pare…

…nt to v0.16.1 (#215)

* update

* update

* version bump

Co-authored-by: Tianzi Cai <tianzi@google.com>

pom.xml

@@ -3,7 +3,7 @@
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-pubsublite-parent</artifactId>
-    <version>0.16.0</version>
+    <version>0.17.0</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
   <groupId>com.google.cloud</groupId>

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

@@ -136,6 +136,7 @@ MultiPartitionCommitter newMultiPartitionCommitter(long topicPartitionCount) {
                 .instantiate());
   }
 
+  @SuppressWarnings("CheckReturnValue")
   PartitionSubscriberFactory getSubscriberFactory() {
     return (partition, offset, consumer) -> {
       PubsubContext context = PubsubContext.of(Constants.FRAMEWORK);

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

@@ -117,6 +117,7 @@ private static <T> void extractVal(
     }
   }
 
+  @SuppressWarnings("CheckReturnValue")
   public static Message toPubSubMessage(StructType inputSchema, InternalRow row) {
     Message.Builder builder = Message.builder();
     extractVal(

src/main/java/com/google/cloud/pubsublite/spark/internal/CachedPublishers.java

@@ -34,7 +34,7 @@ public class CachedPublishers {
   private final Executor listenerExecutor = Executors.newSingleThreadExecutor();
 
   @GuardedBy("this")
-  private static final Map<PslWriteDataSourceOptions, Publisher<MessageMetadata>> publishers =
+  private final Map<PslWriteDataSourceOptions, Publisher<MessageMetadata>> publishers =
       new HashMap<>();
 
   public synchronized Publisher<MessageMetadata> getOrCreate(

src/main/java/com/google/cloud/pubsublite/spark/internal/MultiPartitionCommitterImpl.java

@@ -19,6 +19,7 @@
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutureCallback;
 import com.google.api.core.ApiFutures;
+import com.google.cloud.pubsublite.Offset;
 import com.google.cloud.pubsublite.Partition;
 import com.google.cloud.pubsublite.internal.wire.Committer;
 import com.google.cloud.pubsublite.spark.PslSourceOffset;
@@ -115,30 +116,29 @@ private synchronized void cleanUpCommitterMap() {
   @Override
   public synchronized void commit(PslSourceOffset offset) {
     updateCommitterMap(offset);
-    offset
-        .partitionOffsetMap()
-        .forEach(
-            (p, o) -> {
-              // Note we don't need to worry about commit offset disorder here since Committer
-              // guarantees the ordering. Once commitOffset() returns, it's either already
-              // sent to stream, or waiting for next connection to open to be sent in order.
-              ApiFuture<Void> future = committerMap.get(p).commitOffset(o);
-              ApiFutures.addCallback(
-                  future,
-                  new ApiFutureCallback<Void>() {
-                    @Override
-                    public void onFailure(Throwable t) {
-                      if (!future.isCancelled()) {
-                        log.atWarning().log("Failed to commit %s,%s.", p.value(), o.value(), t);
-                      }
-                    }
+    for (Map.Entry<Partition, Offset> entry : offset.partitionOffsetMap().entrySet()) {
+      // Note we don't need to worry about commit offset disorder here since Committer
+      // guarantees the ordering. Once commitOffset() returns, it's either already
+      // sent to stream, or waiting for next connection to open to be sent in order.
+      ApiFuture<Void> future = committerMap.get(entry.getKey()).commitOffset(entry.getValue());
+      ApiFutures.addCallback(
+          future,
+          new ApiFutureCallback<Void>() {
+            @Override
+            public void onFailure(Throwable t) {
+              if (!future.isCancelled()) {
+                log.atWarning().withCause(t).log(
+                    "Failed to commit %s,%s.", entry.getKey().value(), entry.getValue().value());
+              }
+            }
 
-                    @Override
-                    public void onSuccess(Void result) {
-                      log.atInfo().log("Committed %s,%s.", p.value(), o.value());
-                    }
-                  },
-                  MoreExecutors.directExecutor());
-            });
+            @Override
+            public void onSuccess(Void result) {
+              log.atInfo().log(
+                  "Committed %s,%s.", entry.getKey().value(), entry.getValue().value());
+            }
+          },
+          MoreExecutors.directExecutor());
+    }
   }
 }