fix: Upgrade 3PP versions and fix gRPC stream leak exposed by gRPC 1.80 (#168)

## Versioning restructuring

Netty version management was previously split across two separate
version entries (io-netty for netty-common at 4.2.x and netty-codec at
4.1.x). This is consolidated into a single `netty` version (4.1.132)
applied uniformly via dependency constraints in
java-base-conventions.gradle.kts. The soft-pinning rule in
version-updates.gradle.kts now applies to all io.netty modules (not just
netty-codec), since gRPC requires the 4.1.x line.

Jackson versions were similarly consolidated from two separate entries
into a single `jackson` version.

JUnit library versions are now managed via the BOM instead of explicit
per-artifact versions. The soft-pinning rule was split to handle JUnit
Platform (1.x) separately from JUnit Jupiter (5.x).

## gRPC stream close fix

gRPC 1.80 is stricter about channel shutdown: `channel.shutdown()` now
properly waits for all active streams to complete before terminating.
This exposed a pre-existing stream leak in the driver where closing a
ResultSet did not cancel the underlying gRPC stream.

The leak path was: ArrowStreamReader wraps ByteStringReadableByteChannel
which wraps the gRPC iterator, but ByteStringReadableByteChannel.close()
did not propagate close to the iterator, leaving orphaned gRPC streams
on the channel.

Fix: Introduce CloseableIterator<T> (Iterator + AutoCloseable) and use
it throughout the stream chain. ByteStringReadableByteChannel now
accepts a CloseableIterator and closes it when the channel is closed.
SQLExceptionQueryResultIterator implements CloseableIterator and
delegates close to the wrapped gRPC iterator. This ensures the full
close chain works: StreamingResultSet.close() → ArrowStreamReader →
ByteStringReadableByteChannel → gRPC stream cancel.

JdbcDriverStubProvider.close() is also changed from graceful shutdown()
to shutdownNow() as defense-in-depth, since Connection.close() means the
caller is done and there is no need to wait for in-flight RPCs.

## Dependency upgrades

- gRPC 1.78.0 → 1.80.0
- protobuf 4.33.4 → 4.34.1
- grpcmock 0.16.0 → 1.0.0
- jackson 2.21.0 → 2.21.2
- log4j 2.25.3 → 2.25.4
- JUnit BOM 5.14.2 → 5.14.3
- PostgreSQL JDBC 42.7.9 → 42.7.10
- scalatest 3.2.19 → 3.2.20
- shadow plugin 9.3.1 → 9.4.1
- buf plugin 0.10.3 → 0.11.0
- Netty 4.1.130 → 4.1.132

Author: mkaufmann

buildSrc/src/main/kotlin/java-base-conventions.gradle.kts

@@ -13,6 +13,39 @@ repositories {
     }
 }
 
+val nettyVersion = extensions.getByType<VersionCatalogsExtension>()
+    .named("libs")
+    .findVersion("netty")
+    .get()
+    .requiredVersion
+
+// Exclude grpc-netty-shaded (pulled in transitively by grpcmock) to prevent it from winning
+// gRPC provider discovery over our grpc-netty. The shaded variant has higher priority, and
+// mixing its older transport with our grpc-core version causes channels to never terminate.
+configurations.all {
+    exclude(group = "io.grpc", module = "grpc-netty-shaded")
+}
+
+dependencies {
+    constraints {
+        // gRPC and Arrow pull in Netty 4.1.x transitively. These constraints enforce a minimum
+        // version to fix security vulnerabilities. Keep the version in gradle/libs.versions.toml.
+        listOf(
+            "io.netty:netty-buffer",
+            "io.netty:netty-codec",
+            "io.netty:netty-codec-http",
+            "io.netty:netty-codec-http2",
+            "io.netty:netty-common",
+            "io.netty:netty-handler",
+            "io.netty:netty-resolver",
+            "io.netty:netty-transport",
+            "io.netty:netty-transport-native-unix-common",
+        ).forEach { module ->
+            implementation("$module:$nettyVersion")
+        }
+    }
+}
+
 tasks.withType<Test>().configureEach {
     javaLauncher = javaToolchains.launcherFor {
         languageVersion = JavaLanguageVersion.of(8)

buildSrc/src/main/kotlin/version-updates.gradle.kts

@@ -16,8 +16,11 @@ versionCatalogUpdate {
         val regex = "^[0-9,.v-]+(-r)?$".toRegex()
 
         // Special cases for libraries where we have to do soft pinning. We don't want to fully pin to still get the latest version for the allowed range.
-        if (("org.junit" in it.candidate.group) && !candidateVersion.startsWith("5.")) {
-            // This is soft pinned to 5.* (as 6 is not Java 8 compatible)
+        if ("org.junit.platform" in it.candidate.group && !candidateVersion.startsWith("1.")) {
+            // JUnit Platform uses 1.x versioning; pinned to 1.* (corresponding to JUnit Jupiter 5.x)
+            false
+        } else if ("org.junit" in it.candidate.group && !candidateVersion.startsWith("5.")) {
+            // JUnit BOM and Jupiter are soft pinned to 5.* (as 6 is not Java 8 compatible)
             false
         } else if ("org.apache.spark" in it.candidate.group && !candidateVersion.startsWith("3.")) {
             // This is soft pinned to 3.* (as the driver is targetting in Spark 3)
@@ -28,9 +31,8 @@ versionCatalogUpdate {
         } else if ("org.mockito" in it.candidate.group && !candidateVersion.startsWith("4.")) {
             //This is soft pinned to 4.* (as 5 is not Java 8 compatible)
             false
-        } else if ("io.netty" in it.candidate.group && "netty-codec" in it.candidate.module && !candidateVersion.startsWith("4.1.")) {
-            // netty-codec is soft pinned to 4.1.* (as 4.2+ may have compatibility issues)
-            // This restriction applies only to the netty-codec version entry, not other netty libraries
+        } else if ("io.netty" in it.candidate.group && !candidateVersion.startsWith("4.1.")) {
+            // Netty is soft pinned to 4.1.* as gRPC requires the 4.1.x line
             false
         } else {
             stableKeyword || regex.matches(candidateVersion)

gradle/libs.versions.toml

@@ -4,21 +4,20 @@
 apache-httpcomponents-httpclient = "5.6"
 # @pin this version as it is the last Java 8 compatible version of Arrow
 arrow = "17.0.0"
-com-fasterxml-jackson-core-jackson-databind = "2.21.0"
-com-fasterxml-jackson-module-jackson-module-scala = "2.21.0"
+jackson = "2.21.2"
 # This has special version logic in `buildSrc/src/main/kotlin/version-updates.gradle.kts` to also pick the -jre version
 com-google-guava-guava = "33.5.0-jre"
-grpc = "1.78.0"
-# This is only indirectly used by gRPC & Arrow but we provide an explicit version to fix security issues in the transitive dependencies
-io-netty = "4.2.9.Final"
+grpc = "1.80.0"
 javax-annotation-javax-annotation-api = "1.3.2"
 jjwt = "0.13.0"
-log4j-core = "2.25.3"
+log4j-core = "2.25.4"
 # This is soft pinned to 4.* (as 5 is not Java 8 compatible) in `buildSrc/src/main/kotlin/version-updates.gradle.kts`
 mockito = "4.11.0"
 net-jodah-failsafe = "3.3.2"
-# This is soft pinned to 4.1.* (as 4.2+ may have compatibility issues) in `buildSrc/src/main/kotlin/version-updates.gradle.kts`
-netty-codec = "4.1.130.Final"
+# Minimum Netty 4.1.x version, enforced via dependency constraints in `java-base-conventions.gradle.kts`
+# to fix security vulnerabilities in transitive dependencies pulled in by gRPC and Arrow.
+# This is soft pinned to 4.1.* in `buildSrc/src/main/kotlin/version-updates.gradle.kts`
+netty = "4.1.132.Final"
 # This is soft pinned to 4.* (as we didn't invest in the upgrade yet) in `buildSrc/src/main/kotlin/version-updates.gradle.kts`
 okhttp3 = "4.12.0"
 # @pin this version as the following versions have a breaking Cursor change, that - with the upcoming Avactica removal - we currently don't want to invest in
@@ -27,18 +26,18 @@ org-apache-commons-commons-lang3 = "3.20.0"
 # This is soft pinned to 3.* (as the driver is targetting in Spark 3) in `buildSrc/src/main/kotlin/version-updates.gradle.kts`
 org-apache-spark = "3.5.8"
 org-assertj-assertj-core = "3.27.7"
-org-grpcmock-grpcmock-junit5 = "0.16.0"
+org-grpcmock-grpcmock-junit5 = "1.0.0"
+# JUnit BOM version — manages both Jupiter (5.x) and Platform (1.x) library versions.
 # This is soft pinned to 5.* (as 6 is not Java 8 compatible) in `buildSrc/src/main/kotlin/version-updates.gradle.kts`
-org-junit-bom = "5.14.2"
-org-junit-platform-junit-platform-launcher = "1.14.0"
-org-postgresql-pgjdbc = "42.7.9"
-org-scalatest = "3.2.19"
+org-junit-bom = "5.14.3"
+org-postgresql-pgjdbc = "42.7.10"
+org-scalatest = "3.2.20"
 org-scalatestplus-junit5 = "3.2.19.0"
-plugin-build-buf = "0.10.3"
+plugin-build-buf = "0.11.0"
 plugin-com-google-protobuf = "0.9.6"
-plugin-com-gradleup-shadow = "9.3.1"
+plugin-com-gradleup-shadow = "9.4.1"
 plugin-freefair-lombok = "9.2.0"
-protobuf = "4.33.4"
+protobuf = "4.34.1"
 slf4j = "2.0.17"
 
 [libraries]
@@ -56,25 +55,25 @@ grpc-protoc = { module = "io.grpc:protoc-gen-grpc-java", version.ref = "grpc" }
 grpc-stub = { module = "io.grpc:grpc-stub", version.ref = "grpc" }
 grpcmock = { module = "org.grpcmock:grpcmock-junit5", version.ref = "org-grpcmock-grpcmock-junit5" }
 guava = { module = "com.google.guava:guava", version.ref = "com-google-guava-guava" }
-jackson-databind = { module = "com.fasterxml.jackson.core:jackson-databind", version.ref = "com-fasterxml-jackson-core-jackson-databind" }
-jackson-module-scala = { module = "com.fasterxml.jackson.module:jackson-module-scala_2.13", version.ref = "com-fasterxml-jackson-module-jackson-module-scala" }
+jackson-databind = { module = "com.fasterxml.jackson.core:jackson-databind", version.ref = "jackson" }
+jackson-module-scala = { module = "com.fasterxml.jackson.module:jackson-module-scala_2.13", version.ref = "jackson" }
 javax-annotation-api = { module = "javax.annotation:javax.annotation-api", version.ref = "javax-annotation-javax-annotation-api" }
 jjwt-api = { module = "io.jsonwebtoken:jjwt-api", version.ref = "jjwt" }
 jjwt-impl = { module = "io.jsonwebtoken:jjwt-impl", version.ref = "jjwt" }
 jjwt-jackson = { module = "io.jsonwebtoken:jjwt-jackson", version.ref = "jjwt" }
-junit-bom = "org.junit:junit-bom:5.14.2"
-junit-jupiter-api = { module = "org.junit.jupiter:junit-jupiter-api", version.ref = "org-junit-bom" }
-junit-jupiter-base = { module = "org.junit.jupiter:junit-jupiter", version.ref = "org-junit-bom" }
-junit-jupiter-engine = { module = "org.junit.jupiter:junit-jupiter-engine", version.ref = "org-junit-bom" }
-junit-jupiter-params = { module = "org.junit.jupiter:junit-jupiter-params", version.ref = "org-junit-bom" }
-junit-platform-engine = { module = "org.junit.platform:junit-platform-engine", version.ref = "org-junit-platform-junit-platform-launcher" }
-junit-platform-launcher = { module = "org.junit.platform:junit-platform-launcher", version.ref = "org-junit-platform-junit-platform-launcher" }
-junit-platform-launcher-test = { module = "org.junit.platform:junit-platform-launcher", version.ref = "org-junit-platform-junit-platform-launcher" }
+# Version managed by org-junit-bom; individual JUnit libraries below are versionless (BOM-managed)
+junit-bom = { module = "org.junit:junit-bom", version.ref = "org-junit-bom" }
+junit-jupiter-api = { module = "org.junit.jupiter:junit-jupiter-api" }
+junit-jupiter-base = { module = "org.junit.jupiter:junit-jupiter" }
+junit-jupiter-engine = { module = "org.junit.jupiter:junit-jupiter-engine" }
+junit-jupiter-params = { module = "org.junit.jupiter:junit-jupiter-params" }
+junit-platform-engine = { module = "org.junit.platform:junit-platform-engine" }
+junit-platform-launcher = { module = "org.junit.platform:junit-platform-launcher" }
 log4j-core = { module = "org.apache.logging.log4j:log4j-core", version.ref = "log4j-core" }
 mockito-inline = { module = "org.mockito:mockito-inline", version.ref = "mockito" }
 mockito-junit-jupiter = { module = "org.mockito:mockito-junit-jupiter", version.ref = "mockito" }
-netty-codec = { module = "io.netty:netty-codec", version.ref = "netty-codec" }
-netty-common = { module = "io.netty:netty-common", version.ref = "io-netty" }
+netty-codec = { module = "io.netty:netty-codec", version.ref = "netty" }
+netty-common = { module = "io.netty:netty-common", version.ref = "netty" }
 okhttp3-client = { module = "com.squareup.okhttp3:okhttp", version.ref = "okhttp3" }
 okhttp3-logging-interceptor = { module = "com.squareup.okhttp3:logging-interceptor", version.ref = "okhttp3" }
 okhttp3-mockwebserver = { module = "com.squareup.okhttp3:mockwebserver", version.ref = "okhttp3" }
@@ -112,7 +111,7 @@ mocking = [
 ]
 scala-testing = [
     "junit-platform-engine",
-    "junit-platform-launcher-test",
+    "junit-platform-launcher",
     "scalatest",
     "scalatestplus-junit5",
 ]

jdbc-core/src/main/java/com/salesforce/datacloud/jdbc/core/ByteStringReadableByteChannel.java

@@ -5,24 +5,24 @@
 package com.salesforce.datacloud.jdbc.core;
 
 import com.google.protobuf.ByteString;
+import com.salesforce.datacloud.jdbc.protocol.CloseableIterator;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ClosedChannelException;
 import java.nio.channels.ReadableByteChannel;
-import java.util.Iterator;
 import lombok.NonNull;
 
 /**
  * A ReadableByteChannel that exposes an Iterator<ByteString> as a stream of bytes.
  * This class has a single responsibility: converting ByteString iterator to byte stream.
  */
 public class ByteStringReadableByteChannel implements ReadableByteChannel {
-    @NonNull private final Iterator<ByteString> iterator;
+    @NonNull private final CloseableIterator<ByteString> iterator;
 
     private boolean open = true;
     private ByteBuffer currentBuffer = null;
 
-    public ByteStringReadableByteChannel(@NonNull Iterator<ByteString> iterator) {
+    public ByteStringReadableByteChannel(@NonNull CloseableIterator<ByteString> iterator) {
         this.iterator = iterator;
     }
 
@@ -61,6 +61,13 @@ public boolean isOpen() {
     @Override
     public void close() throws IOException {
         open = false;
+        try {
+            iterator.close();
+        } catch (IOException e) {
+            throw e;
+        } catch (Exception e) {
+            throw new IOException("Failed to close underlying resource", e);
+        }
     }
 
     private static int transferToDestination(ByteBuffer source, ByteBuffer destination) {

jdbc-core/src/main/java/com/salesforce/datacloud/jdbc/core/DataCloudStatement.java

@@ -195,6 +195,12 @@ public int executeUpdate(String sql) throws SQLException {
         throw new SQLException(NOT_SUPPORTED_IN_DATACLOUD_QUERY, SqlErrorCodes.FEATURE_NOT_SUPPORTED);
     }
 
+    /**
+     * Closes the statement and releases any resources associated with it. Might cancel the query
+     * if it was not yet completed.
+     *
+     * @throws SQLException
+     */
     @Override
     public void close() throws SQLException {
         log.debug("Entering close");

jdbc-core/src/main/java/com/salesforce/datacloud/jdbc/core/JdbcDriverStubProvider.java

@@ -53,16 +53,20 @@ public void close() {
             return;
         }
 
-        channel.shutdown();
+        // Use shutdownNow() to cancel any in-flight RPCs. A JDBC Connection.close() means the
+        // caller is done with the connection, so immediate cancellation is appropriate.
+        // Note: With gRPC 1.80+, graceful shutdown() can hang if streams aren't fully closed,
+        // e.g., when result sets are not explicitly closed before closing the connection.
+        channel.shutdownNow();
 
         try {
+            // Brief wait after forceful shutdown. Any remaining transport cleanup happens
+            // asynchronously in the background and will be handled by the JVM.
             if (!channel.awaitTermination(5, TimeUnit.SECONDS)) {
-                log.warn("Channel did not terminate within 5 seconds, forcing shutdown");
-                channel.shutdownNow();
+                log.debug("Channel transport still shutting down in background after shutdownNow");
             }
         } catch (InterruptedException e) {
-            log.warn("Channel shutdown interrupted, forcing immediate termination", e);
-            channel.shutdownNow();
+            log.warn("Channel shutdown interrupted", e);
             Thread.currentThread().interrupt();
         }
     }

jdbc-core/src/main/java/com/salesforce/datacloud/jdbc/core/SQLExceptionQueryResultIterator.java

@@ -5,9 +5,9 @@
 package com.salesforce.datacloud.jdbc.core;
 
 import com.salesforce.datacloud.jdbc.exception.QueryExceptionHandler;
+import com.salesforce.datacloud.jdbc.protocol.CloseableIterator;
 import com.salesforce.datacloud.jdbc.protocol.QueryResultArrowStream;
 import io.grpc.StatusRuntimeException;
-import java.util.Iterator;
 import lombok.AccessLevel;
 import lombok.AllArgsConstructor;
 import lombok.SneakyThrows;
@@ -26,12 +26,17 @@
  * @see QueryExceptionHandler
  */
 @AllArgsConstructor(access = AccessLevel.PRIVATE)
-class SQLExceptionQueryResultIterator implements Iterator<QueryResult> {
-    Iterator<QueryResult> grpcIterator;
+class SQLExceptionQueryResultIterator implements CloseableIterator<QueryResult> {
+    CloseableIterator<QueryResult> resultIterator;
     boolean includeCustomerDetail;
     String queryId;
     String sql;
 
+    @Override
+    public void close() throws Exception {
+        resultIterator.close();
+    }
+
     /**
      * Creates an {@link ArrowStreamReader} that wraps the given iterator with SQL exception handling.
      * <p>
@@ -47,7 +52,7 @@ class SQLExceptionQueryResultIterator implements Iterator<QueryResult> {
      * @return an {@link ArrowStreamReader} that converts gRPC exceptions to SQL exceptions
      */
     public static ArrowStreamReader createSqlExceptionArrowStreamReader(
-            Iterator<QueryResult> resultIterator, boolean includeCustomerDetail, String queryId, String sql) {
+            CloseableIterator<QueryResult> resultIterator, boolean includeCustomerDetail, String queryId, String sql) {
         val throwingSqlExceptionIterator =
                 new SQLExceptionQueryResultIterator(resultIterator, includeCustomerDetail, queryId, sql);
         return QueryResultArrowStream.toArrowStreamReader(throwingSqlExceptionIterator);
@@ -67,7 +72,7 @@ public static ArrowStreamReader createSqlExceptionArrowStreamReader(
     @Override
     public boolean hasNext() {
         try {
-            return grpcIterator.hasNext();
+            return resultIterator.hasNext();
         } catch (StatusRuntimeException ex) {
             throw QueryExceptionHandler.createException(includeCustomerDetail, sql, queryId, ex);
         }
@@ -88,7 +93,7 @@ public boolean hasNext() {
     @Override
     public QueryResult next() {
         try {
-            return grpcIterator.next();
+            return resultIterator.next();
         } catch (StatusRuntimeException ex) {
             throw QueryExceptionHandler.createException(includeCustomerDetail, queryId, sql, ex);
         }

jdbc-core/src/main/java/com/salesforce/datacloud/jdbc/core/StreamingResultSet.java

@@ -6,8 +6,6 @@
 
 import static com.salesforce.datacloud.jdbc.util.ArrowUtils.toColumnMetaData;
 
-import com.salesforce.datacloud.jdbc.util.ThrowingJdbcSupplier;
-import com.salesforce.datacloud.query.v3.QueryStatus;
 import java.io.IOException;
 import java.sql.ResultSet;
 import java.sql.ResultSetMetaData;
@@ -30,7 +28,6 @@ public class StreamingResultSet extends AvaticaResultSet implements DataCloudRes
     private final String queryId;
 
     private final ArrowStreamReaderCursor cursor;
-    ThrowingJdbcSupplier<QueryStatus> getQueryStatus;
     private final ColumnNameResolver columnNameResolver;
 
     private StreamingResultSet(
@@ -68,6 +65,15 @@ public static StreamingResultSet of(ArrowStreamReader resultStream, String query
         }
     }
 
+    @Override
+    public void close() {
+        try {
+            cursor.close();
+        } finally {
+            super.close();
+        }
+    }
+
     @Override
     public int getType() {
         return ResultSet.TYPE_FORWARD_ONLY;

jdbc-core/src/main/java/com/salesforce/datacloud/jdbc/protocol/CloseableIterator.java

@@ -0,0 +1,30 @@
+/**
+ * This file is part of https://github.com/forcedotcom/datacloud-jdbc which is released under the
+ * Apache 2.0 license. See https://github.com/forcedotcom/datacloud-jdbc/blob/main/LICENSE.txt
+ */
+package com.salesforce.datacloud.jdbc.protocol;
+
+import java.util.Iterator;
+
+public interface CloseableIterator<T> extends Iterator<T>, AutoCloseable {
+
+    /**
+     * Wraps a plain iterator as a CloseableIterator with a no-op close.
+     */
+    static <T> CloseableIterator<T> of(Iterator<T> iterator) {
+        return new CloseableIterator<T>() {
+            @Override
+            public boolean hasNext() {
+                return iterator.hasNext();
+            }
+
+            @Override
+            public T next() {
+                return iterator.next();
+            }
+
+            @Override
+            public void close() {}
+        };
+    }
+}