[FLINK-38733] Clean Generics on source enumerator

Author: eskabetxe

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

@@ -71,7 +71,7 @@ public class JdbcSource<OUT>
     private final @Nullable ContinuousUnBoundingSettings continuousUnBoundingSettings;
 
     private final Configuration configuration;
-    private final JdbcSqlSplitEnumeratorBase.Provider<JdbcSourceSplit> sqlSplitEnumeratorProvider;
+    private final JdbcSqlSplitEnumeratorBase.Provider sqlSplitEnumeratorProvider;
 
     protected JdbcConnectionProvider connectionProvider;
     private final ResultExtractor<OUT> resultExtractor;
@@ -80,7 +80,7 @@ public class JdbcSource<OUT>
     JdbcSource(
             Configuration configuration,
             JdbcConnectionProvider connectionProvider,
-            JdbcSqlSplitEnumeratorBase.Provider<JdbcSourceSplit> sqlSplitEnumeratorProvider,
+            JdbcSqlSplitEnumeratorBase.Provider sqlSplitEnumeratorProvider,
             ResultExtractor<OUT> resultExtractor,
             TypeInformation<OUT> typeInformation,
             @Nullable DeliveryGuarantee deliveryGuarantee,
@@ -167,7 +167,7 @@ public static <OUT> JdbcSourceBuilder<OUT> builder() {
     // ---- Visible for testing methods. ---
 
     @VisibleForTesting
-    public JdbcSqlSplitEnumeratorBase.Provider<JdbcSourceSplit> getSqlSplitEnumeratorProvider() {
+    public JdbcSqlSplitEnumeratorBase.Provider getSqlSplitEnumeratorProvider() {
         return sqlSplitEnumeratorProvider;
     }
 

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

@@ -50,12 +50,12 @@ public class JdbcSourceEnumerator
     private final Boundedness boundedness;
     private final LinkedHashMap<Integer, String> readersAwaitingSplit;
     private final List<JdbcSourceSplit> unassigned;
-    private final JdbcSqlSplitEnumeratorBase<JdbcSourceSplit> sqlSplitEnumerator;
+    private final JdbcSqlSplitEnumeratorBase sqlSplitEnumerator;
     private final @Nullable ContinuousUnBoundingSettings continuousUnBoundingSettings;
 
     public JdbcSourceEnumerator(
             SplitEnumeratorContext<JdbcSourceSplit> context,
-            JdbcSqlSplitEnumeratorBase<JdbcSourceSplit> sqlSplitEnumerator,
+            JdbcSqlSplitEnumeratorBase sqlSplitEnumerator,
             ContinuousUnBoundingSettings continuousUnBoundingSettings,
             List<JdbcSourceSplit> unassigned) {
         this.context = Preconditions.checkNotNull(context);

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

@@ -29,13 +29,9 @@
 import java.util.List;
 import java.util.function.Supplier;
 
-/**
- * Base class for jdbc sql split enumerator.
- *
- * @param <SplitT> JDBC split type.
- */
+/** Base class for jdbc sql split enumerator. */
 @PublicEvolving
-public abstract class JdbcSqlSplitEnumeratorBase<SplitT> implements AutoCloseable, Serializable {
+public abstract class JdbcSqlSplitEnumeratorBase implements AutoCloseable, Serializable {
     private final char[] currentId = "0000000000".toCharArray();
 
     protected @Nullable Serializable optionalSqlSplitEnumeratorState;
@@ -73,29 +69,24 @@ private static void incrementCharArrayByOne(char[] array, int pos) {
     public abstract List<JdbcSourceSplit> enumerateSplits(@Nonnull Supplier<Boolean> splitGettable)
             throws IOException;
 
-    /**
-     * A provider to create or restore a JDBC sql splits enumerator.
-     *
-     * @param <SplitT> Split type.
-     */
+    /** A provider to create or restore a JDBC sql splits enumerator. */
     @PublicEvolving
-    public interface Provider<SplitT> extends Serializable {
+    public interface Provider extends Serializable {
 
         /**
          * Called when init the provider without state.
          *
          * @return An instance of {@link JdbcSqlSplitEnumeratorBase}.
          */
-        JdbcSqlSplitEnumeratorBase<SplitT> create();
+        JdbcSqlSplitEnumeratorBase create();
 
         /**
          * Called when restore the provider without state.
          *
          * @param optionalSqlSplitEnumeratorState The state defined by users.
          * @return An instance of {@link JdbcSqlSplitEnumeratorBase}.
          */
-        JdbcSqlSplitEnumeratorBase<SplitT> restore(
-                @Nullable Serializable optionalSqlSplitEnumeratorState);
+        JdbcSqlSplitEnumeratorBase restore(@Nullable Serializable optionalSqlSplitEnumeratorState);
     }
 
     public void open() {}

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

@@ -38,7 +38,7 @@
 import java.util.function.Supplier;
 
 /** A split enumerator based on sql-parameters grains. */
-public final class SqlTemplateSplitEnumerator extends JdbcSqlSplitEnumeratorBase<JdbcSourceSplit> {
+public final class SqlTemplateSplitEnumerator extends JdbcSqlSplitEnumeratorBase {
 
     public static final Logger LOG = LoggerFactory.getLogger(SqlTemplateSplitEnumerator.class);
 
@@ -106,7 +106,7 @@ public JdbcParameterValuesProvider getParameterValuesProvider() {
 
     /** The {@link TemplateSqlSplitEnumeratorProvider} for {@link SqlTemplateSplitEnumerator}. */
     public static class TemplateSqlSplitEnumeratorProvider
-            implements JdbcSqlSplitEnumeratorBase.Provider<JdbcSourceSplit> {
+            implements JdbcSqlSplitEnumeratorBase.Provider {
 
         private String sqlTemplate;
 
@@ -135,13 +135,13 @@ public TemplateSqlSplitEnumeratorProvider setOptionalSqlSplitEnumeratorState(
         }
 
         @Override
-        public JdbcSqlSplitEnumeratorBase<JdbcSourceSplit> create() {
+        public JdbcSqlSplitEnumeratorBase create() {
             return new SqlTemplateSplitEnumerator(
                     this.optionalSqlSplitEnumeratorState, sqlTemplate, parameterValuesProvider);
         }
 
         @Override
-        public JdbcSqlSplitEnumeratorBase<JdbcSourceSplit> restore(
+        public JdbcSqlSplitEnumeratorBase restore(
                 @Nullable Serializable optionalSqlSplitEnumeratorState) {
             return new SqlTemplateSplitEnumerator(
                     optionalSqlSplitEnumeratorState, sqlTemplate, parameterValuesProvider);

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

@@ -108,7 +108,7 @@ private static JdbcSourceEnumerator createEnumerator(
 
         return new JdbcSourceEnumerator(
                 context,
-                new JdbcSqlSplitEnumeratorBase<JdbcSourceSplit>(null) {
+                new JdbcSqlSplitEnumeratorBase(null) {
                     @Override
                     public @Nonnull List<JdbcSourceSplit> enumerateSplits(
                             @Nonnull Supplier<Boolean> splitGettable) throws IOException {