From 21c181102adaa5b4a39adcd732132f27c806bf95 Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Thu, 19 Jun 2025 21:38:54 -0400 Subject: [PATCH 01/38] Deps: Update to apache spark-sql 3.5.5 --- samples/snapshot/pom.xml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/samples/snapshot/pom.xml b/samples/snapshot/pom.xml index 24a36384..9b682fa8 100644 --- a/samples/snapshot/pom.xml +++ b/samples/snapshot/pom.xml @@ -38,8 +38,8 @@ org.apache.spark - spark-sql_2.11 - 2.4.8 + spark-sql_2.13 + 3.5.5 com.google.cloud From 32fe266898eea58fade061386854f4b7c44b19b8 Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Thu, 19 Jun 2025 21:42:13 -0400 Subject: [PATCH 02/38] Update scala version to 2.13.8 --- pom.xml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pom.xml b/pom.xml index c8ff3031..2e989ab1 100644 --- a/pom.xml +++ b/pom.xml @@ -16,8 +16,8 @@ 1.8 1.8 UTF-8 - 2.12.15 - 2.12 + 2.13.8 + 2.13 3.1.2 3.2.2 From 2a58f39e50ff99dfc5978e1da175c27f3f1bef8d Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Thu, 19 Jun 2025 21:44:21 -0400 Subject: [PATCH 03/38] Update spark in snippets/pom.xml --- samples/snippets/pom.xml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/samples/snippets/pom.xml b/samples/snippets/pom.xml index 89ab13d0..5e681828 100644 --- a/samples/snippets/pom.xml +++ b/samples/snippets/pom.xml @@ -38,8 +38,8 @@ org.apache.spark - spark-sql_2.11 - 2.4.8 + spark-sql_2.13 + 3.5.5 com.google.cloud From 86f0d1ea8c4f47874067f8a82b28ed111c49c745 Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Thu, 19 Jun 2025 21:50:48 -0400 Subject: [PATCH 04/38] Update pom.xml --- pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index 2e989ab1..63ccb063 100644 --- a/pom.xml +++ b/pom.xml @@ -18,7 +18,7 @@ UTF-8 2.13.8 2.13 - 3.1.2 + 3.5.5 3.2.2 From 179f4d0c84a721cdb62e50657315842243214dc8 Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Thu, 19 Jun 2025 22:03:36 -0400 Subject: [PATCH 05/38] Upgrade to scala 2.11.12 --- pom.xml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pom.xml b/pom.xml index 63ccb063..fa6ca05b 100644 --- a/pom.xml +++ b/pom.xml @@ -16,8 +16,8 @@ 1.8 1.8 UTF-8 - 2.13.8 - 2.13 + 2.11.12 + 2.11 3.5.5 3.2.2 From 3b9e2b7f5cff0bf0f59a92abb9b93129db803a62 Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Thu, 19 Jun 2025 22:04:36 -0400 Subject: [PATCH 06/38] Upgrade scale to 2.12.20 --- pom.xml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pom.xml b/pom.xml index fa6ca05b..b7033457 100644 --- a/pom.xml +++ b/pom.xml @@ -16,8 +16,8 @@ 1.8 1.8 UTF-8 - 2.11.12 - 2.11 + 2.12.20 + 2.12 3.5.5 3.2.2 From a3d060df0623356bec42ef1cdd6c8644278f4afa Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Thu, 19 Jun 2025 22:05:59 -0400 Subject: [PATCH 07/38] Set to spark-sql_3.12 --- samples/snippets/pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/samples/snippets/pom.xml b/samples/snippets/pom.xml index 5e681828..45c3d8df 100644 --- a/samples/snippets/pom.xml +++ b/samples/snippets/pom.xml @@ -38,7 +38,7 @@ org.apache.spark - spark-sql_2.13 + spark-sql_2.12 3.5.5 From 3a7d376848caf59526f98fe0d901e5c21711d746 Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Thu, 19 Jun 2025 22:06:33 -0400 Subject: [PATCH 08/38] spark-sql_2.12 --- samples/snapshot/pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/samples/snapshot/pom.xml b/samples/snapshot/pom.xml index 9b682fa8..50dda3d5 100644 --- a/samples/snapshot/pom.xml +++ b/samples/snapshot/pom.xml @@ -38,7 +38,7 @@ org.apache.spark - spark-sql_2.13 + spark-sql_2.12 3.5.5 From ecfb860b1ce0220fc09f35d4e9f95826a787db98 Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Thu, 19 Jun 2025 22:21:38 -0400 Subject: [PATCH 09/38] Update PslSparkUtils.java with scala 2.13 CollectionConverters --- .../google/cloud/pubsublite/spark/PslSparkUtils.java | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/main/java/com/google/cloud/pubsublite/spark/PslSparkUtils.java b/src/main/java/com/google/cloud/pubsublite/spark/PslSparkUtils.java index 4ec2bfbe..927f56de 100644 --- a/src/main/java/com/google/cloud/pubsublite/spark/PslSparkUtils.java +++ b/src/main/java/com/google/cloud/pubsublite/spark/PslSparkUtils.java @@ -17,7 +17,7 @@ package com.google.cloud.pubsublite.spark; import static com.google.common.base.Preconditions.checkArgument; -import static scala.collection.JavaConverters.asScalaBufferConverter; +import static scala.jdk.javaapi.CollectionConverters; import com.google.cloud.pubsublite.Message; import com.google.cloud.pubsublite.Offset; @@ -73,12 +73,12 @@ public static ArrayBasedMapData convertAttributesToSparkMap( value.stream() .map(v -> ByteArray.concat(v.toByteArray())) .collect(Collectors.toList()); - valueList.add(new GenericArrayData(asScalaBufferConverter(attributeVals).asScala())); + valueList.add(new GenericArrayData(CollectionConverters.asScala(attributeVals).toSeq())); }); return new ArrayBasedMapData( - new GenericArrayData(asScalaBufferConverter(keyList).asScala()), - new GenericArrayData(asScalaBufferConverter(valueList).asScala())); + new GenericArrayData(CollectionConverters.asScala(keyList).toSeq()), + new GenericArrayData(CollectionConverters.asScala(valueList).toSeq())); } public static InternalRow toInternalRow( @@ -96,7 +96,7 @@ public static InternalRow toInternalRow( ? Timestamps.toMicros(msg.message().eventTime().get()) : null, convertAttributesToSparkMap(msg.message().attributes()))); - return InternalRow.apply(asScalaBufferConverter(list).asScala()); + return InternalRow.apply(CollectionConverters.asScala(list).toSeq()); } @SuppressWarnings("unchecked") From 41a359944c7da0d6a385435b2c55e653b31d2200 Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Thu, 19 Jun 2025 22:22:52 -0400 Subject: [PATCH 10/38] Update pom.xml with scala 2.13.16 --- pom.xml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pom.xml b/pom.xml index b7033457..33e9fea9 100644 --- a/pom.xml +++ b/pom.xml @@ -16,8 +16,8 @@ 1.8 1.8 UTF-8 - 2.12.20 - 2.12 + 2.13.16 + 2.13 3.5.5 3.2.2 From 1f85e8f58af80e33c57b88900b5bdabf63e1e543 Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Thu, 19 Jun 2025 22:24:11 -0400 Subject: [PATCH 11/38] Update snippets/pom.xml with 2.13 --- samples/snippets/pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/samples/snippets/pom.xml b/samples/snippets/pom.xml index 45c3d8df..5e681828 100644 --- a/samples/snippets/pom.xml +++ b/samples/snippets/pom.xml @@ -38,7 +38,7 @@ org.apache.spark - spark-sql_2.12 + spark-sql_2.13 3.5.5 From d150e7fe921ec2af1894616edbd1c1ef218715e2 Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Thu, 19 Jun 2025 22:24:46 -0400 Subject: [PATCH 12/38] Update samples/snapshot/pom.xml with 2.12 --- samples/snapshot/pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/samples/snapshot/pom.xml b/samples/snapshot/pom.xml index 50dda3d5..9b682fa8 100644 --- a/samples/snapshot/pom.xml +++ b/samples/snapshot/pom.xml @@ -38,7 +38,7 @@ org.apache.spark - spark-sql_2.12 + spark-sql_2.13 3.5.5 From 030aff190f87f3590dc39f912091660e14595dfd Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Thu, 19 Jun 2025 22:33:08 -0400 Subject: [PATCH 13/38] Update PslSparkUtils.java fix import --- .../java/com/google/cloud/pubsublite/spark/PslSparkUtils.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/java/com/google/cloud/pubsublite/spark/PslSparkUtils.java b/src/main/java/com/google/cloud/pubsublite/spark/PslSparkUtils.java index 927f56de..36452e29 100644 --- a/src/main/java/com/google/cloud/pubsublite/spark/PslSparkUtils.java +++ b/src/main/java/com/google/cloud/pubsublite/spark/PslSparkUtils.java @@ -17,7 +17,6 @@ package com.google.cloud.pubsublite.spark; import static com.google.common.base.Preconditions.checkArgument; -import static scala.jdk.javaapi.CollectionConverters; import com.google.cloud.pubsublite.Message; import com.google.cloud.pubsublite.Offset; @@ -52,6 +51,7 @@ import org.apache.spark.unsafe.types.UTF8String; import scala.Option; import scala.compat.java8.functionConverterImpls.FromJavaBiConsumer; +import scala.jdk.javaapi.CollectionConverters; public class PslSparkUtils { From bc00818d71992f2a02aef41a3ee611029224ee9c Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Thu, 19 Jun 2025 22:45:20 -0400 Subject: [PATCH 14/38] Update PslWrite.java to override useCommitCoordinator() --- src/main/java/com/google/cloud/pubsublite/spark/PslWrite.java | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/main/java/com/google/cloud/pubsublite/spark/PslWrite.java b/src/main/java/com/google/cloud/pubsublite/spark/PslWrite.java index c9868107..81c735c3 100644 --- a/src/main/java/com/google/cloud/pubsublite/spark/PslWrite.java +++ b/src/main/java/com/google/cloud/pubsublite/spark/PslWrite.java @@ -102,4 +102,8 @@ public BatchWrite buildForBatch() { public StreamingWrite buildForStreaming() { return this; } + + @Override +public boolean useCommitCoordinator() { + return false; } From 984d342f0da2b4b418566848f701ead7f0e88011 Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Fri, 20 Jun 2025 11:13:15 -0400 Subject: [PATCH 15/38] Update PslWrite.java --- src/main/java/com/google/cloud/pubsublite/spark/PslWrite.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/main/java/com/google/cloud/pubsublite/spark/PslWrite.java b/src/main/java/com/google/cloud/pubsublite/spark/PslWrite.java index 81c735c3..981f084d 100644 --- a/src/main/java/com/google/cloud/pubsublite/spark/PslWrite.java +++ b/src/main/java/com/google/cloud/pubsublite/spark/PslWrite.java @@ -104,6 +104,7 @@ public StreamingWrite buildForStreaming() { } @Override -public boolean useCommitCoordinator() { + public boolean useCommitCoordinator() { return false; + } } From 7836685f3209635c4dc80780c6ff978a96a1c21d Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Fri, 20 Jun 2025 17:28:24 +0000 Subject: [PATCH 16/38] Fix broken pom.xml --- pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index b81ebcff..985954a3 100644 --- a/pom.xml +++ b/pom.xml @@ -67,7 +67,7 @@ org.eclipse.jetty jetty-server - 11.0.24 + 11.0.14 jakarta.xml.bind From 2702694a766f326c7153eb951b56692c7ac4be83 Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Fri, 20 Jun 2025 17:33:03 +0000 Subject: [PATCH 17/38] Fix scala in PslSparkUtilsTest --- .../google/cloud/pubsublite/spark/PslSparkUtilsTest.java | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/src/test/java/com/google/cloud/pubsublite/spark/PslSparkUtilsTest.java b/src/test/java/com/google/cloud/pubsublite/spark/PslSparkUtilsTest.java index d76b668f..c25c0584 100644 --- a/src/test/java/com/google/cloud/pubsublite/spark/PslSparkUtilsTest.java +++ b/src/test/java/com/google/cloud/pubsublite/spark/PslSparkUtilsTest.java @@ -18,7 +18,6 @@ import static com.google.common.truth.Truth.assertThat; import static org.junit.Assert.assertThrows; -import static scala.collection.JavaConverters.asScalaBufferConverter; import com.google.cloud.pubsublite.Message; import com.google.cloud.pubsublite.Offset; @@ -44,6 +43,8 @@ import org.apache.spark.sql.types.StructType; import org.apache.spark.unsafe.types.ByteArray; import org.junit.Test; +import scala.jdk.javaapi.CollectionConverters; + public class PslSparkUtilsTest { @@ -138,7 +139,7 @@ public void testToPubSubMessage() { PslSparkUtils.convertAttributesToSparkMap(message.attributes()), Timestamps.toMicros(message.eventTime().get()), "abc".getBytes(StandardCharsets.UTF_8))); - InternalRow row = InternalRow.apply(asScalaBufferConverter(list).asScala()); + InternalRow row = InternalRow.apply(CollectionConverters.asScala(list).toSeq()); StructType structType = new StructType( @@ -164,7 +165,7 @@ public void testToPubSubMessageLongForEventTimestamp() { new StructField("event_timestamp", DataTypes.LongType, false, Metadata.empty()) }); List list = Collections.singletonList(100000L); - InternalRow row = InternalRow.apply(asScalaBufferConverter(list).asScala()); + InternalRow row = InternalRow.apply(CollectionConverters.asScala(list).toSeq()); Message message = PslSparkUtils.toPubSubMessage(structType, row); assertThat(message).isEqualTo(expectedMsg); From 9d6c352cbcaedc8e0a565001e3ce870999e9b331 Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Fri, 20 Jun 2025 17:42:38 +0000 Subject: [PATCH 18/38] Add spark-sql-api --- pom.xml | 6 ++++++ samples/snapshot/pom.xml | 5 +++++ samples/snippets/pom.xml | 5 +++++ 3 files changed, 16 insertions(+) diff --git a/pom.xml b/pom.xml index 985954a3..29f20a1a 100644 --- a/pom.xml +++ b/pom.xml @@ -108,6 +108,12 @@ ${spark.version} provided + + org.apache.spark + spark-sql-api_${scala.version.short} + ${spark.version} + provided + org.apache.spark spark-catalyst_${scala.version.short} diff --git a/samples/snapshot/pom.xml b/samples/snapshot/pom.xml index 9b682fa8..53b74d78 100644 --- a/samples/snapshot/pom.xml +++ b/samples/snapshot/pom.xml @@ -41,6 +41,11 @@ spark-sql_2.13 3.5.5 + + org.apache.spark + spark-sql-api_2.13 + 3.5.5 + com.google.cloud google-cloud-pubsublite diff --git a/samples/snippets/pom.xml b/samples/snippets/pom.xml index 5e681828..e2bf7bce 100644 --- a/samples/snippets/pom.xml +++ b/samples/snippets/pom.xml @@ -41,6 +41,11 @@ spark-sql_2.13 3.5.5 + + org.apache.spark + spark-sql-api_2.13 + 3.5.5 + com.google.cloud google-cloud-pubsublite From 58da70c59fba8028119ebc657ffb2c1db6f1aeaa Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Fri, 20 Jun 2025 17:47:08 +0000 Subject: [PATCH 19/38] Fix lint errors --- .../java/com/google/cloud/pubsublite/spark/PslSparkUtils.java | 3 ++- src/main/java/com/google/cloud/pubsublite/spark/PslWrite.java | 2 +- .../com/google/cloud/pubsublite/spark/PslSparkUtilsTest.java | 1 - 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/main/java/com/google/cloud/pubsublite/spark/PslSparkUtils.java b/src/main/java/com/google/cloud/pubsublite/spark/PslSparkUtils.java index 36452e29..3c5dd2d3 100644 --- a/src/main/java/com/google/cloud/pubsublite/spark/PslSparkUtils.java +++ b/src/main/java/com/google/cloud/pubsublite/spark/PslSparkUtils.java @@ -73,7 +73,8 @@ public static ArrayBasedMapData convertAttributesToSparkMap( value.stream() .map(v -> ByteArray.concat(v.toByteArray())) .collect(Collectors.toList()); - valueList.add(new GenericArrayData(CollectionConverters.asScala(attributeVals).toSeq())); + valueList.add( + new GenericArrayData(CollectionConverters.asScala(attributeVals).toSeq())); }); return new ArrayBasedMapData( diff --git a/src/main/java/com/google/cloud/pubsublite/spark/PslWrite.java b/src/main/java/com/google/cloud/pubsublite/spark/PslWrite.java index 981f084d..6de96738 100644 --- a/src/main/java/com/google/cloud/pubsublite/spark/PslWrite.java +++ b/src/main/java/com/google/cloud/pubsublite/spark/PslWrite.java @@ -104,7 +104,7 @@ public StreamingWrite buildForStreaming() { } @Override - public boolean useCommitCoordinator() { + public boolean useCommitCoordinator() { return false; } } diff --git a/src/test/java/com/google/cloud/pubsublite/spark/PslSparkUtilsTest.java b/src/test/java/com/google/cloud/pubsublite/spark/PslSparkUtilsTest.java index c25c0584..39747612 100644 --- a/src/test/java/com/google/cloud/pubsublite/spark/PslSparkUtilsTest.java +++ b/src/test/java/com/google/cloud/pubsublite/spark/PslSparkUtilsTest.java @@ -45,7 +45,6 @@ import org.junit.Test; import scala.jdk.javaapi.CollectionConverters; - public class PslSparkUtilsTest { @Test From f14e5c97e594aa2dc2d54f882ad257490615f623 Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Fri, 20 Jun 2025 18:11:07 +0000 Subject: [PATCH 20/38] Fix commons-lang missing dependency --- samples/snippets/pom.xml | 5 +++++ .../src/test/java/pubsublite/spark/SampleTestBase.java | 2 +- 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/samples/snippets/pom.xml b/samples/snippets/pom.xml index e2bf7bce..13a0b7e7 100644 --- a/samples/snippets/pom.xml +++ b/samples/snippets/pom.xml @@ -85,6 +85,11 @@ 1.1.3 test + + org.apache.commons + commons-lang3 + 3.12 + diff --git a/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java b/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java index a558208c..cf7e742a 100644 --- a/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java +++ b/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java @@ -43,7 +43,7 @@ import java.util.Map; import java.util.Optional; import java.util.UUID; -import org.apache.commons.lang.StringUtils; +import org.apache.commons.lang3.StringUtils; import org.apache.maven.shared.invoker.DefaultInvocationRequest; import org.apache.maven.shared.invoker.DefaultInvoker; import org.apache.maven.shared.invoker.InvocationOutputHandler; From ac0e9fba7792b3c14f7e1d4d09f8dd2a37b002fc Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Fri, 20 Jun 2025 18:45:20 +0000 Subject: [PATCH 21/38] Remove dependency on StringUtils --- samples/snippets/pom.xml | 5 ----- .../src/test/java/pubsublite/spark/SampleTestBase.java | 4 ++-- 2 files changed, 2 insertions(+), 7 deletions(-) diff --git a/samples/snippets/pom.xml b/samples/snippets/pom.xml index 13a0b7e7..e2bf7bce 100644 --- a/samples/snippets/pom.xml +++ b/samples/snippets/pom.xml @@ -85,11 +85,6 @@ 1.1.3 test - - org.apache.commons - commons-lang3 - 3.12 - diff --git a/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java b/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java index cf7e742a..231d575d 100644 --- a/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java +++ b/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java @@ -42,8 +42,8 @@ import java.util.Arrays; import java.util.Map; import java.util.Optional; +import java.util.String; import java.util.UUID; -import org.apache.commons.lang3.StringUtils; import org.apache.maven.shared.invoker.DefaultInvocationRequest; import org.apache.maven.shared.invoker.DefaultInvoker; import org.apache.maven.shared.invoker.InvocationOutputHandler; @@ -110,7 +110,7 @@ protected void findMavenHome() throws Exception { assertThat(p.waitFor()).isEqualTo(0); String s; while ((s = stdOut.readLine()) != null) { - if (StringUtils.startsWith(s, "Maven home: ")) { + if (String.startsWith(s, "Maven home: ")) { mavenHome = s.replace("Maven home: ", ""); } } From 95ca0baf5c33014ce288be9af768344aa3fc3c7b Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Fri, 20 Jun 2025 19:31:10 +0000 Subject: [PATCH 22/38] remove uneeded import --- .../snippets/src/test/java/pubsublite/spark/SampleTestBase.java | 1 - 1 file changed, 1 deletion(-) diff --git a/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java b/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java index 231d575d..0782b40e 100644 --- a/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java +++ b/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java @@ -42,7 +42,6 @@ import java.util.Arrays; import java.util.Map; import java.util.Optional; -import java.util.String; import java.util.UUID; import org.apache.maven.shared.invoker.DefaultInvocationRequest; import org.apache.maven.shared.invoker.DefaultInvoker; From 2e3f6165db3d7fe0a1e5f68648d8703aa75c7312 Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Fri, 20 Jun 2025 21:18:37 +0000 Subject: [PATCH 23/38] Fix String.startsWith --- .../snippets/src/test/java/pubsublite/spark/SampleTestBase.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java b/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java index 0782b40e..d82612c9 100644 --- a/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java +++ b/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java @@ -109,7 +109,7 @@ protected void findMavenHome() throws Exception { assertThat(p.waitFor()).isEqualTo(0); String s; while ((s = stdOut.readLine()) != null) { - if (String.startsWith(s, "Maven home: ")) { + if (s.startsWith("Maven home: ")) { mavenHome = s.replace("Maven home: ", ""); } } From 6f8600efad077533be91f1a7e483edbe13a4ba7c Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Sun, 22 Jun 2025 18:40:49 +0000 Subject: [PATCH 24/38] Set useCommitCoordinator true --- src/main/java/com/google/cloud/pubsublite/spark/PslWrite.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/java/com/google/cloud/pubsublite/spark/PslWrite.java b/src/main/java/com/google/cloud/pubsublite/spark/PslWrite.java index 6de96738..7f368a9b 100644 --- a/src/main/java/com/google/cloud/pubsublite/spark/PslWrite.java +++ b/src/main/java/com/google/cloud/pubsublite/spark/PslWrite.java @@ -105,6 +105,6 @@ public StreamingWrite buildForStreaming() { @Override public boolean useCommitCoordinator() { - return false; + return true; } } From 32baffb99823544338be5d1ed773617deeab916e Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Tue, 24 Jun 2025 18:22:24 +0000 Subject: [PATCH 25/38] Unpin scala --- renovate.json | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/renovate.json b/renovate.json index cc6f931a..c522909d 100644 --- a/renovate.json +++ b/renovate.json @@ -71,12 +71,10 @@ { "packagePatterns": [ "^spark.version", - "^org.apache.spark", - "^scala.version", - "^org.scala-lang" + "^org.apache.spark" ], "enabled": false, - "groupName": "spark and scala pinned dependencies" + "groupName": "spark and pinned dependencies" } ], "semanticCommits": true, From 32e216c44e662af14fbcc332ea6d4cd5581de09c Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Mon, 30 Jun 2025 19:04:22 +0000 Subject: [PATCH 26/38] removing spark pinning --- renovate.json | 8 -------- 1 file changed, 8 deletions(-) diff --git a/renovate.json b/renovate.json index c522909d..768fc249 100644 --- a/renovate.json +++ b/renovate.json @@ -68,14 +68,6 @@ ], "groupName": "jackson dependencies" }, - { - "packagePatterns": [ - "^spark.version", - "^org.apache.spark" - ], - "enabled": false, - "groupName": "spark and pinned dependencies" - } ], "semanticCommits": true, "dependencyDashboard": true From aa180b3babfed6c12e458d566181e86c45d370b2 Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Wed, 2 Jul 2025 16:33:53 +0000 Subject: [PATCH 27/38] Update current version --- pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index 29f20a1a..983fe56d 100644 --- a/pom.xml +++ b/pom.xml @@ -7,7 +7,7 @@ 4.0.0 pubsublite-spark-sql-streaming - 1.0.1-SNAPSHOT + 1.0.2 jar Pub/Sub Lite Spark SQL Streaming https://github.com/googleapis/java-pubsublite-spark From b448677cb45f38625933ef0e011b0294be3cbb43 Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Wed, 2 Jul 2025 20:54:47 +0000 Subject: [PATCH 28/38] Update dataproc image version --- samples/simple_read_sample.sh | 2 +- samples/simple_write_sample.sh | 2 +- samples/word_count_sample.sh | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/samples/simple_read_sample.sh b/samples/simple_read_sample.sh index bf81cadd..b3cb1ca1 100644 --- a/samples/simple_read_sample.sh +++ b/samples/simple_read_sample.sh @@ -25,7 +25,7 @@ if [ "$1" == "run" ]; then gcloud dataproc clusters create $CLUSTER_NAME \ --region=$REGION \ --zone=$REGION-$ZONE_ID \ - --image-version=1.5-debian10 \ + --image-version=2.3-debian12 \ --scopes=cloud-platform # Package sample jar diff --git a/samples/simple_write_sample.sh b/samples/simple_write_sample.sh index 6688b752..47ac0767 100644 --- a/samples/simple_write_sample.sh +++ b/samples/simple_write_sample.sh @@ -21,7 +21,7 @@ if [ "$1" == "run" ]; then gcloud dataproc clusters create $CLUSTER_NAME \ --region=$REGION \ --zone=$REGION-$ZONE_ID \ - --image-version=1.5-debian10 \ + --image-version=2.3-debian12 \ --scopes=cloud-platform # Package sample jar diff --git a/samples/word_count_sample.sh b/samples/word_count_sample.sh index 052eee36..361c25db 100644 --- a/samples/word_count_sample.sh +++ b/samples/word_count_sample.sh @@ -26,7 +26,7 @@ if [ "$1" == "run" ]; then gcloud dataproc clusters create $CLUSTER_NAME \ --region=$REGION \ --zone=$REGION-$ZONE_ID \ - --image-version=1.5-debian10 \ + --image-version=2.3-debian12 \ --scopes=cloud-platform # Package sample jar From b374e97ef5576d67de6528995da50c1307049b7e Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Mon, 7 Jul 2025 15:07:20 +0000 Subject: [PATCH 29/38] Temporarily downgrade to 3.5.3 to test on dataproc --- pom.xml | 2 +- samples/snapshot/pom.xml | 4 ++-- samples/snippets/pom.xml | 4 ++-- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/pom.xml b/pom.xml index 43ec8218..06785373 100644 --- a/pom.xml +++ b/pom.xml @@ -18,7 +18,7 @@ UTF-8 2.13.16 2.13 - 3.5.5 + 3.5.3 3.2.2 diff --git a/samples/snapshot/pom.xml b/samples/snapshot/pom.xml index 198bc7c3..c73a5fe0 100644 --- a/samples/snapshot/pom.xml +++ b/samples/snapshot/pom.xml @@ -39,12 +39,12 @@ org.apache.spark spark-sql_2.13 - 3.5.5 + 3.5.3 org.apache.spark spark-sql-api_2.13 - 3.5.5 + 3.5.3 com.google.cloud diff --git a/samples/snippets/pom.xml b/samples/snippets/pom.xml index 9741543a..ba144c4d 100644 --- a/samples/snippets/pom.xml +++ b/samples/snippets/pom.xml @@ -39,12 +39,12 @@ org.apache.spark spark-sql_2.13 - 3.5.5 + 3.5.3 org.apache.spark spark-sql-api_2.13 - 3.5.5 + 3.5.3 com.google.cloud From bfdbbe282ead30cd7aa455d06a60900396a84cee Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Mon, 7 Jul 2025 16:28:18 +0000 Subject: [PATCH 30/38] Update samples README and versions.txt --- samples/README.md | 2 +- versions.txt | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/samples/README.md b/samples/README.md index ea1ae264..fb8d1713 100644 --- a/samples/README.md +++ b/samples/README.md @@ -56,7 +56,7 @@ follow the steps: ``` 4. Create a Dataproc cluster ```sh - gcloud dataproc clusters create $CLUSTER_NAME --region=$REGION --zone=$REGION-$ZONE_ID --image-version=1.5-debian10 --scopes=cloud-platform + gcloud dataproc clusters create $CLUSTER_NAME --region=$REGION --zone=$REGION-$ZONE_ID --image-version=2.3-debian12 --scopes=cloud-platform ``` 5. Package sample jar ```sh diff --git a/versions.txt b/versions.txt index 01082cbc..54c784b7 100644 --- a/versions.txt +++ b/versions.txt @@ -1,4 +1,4 @@ # Format: # module:released-version:current-version -pubsublite-spark-sql-streaming:1.0.0:1.0.1-SNAPSHOT +pubsublite-spark-sql-streaming:1.0.2:1.0.2-SNAPSHOT From f0a832e5a4a5c678633a9f609f25866c6a87b6a4 Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Mon, 7 Jul 2025 21:21:32 +0000 Subject: [PATCH 31/38] Revert pom changes in samples --- renovate.json | 10 ++++++++++ samples/snapshot/pom.xml | 9 ++------- samples/snippets/pom.xml | 9 ++------- 3 files changed, 14 insertions(+), 14 deletions(-) diff --git a/renovate.json b/renovate.json index 768fc249..cc6f931a 100644 --- a/renovate.json +++ b/renovate.json @@ -68,6 +68,16 @@ ], "groupName": "jackson dependencies" }, + { + "packagePatterns": [ + "^spark.version", + "^org.apache.spark", + "^scala.version", + "^org.scala-lang" + ], + "enabled": false, + "groupName": "spark and scala pinned dependencies" + } ], "semanticCommits": true, "dependencyDashboard": true diff --git a/samples/snapshot/pom.xml b/samples/snapshot/pom.xml index c73a5fe0..51d18b0a 100644 --- a/samples/snapshot/pom.xml +++ b/samples/snapshot/pom.xml @@ -38,13 +38,8 @@ org.apache.spark - spark-sql_2.13 - 3.5.3 - - - org.apache.spark - spark-sql-api_2.13 - 3.5.3 + spark-sql_2.11 + 2.4.8 com.google.cloud diff --git a/samples/snippets/pom.xml b/samples/snippets/pom.xml index ba144c4d..d34a56cb 100644 --- a/samples/snippets/pom.xml +++ b/samples/snippets/pom.xml @@ -38,13 +38,8 @@ org.apache.spark - spark-sql_2.13 - 3.5.3 - - - org.apache.spark - spark-sql-api_2.13 - 3.5.3 + spark-sql_2.11 + 2.4.8 com.google.cloud From fd08237ccd06db7b6d6c177c039686066603bc98 Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Mon, 7 Jul 2025 21:23:02 +0000 Subject: [PATCH 32/38] Revert "Remove dependency on StringUtils" This reverts commit ac0e9fba7792b3c14f7e1d4d09f8dd2a37b002fc. --- samples/snippets/pom.xml | 5 +++++ .../src/test/java/pubsublite/spark/SampleTestBase.java | 3 ++- 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/samples/snippets/pom.xml b/samples/snippets/pom.xml index d34a56cb..029681ac 100644 --- a/samples/snippets/pom.xml +++ b/samples/snippets/pom.xml @@ -80,6 +80,11 @@ 1.1.3 test + + org.apache.commons + commons-lang3 + 3.12 + diff --git a/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java b/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java index d82612c9..cf7e742a 100644 --- a/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java +++ b/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java @@ -43,6 +43,7 @@ import java.util.Map; import java.util.Optional; import java.util.UUID; +import org.apache.commons.lang3.StringUtils; import org.apache.maven.shared.invoker.DefaultInvocationRequest; import org.apache.maven.shared.invoker.DefaultInvoker; import org.apache.maven.shared.invoker.InvocationOutputHandler; @@ -109,7 +110,7 @@ protected void findMavenHome() throws Exception { assertThat(p.waitFor()).isEqualTo(0); String s; while ((s = stdOut.readLine()) != null) { - if (s.startsWith("Maven home: ")) { + if (StringUtils.startsWith(s, "Maven home: ")) { mavenHome = s.replace("Maven home: ", ""); } } From 47f5e62928256dffe3f867ce906a602be20b3230 Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Mon, 7 Jul 2025 21:27:30 +0000 Subject: [PATCH 33/38] Separate versioning from PR --- pom.xml | 2 +- versions.txt | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pom.xml b/pom.xml index 06785373..09f5bf75 100644 --- a/pom.xml +++ b/pom.xml @@ -7,7 +7,7 @@ 4.0.0 pubsublite-spark-sql-streaming - 1.0.2 + 1.0.1-SNAPSHOT jar Pub/Sub Lite Spark SQL Streaming https://github.com/googleapis/java-pubsublite-spark diff --git a/versions.txt b/versions.txt index 54c784b7..9c13ef78 100644 --- a/versions.txt +++ b/versions.txt @@ -1,4 +1,4 @@ # Format: # module:released-version:current-version -pubsublite-spark-sql-streaming:1.0.2:1.0.2-SNAPSHOT +pubsublite-spark-sql-streaming:1.0.1:1.0.1-SNAPSHOT From 6a03a7163e7e77c587b623ca462632b13ae530be Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Mon, 7 Jul 2025 22:24:39 +0000 Subject: [PATCH 34/38] Revert string utils update --- samples/snippets/pom.xml | 5 ----- .../src/test/java/pubsublite/spark/SampleTestBase.java | 2 +- 2 files changed, 1 insertion(+), 6 deletions(-) diff --git a/samples/snippets/pom.xml b/samples/snippets/pom.xml index 029681ac..d34a56cb 100644 --- a/samples/snippets/pom.xml +++ b/samples/snippets/pom.xml @@ -80,11 +80,6 @@ 1.1.3 test - - org.apache.commons - commons-lang3 - 3.12 - diff --git a/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java b/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java index cf7e742a..a558208c 100644 --- a/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java +++ b/samples/snippets/src/test/java/pubsublite/spark/SampleTestBase.java @@ -43,7 +43,7 @@ import java.util.Map; import java.util.Optional; import java.util.UUID; -import org.apache.commons.lang3.StringUtils; +import org.apache.commons.lang.StringUtils; import org.apache.maven.shared.invoker.DefaultInvocationRequest; import org.apache.maven.shared.invoker.DefaultInvoker; import org.apache.maven.shared.invoker.InvocationOutputHandler; From 36138fc4bdf501732a890a8fe7cd6e27f71d54e6 Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Mon, 7 Jul 2025 23:26:59 +0000 Subject: [PATCH 35/38] Revert scala version to 2.12.20 --- pom.xml | 4 ++-- versions.txt | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/pom.xml b/pom.xml index 09f5bf75..39286a1f 100644 --- a/pom.xml +++ b/pom.xml @@ -16,8 +16,8 @@ 1.8 1.8 UTF-8 - 2.13.16 - 2.13 + 2.12.20 + 2.12 3.5.3 3.2.2 diff --git a/versions.txt b/versions.txt index 9c13ef78..01082cbc 100644 --- a/versions.txt +++ b/versions.txt @@ -1,4 +1,4 @@ # Format: # module:released-version:current-version -pubsublite-spark-sql-streaming:1.0.1:1.0.1-SNAPSHOT +pubsublite-spark-sql-streaming:1.0.0:1.0.1-SNAPSHOT From 200c5b7ec8d231352cfff90659083911ab473d9d Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Tue, 8 Jul 2025 18:17:13 +0000 Subject: [PATCH 36/38] Revert CollectionConverter changes --- pom.xml | 6 ------ .../google/cloud/pubsublite/spark/PslSparkUtils.java | 11 +++++------ .../cloud/pubsublite/spark/PslSparkUtilsTest.java | 6 +++--- 3 files changed, 8 insertions(+), 15 deletions(-) diff --git a/pom.xml b/pom.xml index 39286a1f..e640840a 100644 --- a/pom.xml +++ b/pom.xml @@ -108,12 +108,6 @@ ${spark.version} provided - - org.apache.spark - spark-sql-api_${scala.version.short} - ${spark.version} - provided - org.apache.spark spark-catalyst_${scala.version.short} diff --git a/src/main/java/com/google/cloud/pubsublite/spark/PslSparkUtils.java b/src/main/java/com/google/cloud/pubsublite/spark/PslSparkUtils.java index 3c5dd2d3..4ec2bfbe 100644 --- a/src/main/java/com/google/cloud/pubsublite/spark/PslSparkUtils.java +++ b/src/main/java/com/google/cloud/pubsublite/spark/PslSparkUtils.java @@ -17,6 +17,7 @@ package com.google.cloud.pubsublite.spark; import static com.google.common.base.Preconditions.checkArgument; +import static scala.collection.JavaConverters.asScalaBufferConverter; import com.google.cloud.pubsublite.Message; import com.google.cloud.pubsublite.Offset; @@ -51,7 +52,6 @@ import org.apache.spark.unsafe.types.UTF8String; import scala.Option; import scala.compat.java8.functionConverterImpls.FromJavaBiConsumer; -import scala.jdk.javaapi.CollectionConverters; public class PslSparkUtils { @@ -73,13 +73,12 @@ public static ArrayBasedMapData convertAttributesToSparkMap( value.stream() .map(v -> ByteArray.concat(v.toByteArray())) .collect(Collectors.toList()); - valueList.add( - new GenericArrayData(CollectionConverters.asScala(attributeVals).toSeq())); + valueList.add(new GenericArrayData(asScalaBufferConverter(attributeVals).asScala())); }); return new ArrayBasedMapData( - new GenericArrayData(CollectionConverters.asScala(keyList).toSeq()), - new GenericArrayData(CollectionConverters.asScala(valueList).toSeq())); + new GenericArrayData(asScalaBufferConverter(keyList).asScala()), + new GenericArrayData(asScalaBufferConverter(valueList).asScala())); } public static InternalRow toInternalRow( @@ -97,7 +96,7 @@ public static InternalRow toInternalRow( ? Timestamps.toMicros(msg.message().eventTime().get()) : null, convertAttributesToSparkMap(msg.message().attributes()))); - return InternalRow.apply(CollectionConverters.asScala(list).toSeq()); + return InternalRow.apply(asScalaBufferConverter(list).asScala()); } @SuppressWarnings("unchecked") diff --git a/src/test/java/com/google/cloud/pubsublite/spark/PslSparkUtilsTest.java b/src/test/java/com/google/cloud/pubsublite/spark/PslSparkUtilsTest.java index 39747612..d76b668f 100644 --- a/src/test/java/com/google/cloud/pubsublite/spark/PslSparkUtilsTest.java +++ b/src/test/java/com/google/cloud/pubsublite/spark/PslSparkUtilsTest.java @@ -18,6 +18,7 @@ import static com.google.common.truth.Truth.assertThat; import static org.junit.Assert.assertThrows; +import static scala.collection.JavaConverters.asScalaBufferConverter; import com.google.cloud.pubsublite.Message; import com.google.cloud.pubsublite.Offset; @@ -43,7 +44,6 @@ import org.apache.spark.sql.types.StructType; import org.apache.spark.unsafe.types.ByteArray; import org.junit.Test; -import scala.jdk.javaapi.CollectionConverters; public class PslSparkUtilsTest { @@ -138,7 +138,7 @@ public void testToPubSubMessage() { PslSparkUtils.convertAttributesToSparkMap(message.attributes()), Timestamps.toMicros(message.eventTime().get()), "abc".getBytes(StandardCharsets.UTF_8))); - InternalRow row = InternalRow.apply(CollectionConverters.asScala(list).toSeq()); + InternalRow row = InternalRow.apply(asScalaBufferConverter(list).asScala()); StructType structType = new StructType( @@ -164,7 +164,7 @@ public void testToPubSubMessageLongForEventTimestamp() { new StructField("event_timestamp", DataTypes.LongType, false, Metadata.empty()) }); List list = Collections.singletonList(100000L); - InternalRow row = InternalRow.apply(CollectionConverters.asScala(list).toSeq()); + InternalRow row = InternalRow.apply(asScalaBufferConverter(list).asScala()); Message message = PslSparkUtils.toPubSubMessage(structType, row); assertThat(message).isEqualTo(expectedMsg); From 52b9f98479bce64ccc42b9de2ae7efdd4240818a Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Wed, 9 Jul 2025 15:35:43 +0000 Subject: [PATCH 37/38] Add spark-sql-api undeclared dependency --- pom.xml | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/pom.xml b/pom.xml index e640840a..5afe5693 100644 --- a/pom.xml +++ b/pom.xml @@ -114,6 +114,12 @@ ${spark.version} provided + + org.apache.spark + spark-sql-api_${scala.version.short} + ${spark.version} + provided + org.apache.spark spark-unsafe_${scala.version.short} From c18c5ff7eecf3204710d3b9f408eb7860b5764f4 Mon Sep 17 00:00:00 2001 From: Anna Cocuzzo <63511057+acocuzzo@users.noreply.github.com> Date: Wed, 9 Jul 2025 16:20:23 +0000 Subject: [PATCH 38/38] Update to 3.5.5 again and fix warnings --- pom.xml | 2 +- .../java/com/google/cloud/pubsublite/spark/BaseDataStream.java | 1 + .../pubsublite/spark/internal/LimitingHeadOffsetReader.java | 1 + .../pubsublite/spark/internal/MultiPartitionCommitter.java | 1 + 4 files changed, 4 insertions(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index 5afe5693..1c3dc5b2 100644 --- a/pom.xml +++ b/pom.xml @@ -18,7 +18,7 @@ UTF-8 2.12.20 2.12 - 3.5.3 + 3.5.5 3.2.2 diff --git a/src/main/java/com/google/cloud/pubsublite/spark/BaseDataStream.java b/src/main/java/com/google/cloud/pubsublite/spark/BaseDataStream.java index d95a1476..e91fbd4a 100644 --- a/src/main/java/com/google/cloud/pubsublite/spark/BaseDataStream.java +++ b/src/main/java/com/google/cloud/pubsublite/spark/BaseDataStream.java @@ -81,6 +81,7 @@ public void commit(org.apache.spark.sql.connector.read.streaming.Offset end) { } @Override + @SuppressWarnings("try") public void stop() { try (AutoCloseable a = committer; AutoCloseable b = cursorClient; diff --git a/src/main/java/com/google/cloud/pubsublite/spark/internal/LimitingHeadOffsetReader.java b/src/main/java/com/google/cloud/pubsublite/spark/internal/LimitingHeadOffsetReader.java index a974ba23..48b9602a 100644 --- a/src/main/java/com/google/cloud/pubsublite/spark/internal/LimitingHeadOffsetReader.java +++ b/src/main/java/com/google/cloud/pubsublite/spark/internal/LimitingHeadOffsetReader.java @@ -101,6 +101,7 @@ public PslSourceOffset getHeadOffset() { } @Override + @SuppressWarnings("try") public void close() { try (AutoCloseable a = topicStatsClient; Closeable b = partitionCountReader) { diff --git a/src/main/java/com/google/cloud/pubsublite/spark/internal/MultiPartitionCommitter.java b/src/main/java/com/google/cloud/pubsublite/spark/internal/MultiPartitionCommitter.java index bf6441e8..def26a6e 100644 --- a/src/main/java/com/google/cloud/pubsublite/spark/internal/MultiPartitionCommitter.java +++ b/src/main/java/com/google/cloud/pubsublite/spark/internal/MultiPartitionCommitter.java @@ -29,5 +29,6 @@ interface CommitterFactory { void commit(PslSourceOffset offset); + @Override void close(); }