diff --git a/pom.xml b/pom.xml
index 2e30905d..1c3dc5b2 100644
--- a/pom.xml
+++ b/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>
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/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
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/PslWrite.java b/src/main/java/com/google/cloud/pubsublite/spark/PslWrite.java
index c9868107..7f368a9b 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,9 @@ public BatchWrite buildForBatch() {
   public StreamingWrite buildForStreaming() {
     return this;
   }
+
+  @Override
+  public boolean useCommitCoordinator() {
+    return true;
+  }
 }
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();
 }