[FLINK-38733] Add new SplitterEnumerator on JdbcSource

Author: eskabetxe

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

@@ -87,7 +87,7 @@ public void start() {
         } else {
             context.callAsync(
                     () ->
-                            splitterEnumerator.hasFinishSplits()
+                            splitterEnumerator.isAllSplitsFinished()
                                     ? Collections.emptyList()
                                     : splitterEnumerator.enumerateSplits(),
                     (List<JdbcSourceSplit> splits, Throwable error) ->

flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/PreparedSplitterEnumerator.java

@@ -44,7 +44,7 @@ public class PreparedSplitterEnumerator implements SplitterEnumerator {
 
     private final String sqlTemplate;
     private final Serializable[][] sqlParameters;
-    private Boolean finished;
+    private boolean finished;
 
     protected PreparedSplitterEnumerator(String sqlTemplate, Serializable[][] sqlParameters) {
         this.sqlTemplate = Preconditions.checkNotNull(sqlTemplate);
@@ -68,7 +68,7 @@ public void start(JdbcConnectionProvider connectionProvider) {}
     public void close() {}
 
     @Override
-    public boolean hasFinishSplits() {
+    public boolean isAllSplitsFinished() {
         return this.finished;
     }
 

flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/SlideTimingSplitterEnumerator.java

@@ -52,7 +52,7 @@ public static SlideTimingSplitterEnumerator of(
     }
 
     @Override
-    public boolean hasFinishSplits() {
+    public boolean isAllSplitsFinished() {
         return false;
     }
 

flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/splitter/SplitterEnumerator.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter;
 
+import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.connector.jdbc.core.datastream.source.split.JdbcSourceSplit;
 import org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider;
 
@@ -29,30 +30,67 @@
 import java.util.function.Supplier;
 
 /** Interface for jdbc sql split enumerator. */
+@PublicEvolving
 public interface SplitterEnumerator extends AutoCloseable, Serializable {
 
+    /**
+     * Start the enumerator.
+     *
+     * @param connectionProvider The JDBC connection provider.
+     */
     void start(JdbcConnectionProvider connectionProvider);
 
+    /** Close the enumerator. */
     void close();
 
-    boolean hasFinishSplits();
+    /** All splits have been enumerated. */
+    boolean isAllSplitsFinished();
 
+    /** Enumerate the JDBC splits. */
     List<JdbcSourceSplit> enumerateSplits();
 
+    /**
+     * Enumerate the JDBC splits.
+     *
+     * @param splitGettable If the next batch splits are gettable.
+     * @return The result splits generated by the split enumerator.
+     */
     default List<JdbcSourceSplit> enumerateSplits(@Nonnull Supplier<Boolean> splitGettable) {
         return enumerateSplits(splitGettable.get());
     }
 
-    default List<JdbcSourceSplit> enumerateSplits(Boolean splitGettable) {
+    /**
+     * Enumerate the JDBC splits.
+     *
+     * @param splitGettable If the next batch splits are gettable.
+     * @return The result splits generated by the split enumerator.
+     */
+    default List<JdbcSourceSplit> enumerateSplits(boolean splitGettable) {
         if (!splitGettable) {
             return Collections.emptyList();
         }
         return enumerateSplits();
     }
 
+    /**
+     * Get lineage queries for splits.
+     *
+     * @return The lineage queries.
+     */
     List<String> lineageQueries();
 
+    /**
+     * Get the serializable state of the enumerator.
+     *
+     * @return The serializable state.
+     */
     Serializable serializableState();
 
+    /**
+     * Restore the enumerator state from the given serializable state.
+     *
+     * @param state The serializable state.
+     * @return The restored SplitterEnumerator.
+     */
     SplitterEnumerator restoreState(Serializable state);
 }

flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/enumerator/JdbcSourceEnumeratorTest.java

@@ -114,7 +114,7 @@ public void start(JdbcConnectionProvider connectionProvider) {}
                     public void close() {}
 
                     @Override
-                    public boolean hasFinishSplits() {
+                    public boolean isAllSplitsFinished() {
                         return false;
                     }