Updated based on review comments from Boto.

Author: RocMarshal

flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/source/JdbcSource.java

@@ -40,7 +40,7 @@
 import org.apache.flink.connector.jdbc.source.reader.extractor.ResultExtractor;
 import org.apache.flink.connector.jdbc.source.split.JdbcSourceSplit;
 import org.apache.flink.connector.jdbc.source.split.JdbcSourceSplitSerializer;
-import org.apache.flink.connector.jdbc.utils.ContinuousEnumerationSettings;
+import org.apache.flink.connector.jdbc.utils.ContinuousUnBoundingSettings;
 import org.apache.flink.core.io.SimpleVersionedSerializer;
 import org.apache.flink.util.Preconditions;
 
@@ -58,7 +58,7 @@ public class JdbcSource<OUT>
 
     private final Boundedness boundedness;
     private final TypeInformation<OUT> typeInformation;
-    private final @Nullable ContinuousEnumerationSettings continuousEnumerationSettings;
+    private final @Nullable ContinuousUnBoundingSettings continuousUnBoundingSettings;
 
     private final Configuration configuration;
     private final JdbcSqlSplitEnumeratorBase.Provider<JdbcSourceSplit> sqlSplitEnumeratorProvider;
@@ -74,17 +74,17 @@ public class JdbcSource<OUT>
             ResultExtractor<OUT> resultExtractor,
             TypeInformation<OUT> typeInformation,
             @Nullable DeliveryGuarantee deliveryGuarantee,
-            @Nullable ContinuousEnumerationSettings continuousEnumerationSettings) {
+            @Nullable ContinuousUnBoundingSettings continuousUnBoundingSettings) {
         this.configuration = Preconditions.checkNotNull(configuration);
         this.connectionProvider = Preconditions.checkNotNull(connectionProvider);
         this.sqlSplitEnumeratorProvider = Preconditions.checkNotNull(sqlSplitEnumeratorProvider);
         this.resultExtractor = Preconditions.checkNotNull(resultExtractor);
         this.deliveryGuarantee =
                 Objects.isNull(deliveryGuarantee) ? DeliveryGuarantee.NONE : deliveryGuarantee;
         this.typeInformation = Preconditions.checkNotNull(typeInformation);
-        this.continuousEnumerationSettings = continuousEnumerationSettings;
+        this.continuousUnBoundingSettings = continuousUnBoundingSettings;
         this.boundedness =
-                Objects.isNull(continuousEnumerationSettings)
+                Objects.isNull(continuousUnBoundingSettings)
                         ? Boundedness.BOUNDED
                         : Boundedness.CONTINUOUS_UNBOUNDED;
     }
@@ -116,7 +116,7 @@ public SplitEnumerator<JdbcSourceSplit, JdbcSourceEnumeratorState> createEnumera
         return new JdbcSourceEnumerator(
                 enumContext,
                 sqlSplitEnumeratorProvider.create(),
-                continuousEnumerationSettings,
+                continuousUnBoundingSettings,
                 new ArrayList<>());
     }
 
@@ -130,7 +130,7 @@ public SplitEnumerator<JdbcSourceSplit, JdbcSourceEnumeratorState> restoreEnumer
         return new JdbcSourceEnumerator(
                 enumContext,
                 sqlSplitEnumeratorProvider.restore(optionalUserDefinedSplitEnumeratorState),
-                continuousEnumerationSettings,
+                continuousUnBoundingSettings,
                 checkpoint.getRemainingSplits());
     }
 
@@ -193,7 +193,6 @@ public boolean equals(Object o) {
                 && Objects.equals(connectionProvider, that.connectionProvider)
                 && Objects.equals(resultExtractor, that.resultExtractor)
                 && deliveryGuarantee == that.deliveryGuarantee
-                && Objects.equals(
-                        continuousEnumerationSettings, that.continuousEnumerationSettings);
+                && Objects.equals(continuousUnBoundingSettings, that.continuousUnBoundingSettings);
     }
 }

flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/source/JdbcSourceBuilder.java

@@ -29,7 +29,7 @@
 import org.apache.flink.connector.jdbc.source.reader.extractor.ResultExtractor;
 import org.apache.flink.connector.jdbc.split.JdbcParameterValuesProvider;
 import org.apache.flink.connector.jdbc.split.JdbcSlideTimingParameterProvider;
-import org.apache.flink.connector.jdbc.utils.ContinuousEnumerationSettings;
+import org.apache.flink.connector.jdbc.utils.ContinuousUnBoundingSettings;
 import org.apache.flink.types.Row;
 import org.apache.flink.util.Preconditions;
 import org.apache.flink.util.StringUtils;
@@ -101,9 +101,9 @@ public class JdbcSourceBuilder<OUT> {
     public static final Logger LOG = LoggerFactory.getLogger(JdbcSourceBuilder.class);
 
     public static final String INVALID_CONTINUOUS_SLIDE_TIMING_HINT =
-            "The 'jdbcParameterValuesProvider' must be specified with in type of 'JdbcSlideTimingParameterProvider' when using 'continuousEnumerationSettings'.";
+            "The 'jdbcParameterValuesProvider' must be specified with in type of 'JdbcSlideTimingParameterProvider' when using 'continuousUnBoundingSettings'.";
     public static final String INVALID_SLIDE_TIMING_CONTINUOUS_HINT =
-            "The 'continuousEnumerationSettings' must be specified with in type of 'ContinuousEnumerationSettings' when using 'jdbcParameterValuesProvider' in type of 'JdbcSlideTimingParameterProvider'.";
+            "The 'continuousUnBoundingSettings' must be specified with in type of 'continuousUnBoundingSettings' when using 'jdbcParameterValuesProvider' in type of 'JdbcSlideTimingParameterProvider'.";
 
     private final Configuration configuration;
 
@@ -115,7 +115,7 @@ public class JdbcSourceBuilder<OUT> {
     private Boolean autoCommit;
 
     private DeliveryGuarantee deliveryGuarantee;
-    private @Nullable ContinuousEnumerationSettings continuousEnumerationSettings;
+    private @Nullable ContinuousUnBoundingSettings continuousUnBoundingSettings;
 
     private TypeInformation<OUT> typeInformation;
 
@@ -189,19 +189,19 @@ public JdbcSourceBuilder<OUT> setTypeInformation(
     // ------ Optional ------------------------------------------------------------------
 
     /**
-     * The continuousEnumerationSettings to discovery the next available batch splits. Note: If the
+     * The continuousUnBoundingSettings to discovery the next available batch splits. Note: If the
      * value was set, the {@link #jdbcParameterValuesProvider} must specified with the {@link
      * org.apache.flink.connector.jdbc.split.JdbcSlideTimingParameterProvider}.
      */
-    public JdbcSourceBuilder<OUT> setContinuousEnumerationSettings(
-            ContinuousEnumerationSettings continuousEnumerationSettings) {
-        this.continuousEnumerationSettings = continuousEnumerationSettings;
+    public JdbcSourceBuilder<OUT> setContinuousUnBoundingSettings(
+            ContinuousUnBoundingSettings continuousUnBoundingSettings) {
+        this.continuousUnBoundingSettings = continuousUnBoundingSettings;
         return this;
     }
 
     /**
      * If the value was set as an instance of {@link JdbcSlideTimingParameterProvider}, it's
-     * required to specify the {@link #continuousEnumerationSettings}.
+     * required to specify the {@link #continuousUnBoundingSettings}.
      */
     public JdbcSourceBuilder<OUT> setJdbcParameterValuesProvider(
             @Nonnull JdbcParameterValuesProvider parameterValuesProvider) {
@@ -296,7 +296,7 @@ public JdbcSource<OUT> build() {
         Preconditions.checkNotNull(resultExtractor, "'resultExtractor' mustn't be null.");
         Preconditions.checkNotNull(typeInformation, "'typeInformation' mustn't be null.");
 
-        if (Objects.nonNull(continuousEnumerationSettings)) {
+        if (Objects.nonNull(continuousUnBoundingSettings)) {
             Preconditions.checkArgument(
                     Objects.nonNull(jdbcParameterValuesProvider)
                             && jdbcParameterValuesProvider
@@ -307,7 +307,7 @@ public JdbcSource<OUT> build() {
         if (Objects.nonNull(jdbcParameterValuesProvider)
                 && jdbcParameterValuesProvider instanceof JdbcSlideTimingParameterProvider) {
             Preconditions.checkArgument(
-                    Objects.nonNull(continuousEnumerationSettings),
+                    Objects.nonNull(continuousUnBoundingSettings),
                     INVALID_CONTINUOUS_SLIDE_TIMING_HINT);
         }
 
@@ -321,6 +321,6 @@ public JdbcSource<OUT> build() {
                 resultExtractor,
                 typeInformation,
                 deliveryGuarantee,
-                continuousEnumerationSettings);
+                continuousUnBoundingSettings);
     }
 }

flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/source/enumerator/JdbcSourceEnumerator.java

@@ -23,7 +23,7 @@
 import org.apache.flink.api.connector.source.SplitEnumerator;
 import org.apache.flink.api.connector.source.SplitEnumeratorContext;
 import org.apache.flink.connector.jdbc.source.split.JdbcSourceSplit;
-import org.apache.flink.connector.jdbc.utils.ContinuousEnumerationSettings;
+import org.apache.flink.connector.jdbc.utils.ContinuousUnBoundingSettings;
 import org.apache.flink.util.Preconditions;
 
 import org.slf4j.Logger;
@@ -51,18 +51,18 @@ public class JdbcSourceEnumerator
     private final LinkedHashMap<Integer, String> readersAwaitingSplit;
     private final List<JdbcSourceSplit> unassigned;
     private final JdbcSqlSplitEnumeratorBase<JdbcSourceSplit> sqlSplitEnumerator;
-    private final @Nullable ContinuousEnumerationSettings continuousEnumerationSettings;
+    private final @Nullable ContinuousUnBoundingSettings continuousUnBoundingSettings;
 
     public JdbcSourceEnumerator(
             SplitEnumeratorContext<JdbcSourceSplit> context,
             JdbcSqlSplitEnumeratorBase<JdbcSourceSplit> sqlSplitEnumerator,
-            ContinuousEnumerationSettings continuousEnumerationSettings,
+            ContinuousUnBoundingSettings continuousUnBoundingSettings,
             List<JdbcSourceSplit> unassigned) {
         this.context = Preconditions.checkNotNull(context);
         this.sqlSplitEnumerator = Preconditions.checkNotNull(sqlSplitEnumerator);
-        this.continuousEnumerationSettings = continuousEnumerationSettings;
+        this.continuousUnBoundingSettings = continuousUnBoundingSettings;
         this.boundedness =
-                Objects.isNull(continuousEnumerationSettings)
+                Objects.isNull(continuousUnBoundingSettings)
                         ? Boundedness.BOUNDED
                         : Boundedness.CONTINUOUS_UNBOUNDED;
         this.unassigned = Preconditions.checkNotNull(unassigned);
@@ -73,12 +73,12 @@ public JdbcSourceEnumerator(
     public void start() {
         sqlSplitEnumerator.open();
         if (boundedness == Boundedness.CONTINUOUS_UNBOUNDED
-                && Objects.nonNull(continuousEnumerationSettings)) {
+                && Objects.nonNull(continuousUnBoundingSettings)) {
             context.callAsync(
                     () -> sqlSplitEnumerator.enumerateSplits(() -> 1024 - unassigned.size() > 0),
                     this::processNewSplits,
-                    continuousEnumerationSettings.getInitialDiscoveryDelay().toMillis(),
-                    continuousEnumerationSettings.getDiscoveryInterval().toMillis());
+                    continuousUnBoundingSettings.getInitialDiscoveryDelay().toMillis(),
+                    continuousUnBoundingSettings.getDiscoveryInterval().toMillis());
         } else {
             try {
                 unassigned.addAll(sqlSplitEnumerator.enumerateSplits(() -> true));

flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/source/reader/JdbcSourceSplitReader.java

@@ -170,14 +170,26 @@ private RecordsWithSplitIds<RecordAndOffset<T>> finishSplit() {
     }
 
     private void closeResultSetAndStatement() {
+        closeResultSetIfNeeded();
+        closeStatementIfNeeded();
+    }
+
+    private void closeResultSetIfNeeded() {
         try {
             if (resultSet != null && !resultSet.isClosed()) {
                 resultSet.close();
             }
+            resultSet = null;
+        } catch (SQLException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private void closeStatementIfNeeded() {
+        try {
             if (statement != null && !statement.isClosed()) {
                 statement.close();
             }
-            resultSet = null;
             statement = null;
         } catch (SQLException e) {
             throw new RuntimeException(e);
@@ -298,6 +310,7 @@ private void openResultSetForSplitWhenAtMostOnce(JdbcSourceSplit split)
     private void openResultSetForSplitWhenExactlyOnce(JdbcSourceSplit split)
             throws SQLException, ClassNotFoundException {
         getOrEstablishConnection();
+        closeResultSetIfNeeded();
         prepareStatement(split);
         resultSet = statement.executeQuery();
         currentSplitOffset = 0;
@@ -330,6 +343,7 @@ private void moveResultSetCursorByOffset() throws SQLException {
     private void openResultSetForSplitWhenAtLeastOnce(JdbcSourceSplit split)
             throws SQLException, ClassNotFoundException {
         getOrEstablishConnection();
+        closeResultSetIfNeeded();
         prepareStatement(split);
         resultSet = statement.executeQuery();
         // AT_LEAST_ONCE
@@ -338,6 +352,7 @@ private void openResultSetForSplitWhenAtLeastOnce(JdbcSourceSplit split)
     }
 
     private void prepareStatement(JdbcSourceSplit split) throws SQLException {
+        closeStatementIfNeeded();
         statement =
                 connection.prepareStatement(
                         split.getSqlTemplate(), resultSetType, resultSetConcurrency);

flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/utils/ContinuousEnumerationSettings.java flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/utils/ContinuousUnBoundingSettings.java

@@ -31,14 +31,14 @@
  * continuous discovery and streaming mode.
  */
 @PublicEvolving
-public final class ContinuousEnumerationSettings implements Serializable {
+public final class ContinuousUnBoundingSettings implements Serializable {
 
     private static final long serialVersionUID = 1L;
 
     private final Duration initialDiscoveryDelay;
     private final Duration discoveryInterval;
 
-    public ContinuousEnumerationSettings(
+    public ContinuousUnBoundingSettings(
             Duration initialDiscoveryDelay, Duration discoveryInterval) {
         this.initialDiscoveryDelay = initialDiscoveryDelay;
         this.discoveryInterval = checkNotNull(discoveryInterval);
@@ -56,7 +56,7 @@ public Duration getInitialDiscoveryDelay() {
 
     @Override
     public String toString() {
-        return "ContinuousEnumerationSettings{"
+        return "ContinuousUnBoundingSettings{"
                 + "initialDiscoveryDelay="
                 + initialDiscoveryDelay
                 + ", discoveryInterval="
@@ -74,7 +74,7 @@ public boolean equals(Object object) {
             return false;
         }
 
-        ContinuousEnumerationSettings that = (ContinuousEnumerationSettings) object;
+        ContinuousUnBoundingSettings that = (ContinuousUnBoundingSettings) object;
         return Objects.equals(initialDiscoveryDelay, that.initialDiscoveryDelay)
                 && Objects.equals(discoveryInterval, that.discoveryInterval);
     }

flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/source/JdbcSourceBuilderTest.java

@@ -26,7 +26,7 @@
 import org.apache.flink.connector.jdbc.split.JdbcNumericBetweenParametersProvider;
 import org.apache.flink.connector.jdbc.split.JdbcParameterValuesProvider;
 import org.apache.flink.connector.jdbc.split.JdbcSlideTimingParameterProvider;
-import org.apache.flink.connector.jdbc.utils.ContinuousEnumerationSettings;
+import org.apache.flink.connector.jdbc.utils.ContinuousUnBoundingSettings;
 import org.apache.flink.types.Row;
 
 import org.junit.jupiter.api.Test;
@@ -152,8 +152,8 @@ void testSetStreamingSemantic() {
         assertThatThrownBy(
                         () ->
                                 sourceBuilder
-                                        .setContinuousEnumerationSettings(
-                                                new ContinuousEnumerationSettings(
+                                        .setContinuousUnBoundingSettings(
+                                                new ContinuousUnBoundingSettings(
                                                         Duration.ofMillis(1L),
                                                         Duration.ofMillis(1L)))
                                         .build())
@@ -170,7 +170,7 @@ void testSetStreamingSemantic() {
                 .isInstanceOf(IllegalArgumentException.class)
                 .hasMessage(JdbcSourceBuilder.INVALID_SLIDE_TIMING_CONTINUOUS_HINT);
 
-        sourceBuilder.setContinuousEnumerationSettings(null);
+        sourceBuilder.setContinuousUnBoundingSettings(null);
         assertThatThrownBy(
                         () ->
                                 sourceBuilder

flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/source/JdbcSourceStreamRelatedITCase.java

@@ -27,17 +27,18 @@
 import org.apache.flink.client.program.ClusterClient;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.connector.base.DeliveryGuarantee;
-import org.apache.flink.connector.jdbc.databases.mysql.MySqlTestBase;
+import org.apache.flink.connector.jdbc.databases.derby.DerbyTestBase;
 import org.apache.flink.connector.jdbc.source.reader.extractor.ResultExtractor;
 import org.apache.flink.connector.jdbc.split.JdbcSlideTimingParameterProvider;
 import org.apache.flink.connector.jdbc.testutils.JdbcITCaseBase;
-import org.apache.flink.connector.jdbc.utils.ContinuousEnumerationSettings;
+import org.apache.flink.connector.jdbc.utils.ContinuousUnBoundingSettings;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
 import org.apache.flink.streaming.api.functions.sink.SinkFunction;
 import org.apache.flink.test.junit5.InjectClusterClient;
 import org.apache.flink.util.Collector;
 
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -65,7 +66,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 /** Test for streaming semantic related cases of {@link JdbcSource}. */
-class JdbcSourceStreamRelatedITCase implements MySqlTestBase, JdbcITCaseBase {
+class JdbcSourceStreamRelatedITCase implements DerbyTestBase, JdbcITCaseBase {
 
     private static final long ONE_SECOND = Duration.ofSeconds(1L).toMillis();
     private static final int TESTING_PARALLELISM = 2;
@@ -75,14 +76,14 @@ class JdbcSourceStreamRelatedITCase implements MySqlTestBase, JdbcITCaseBase {
             (int) (ONE_SECOND / INTERVAL_OF_GENERATING + 1);
     private static final String testingTable = "t_testing";
     private static final String CREATE_SQL =
-            "CREATE TABLE if not exists "
+            "CREATE TABLE "
                     + testingTable
                     + " ("
                     + "id bigint NOT NULL, "
                     + "ts bigint NOT NULL, "
                     + "PRIMARY KEY (id))";
-    private static final ContinuousEnumerationSettings CONTINUOUS_SETTINGS =
-            new ContinuousEnumerationSettings(Duration.ofMillis(10L), Duration.ofSeconds(1L));
+    private static final ContinuousUnBoundingSettings CONTINUOUS_SETTINGS =
+            new ContinuousUnBoundingSettings(Duration.ofMillis(10L), Duration.ofSeconds(1L));
     private static final ResultExtractor<TestEntry> EXTRACTOR =
             resultSet -> new TestEntry(resultSet.getLong("id"), resultSet.getLong("ts"));
     private static final List<TestEntry> testEntries = new ArrayList<>(TESTING_ENTRIES_SIZE);
@@ -97,7 +98,6 @@ class JdbcSourceStreamRelatedITCase implements MySqlTestBase, JdbcITCaseBase {
     void initData() {
         testEntries.clear();
         quickExecutionSQL(CREATE_SQL);
-        quickExecutionSQL("delete from " + testingTable);
         generateTestEntries();
         String insertSQL = generateInsertSQL();
         quickExecutionSQL(insertSQL);
@@ -112,14 +112,20 @@ void initData() {
                         .setDBUrl(getMetadata().getJdbcUrl())
                         .setUsername(getMetadata().getUsername())
                         .setPassword(getMetadata().getPassword())
-                        .setContinuousEnumerationSettings(CONTINUOUS_SETTINGS)
+                        .setContinuousUnBoundingSettings(CONTINUOUS_SETTINGS)
                         .setJdbcParameterValuesProvider(slideTimingParamsProvider)
                         .setDriverName(getMetadata().getDriverClass())
                         .setResultExtractor(EXTRACTOR);
 
         collectedRecords = new ConcurrentLinkedDeque<>();
     }
 
+    @AfterEach
+    void clearData() {
+        quickExecutionSQL("delete from " + testingTable);
+        quickExecutionSQL("DROP TABLE " + testingTable);
+    }
+
     @ParameterizedTest
     @EnumSource(DeliveryGuarantee.class)
     void testForNormalCaseWithoutFailure(