(flink-connector-jdbc) branch main updated: [FLINK-41208][connector-jdbc] Only accept TYPE_SCROLL_INSENSITIVE for exactly-once JdbcSource

[martijnvisser]

This is an automated email from the ASF dual-hosted git repository.

MartijnVisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-jdbc.git


The following commit(s) were added to refs/heads/main by this push:
     new 5e625825 [FLINK-41208][connector-jdbc] Only accept 
TYPE_SCROLL_INSENSITIVE for exactly-once JdbcSource
5e625825 is described below

commit 5e625825418e1d1c1b2e1db2d094ff27c08fb313
Author: Purushottam Sinha <psi...@confluent.io>
AuthorDate: Mon Oct 26 19:10:44 2026 +0530

    [FLINK-41208][connector-jdbc] Only accept TYPE_SCROLL_INSENSITIVE for 
exactly-once JdbcSource
    
    JdbcSourceBuilder.build() also accepted ResultSet.CONCUR_READ_ONLY as the
    result set type for exactly-once, but that is a concurrency constant, not
    a result set type.
    
    Generated-by: Claude Code (Claude Opus 5.5)
---
 .../jdbc/core/datastream/source/JdbcSourceBuilder.java |  5 ++---
 .../core/datastream/source/JdbcSourceBuilderTest.java  | 18 ++++++++++++++++++
 2 files changed, 20 insertions(+), 3 deletions(-)

diff --git 
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceBuilder.java
 
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceBuilder.java
index 3d370a56..be32e1fc 100644
--- 
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceBuilder.java
+++ 
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceBuilder.java
@@ -306,9 +306,8 @@ public class JdbcSourceBuilder<OUT> {
 
         if (deliveryGuarantee == DeliveryGuarantee.EXACTLY_ONCE) {
             Preconditions.checkArgument(
-                    this.resultSetType == ResultSet.TYPE_SCROLL_INSENSITIVE
-                            || this.resultSetType == 
ResultSet.CONCUR_READ_ONLY,
-                    "The 'resultSetType' must be 
ResultSet.TYPE_SCROLL_INSENSITIVE or ResultSet.CONCUR_READ_ONLY when using %s",
+                    this.resultSetType == ResultSet.TYPE_SCROLL_INSENSITIVE,
+                    "The 'resultSetType' must be 
ResultSet.TYPE_SCROLL_INSENSITIVE when using %s",
                     DeliveryGuarantee.EXACTLY_ONCE);
         }
 
diff --git 
a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceBuilderTest.java
 
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceBuilderTest.java
index 488ae807..858e0ea8 100644
--- 
a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceBuilderTest.java
+++ 
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceBuilderTest.java
@@ -20,6 +20,7 @@ package 
org.apache.flink.connector.jdbc.core.datastream.source;
 
 import org.apache.flink.api.common.typeinfo.TypeHint;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.connector.base.DeliveryGuarantee;
 import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
 import 
org.apache.flink.connector.jdbc.core.datastream.source.config.ContinuousUnBoundingSettings;
 import 
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter.JdbcSqlSplitterEnumerator;
@@ -37,6 +38,7 @@ import org.apache.flink.types.Row;
 import org.junit.jupiter.api.Test;
 
 import java.io.Serializable;
+import java.sql.ResultSet;
 import java.time.Duration;
 
 import static 
org.apache.flink.connector.jdbc.core.datastream.source.JdbcSourceOptions.READER_FETCH_BATCH_SIZE;
@@ -211,6 +213,22 @@ class JdbcSourceBuilderTest {
                 
.hasMessage(JdbcSourceBuilder.INVALID_CONTINUOUS_SLIDE_TIMING_HINT);
     }
 
+    @Test
+    void testExactlyOnceRequiresScrollInsensitiveResultSetType() {
+        sourceBuilder.setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE);
+
+        assertThatThrownBy(
+                        () -> 
sourceBuilder.setResultSetType(ResultSet.TYPE_FORWARD_ONLY).build())
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("ResultSet.TYPE_SCROLL_INSENSITIVE");
+        assertThatThrownBy(() -> 
sourceBuilder.setResultSetType(ResultSet.CONCUR_READ_ONLY).build())
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("ResultSet.TYPE_SCROLL_INSENSITIVE");
+
+        
assertThat(sourceBuilder.setResultSetType(ResultSet.TYPE_SCROLL_INSENSITIVE).build())
+                .isNotNull();
+    }
+
     @Test
     void testSetConnectionProvider() {
         assertThatThrownBy(() -> 
JdbcSource.builder().setConnectionProvider(null))