Deps: Update org.apache.spark:spark-sql_2.12 to 3.5.5 (googleapis#703)

* Deps: Update to apache spark-sql 3.5.5

* Update scala version to 2.13.8

* Update spark in snippets/pom.xml

* Update pom.xml

* Upgrade to scala 2.11.12

* Upgrade scale to 2.12.20

* Set to spark-sql_3.12

* spark-sql_2.12

* Update PslSparkUtils.java with scala 2.13 CollectionConverters

* Update pom.xml with scala 2.13.16

* Update snippets/pom.xml with 2.13

* Update samples/snapshot/pom.xml with 2.12

* Update PslSparkUtils.java fix import

* Update PslWrite.java to override  useCommitCoordinator()

* Update PslWrite.java

* Fix broken pom.xml

* Fix scala in PslSparkUtilsTest

* Add spark-sql-api

* Fix lint errors

* Fix commons-lang missing dependency

* Remove dependency on StringUtils

* remove uneeded import

* Fix String.startsWith

* Set useCommitCoordinator true

* Unpin scala

* removing spark pinning

* Update current version

* Update dataproc image version

* Temporarily downgrade to 3.5.3 to test on dataproc

* Update samples README and versions.txt

* Revert pom changes in samples

* Revert "Remove dependency on StringUtils"

This reverts commit ac0e9fb.

* Separate versioning from PR

* Revert string utils update

* Revert scala version to 2.12.20

* Revert CollectionConverter changes

* Add spark-sql-api undeclared dependency

* Update to 3.5.5 again and fix warnings

Author: acocuzzo

pom.xml

@@ -16,9 +16,9 @@
     <maven.compiler.source>1.8</maven.compiler.source>
     <maven.compiler.target>1.8</maven.compiler.target>
     <encoding>UTF-8</encoding>
-    <scala.version>2.12.15</scala.version>
+    <scala.version>2.12.20</scala.version>
     <scala.version.short>2.12</scala.version.short>
-    <spark.version>3.1.2</spark.version>
+    <spark.version>3.5.5</spark.version>
     <hadoop.version>3.2.2</hadoop.version>
   </properties>
   <dependencyManagement>
@@ -114,6 +114,12 @@
       <version>${spark.version}</version>
       <scope>provided</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.spark</groupId>
+      <artifactId>spark-sql-api_${scala.version.short}</artifactId>
+      <version>${spark.version}</version>
+      <scope>provided</scope>
+    </dependency>
     <dependency>
       <groupId>org.apache.spark</groupId>
       <artifactId>spark-unsafe_${scala.version.short}</artifactId>

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

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

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

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

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;

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

@@ -102,4 +102,9 @@ public BatchWrite buildForBatch() {
   public StreamingWrite buildForStreaming() {
     return this;
   }
+
+  @Override
+  public boolean useCommitCoordinator() {
+    return true;
+  }
 }

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) {

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

@@ -29,5 +29,6 @@ interface CommitterFactory {
 
   void commit(PslSourceOffset offset);
 
+  @Override
   void close();
 }