@PublicEvolving
public interface Sink<InputT> extends Serializable {
	... current methods

    @PublicEvolving
    public interface InitContext {
        ... current methods
	    ExecutionConfig getExecutionConfig();
	 	JobID getJobId();
		int getIndexOfThisSubtask():
    }
}

/** Adjust the JDBC connection provider. org.apache.flink.connector.jdbc.internal.connection.JdbcConnectionProvider */
public interface JdbcConnectionProvider {

    // The old methods declaration placeholder....
    
    // Add the new method declaration.
    /**
     * Get existing connection properties.
     *
     * @return existing connection properties
     */
    @Nullable
    default Properties getProperties() {
        return new Properties();
    }

	// Get or create sharding connections.
    default Collection<Connection> getOrCreateShardConnections(
            String remoteCluster, String remoteDataBase) throws SQLException {
        return Collections.emptyList();
    }

	// Get or create connection for one shard.
    Connection getOrCreateShardConnection(String url, String database) throws SQLException;

	// Get all shard urls.
    default List<String> getShardUrls(String remoteCluster) throws SQLException {
        return new ArrayList<>();
    }

	// Close all connections.
    void closeConnections();  

/** Adjust JDBC connection options. org.apache.flink.connector.jdbc.JdbcConnectionOptions */
public class JdbcConnectionOptions implements Serializable {
    
    // Old placeholders....

    // Add the extendProps.
    @Nullable protected final Properties extendProps;

    protected JdbcConnectionOptions(
            String url,
            @Nullable String driverName,
            @Nullable String username,
            @Nullable String password,
            int connectionCheckTimeoutSeconds,
            @Nullable Properties extendProps) {}
    // placeholders......

    /** Builder for {@link JdbcConnectionOptions}. */
    public static class JdbcConnectionOptionsBuilder {
        private String url;
        private String driverName;
        private String username;
        private String password;
        private int connectionCheckTimeoutSeconds = 60;
        protected Properties extendProps;
        // placeholders......

        public JdbcConnectionOptions build() {
            return new JdbcConnectionOptions(
                    url,
                    driverName,
                    username,
                    password,
                    connectionCheckTimeoutSeconds,
                    extendProps);
        }
    }
}

/**
 * A {@link SourceSplit} that represents a JDBC sqlTemplate with optional parameters.
 *
 * <p>The split has an offset, which defines the offset of the queries {@link java.sql.ResultSet}
 * for the split. The offset is the checkpointed position from a reader previously reading this
 * split. This position is typically zero when the split is assigned from the assigner to the
 * readers, and is positive when the readers checkpoint their state in a Jdbc source split.
 */
@Internal
public class JdbcSourceSplit implements SourceSplit, Serializable {     
    private final String id;     
    private final String sqlTemplate;     
    private final @Nullable Serializable[] parameters; 
    // The offset of ResultSet on the sql rendered by the split     
    private final int offset;     
    // .... 
}

/**
 * State of the JdbcSourceSplit.
 *
 * <p>The {@link JdbcSourceSplit} assigned to the reader or stored in the checkpoint points to the
 * position from where to start reading (after recovery), so the current offset need to always point
 * to the record after the last emitted record.
 */ 
@Internal
public class JdbcSourceSplitState<SplitT extends JdbcSourceSplit> implements Serializable {

    private final SplitT split;
    private int offset;

    // ...

    // methods ...
    public void setPosition(int offset)
    public SplitT toJdbcSourceSplit()
}

/** A serializer for the {@link JdbcSourceSplit}. */
@Internal
public class JdbcSourceSplitSerializer implements SimpleVersionedSerializer<JdbcSourceSplit> {
} 

/** The enumerator class for Jdbc source. */
@Internal
public class JdbcSourceEnumerator
        implements SplitEnumerator<JdbcSourceSplit, JdbcSourceEnumeratorState> {

    private final SplitEnumeratorContext<JdbcSourceSplit> context;
    private final Boundedness boundedness;

	// The map to store the info of split-request.
    private final LinkedHashMap<Integer, String> readersAwaitingSplit;

    private final List<JdbcSourceSplit> unassigned;

    // The jdbcSqlSplitAssigner will be responsible for determining whether it can respond to the allocation of a JdbcSourceSplit
    private final JdbcSqlSplitAssigner<JdbcSourceSplit> jdbcSqlSplitAssigner;

	// ....
}

/** The state of Jdbc source enumerator. */
@Internal
public class JdbcSourceEnumeratorState implements Serializable {

    private @Nonnull List<JdbcSourceSplit> remainingSplits;
    
    private final @Nullable Serializable optionalUserDefinedSplitAssignerState;
    
    // ... placeholders...
}

@Internal
public class JdbcSourceEnumStateSerializer
        implements SimpleVersionedSerializer<JdbcSourceEnumeratorState> {}

/**
 * The {@code OptionalUserDefinedState} is responsible for getting and setting the
 * user-defined-state.
 */
public interface OptionalUserDefinedState<T extends Serializable> extends Serializable {

    /**
     * Get the optional user-defined-state. The default value is null.
     *
     * @return {@code null} or an object implemented from {@link java.io.Serializable}.
     */
    @Nullable
    default T getOptionalUserDefinedState() {
        return null;
    }

    /**
     * Set the optional user-defined-state.
     *
     * @param optionalUserDefinedState user-defined-state
     */
    default void setOptionalUserDefinedState(@Nullable T optionalUserDefinedState) {}
}

/**
 * The {@code JdbcSqlSplitAssigner} is responsible for getting the split should be processed
 * next .
 */
@PublicEvolving
public interface JdbcSqlSplitAssigner<SplitT extends JdbcSourceSplit>
        extends Serializable, OptionalUserDefinedState<Serializable> {          /**
     * Called to open the assigner to acquire any resources, like threads or network connections.
     * Integer splitCapacity,
     */
    void open(ReadableConfig config);

    /**
     * Gets the next split.
     *
     * <p>When this method returns an empty {@code Optional}, then the set of splits is assumed to
     * be done and the source will finish once the readers finished their current splits.
     */
    Optional<JdbcSourceSplit> getNextSplit();

    /**
     * Adds a set of splits to this assigner. This happens for example when some split processing
     * failed and the splits need to be re-added.
     */
    void addSplits(Collection<JdbcSourceSplit> splits);

    /**
     * Notifies the listener that the checkpoint with the given {@code checkpointId} completed and
     * was committed.
     *
     * @see CheckpointListener#notifyCheckpointComplete(long)
     */
    void notifyCheckpointComplete(long checkpointId);

    default Serializable snapshotState(long checkpointId) {
        return getOptionalUserDefinedState();
    }

    /**
     * Called to close the assigner, in case it holds on to any resources, like threads or network
     * connections.
     */
    void close();

    char[] getCurrentSplitId();
 }

/** The source reader for Jdbc splits. */
@Internal
public class JdbcSourceReader<OUT>
        extends SingleThreadMultiplexSourceReaderBase<
                RecordAndOffset<OUT>, OUT, JdbcSourceSplit, JdbcSourceSplitState<JdbcSourceSplit>> {}

/** The {@link SplitReader} implementation for the jdbc source. */
@Internal
public class JdbcSourceSplitReader<T>
        implements SplitReader<RecordAndOffset<T>, JdbcSourceSplit>, ResultTypeQueryable<T> {

    private final Configuration config;
    protected TypeInformation<T> typeInformation;
    private final ResultExtractor<T> resultExtractor;
    private final DeliveryGuarantee deliveryGuarantee;
    protected JdbcConnectionProvider connectionProvider;
    private final SourceReaderContext context;
    
    //---------------------------------------------
    @Nullable private JdbcSourceSplit currentSplit;
    private final Queue<JdbcSourceSplit> splits;

    //-----------------------------------------------
    protected transient Connection connection;
    protected transient PreparedStatement statement;
    protected transient ResultSet resultSet;
    protected boolean hasNextRecordCurrentSplit;

    //-----------------------------------------------
    // config for ResultSet 
    private final int splitReaderFetchBatchSize;
    protected int resultSetType;
    protected int resultSetConcurrency;
    protected int resultSetFetchSize;
    // Boolean to distinguish between default value and explicitly set autoCommit mode.
    protected Boolean autoCommit;

    //... placeholder.
}

public class RecordAndOffset<E> {
    final E record;
    final int offset;
    // ------------------------------------------------------------------------
    // methods list
    public E getRecord()
    public int getOffset()
}

public interface ResultExtractor<T> extends Serializable {
    T extract(ResultSet resultSet) throws SQLException;
}

/**
 * The Source implementation of JdbcSource. Please use a {@link JdbcSourceBuilder} to construct a
 * {@link JdbcSource}. The following example shows how to create a JdbcSource emitting records of
 * <code>
 * Row</code> type.
 *
 * <pre>{@code
 * JdbcSource<Row> source = JdbcSource
 *     .<Row>builder()
 *     .setResultExtractor(...)
 *     .setJdbcSqlSplitAssignerProvider(...)
 *     .setConnectionProvider(...)
 *     .build();
 * }</pre>
 *
 * <p>See {@link JdbcSourceBuilder} for more details.
 *
 * @param <OUT> the output type of the source.
 */
@PublicEvolving
public class JdbcSource<OUT>
        implements Source<OUT, JdbcSourceSplit, JdbcSourceEnumeratorState>,
                ResultTypeQueryable<OUT> {

    private final Boundedness boundedness;
    private final TypeInformation<OUT> typeInformation;

    private final JdbcSqlSplitAssigner.Provider<JdbcSourceSplit> jdbcSqlSplitAssignerProvider;

    protected JdbcConnectionProvider connectionProvider;
    private final ResultExtractor<OUT> resultExtractor;
    private final DeliveryGuarantee deliveryGuarantee;
                    
    // methods placeholders.......
}

/**
 * The @builder class for {@link JdbcSource} to make it easier for the users to construct a {@link
 * JdbcSource}.
 */
@PublicEvolving
public class JdbcSourceBuilder<OUT> {

    private final Configuration configuration;
    private int splitReaderFetchBatchSize;
    private int resultSetType;
    private int resultSetConcurrency;
    private int resultSetFetchSize;
    // Boolean to distinguish between default value and explicitly set autoCommit mode.
    private Boolean autoCommit;
    private DeliveryGuarantee deliveryGuarantee;
    private TypeInformation<OUT> typeInformation;
    private final JdbcConnectionOptions.JdbcConnectionOptionsBuilder connOptionsBuilder;
    private ContinuousEnumerationSettings continuousEnumerationSettings;
    private JdbcSqlSplitAssigner.Provider<JdbcSourceSplit> jdbcSqlSplitAssignerProvider;
    private ResultExtractor<OUT> resultExtractor;
    private JdbcConnectionProvider connectionProvider;

    public JdbcSourceBuilder() {
        // placeholders ....    
    }
    
    // builder build methods placeholders ....  

    // ----------------methods list.-----------------------
    // ----- required methods to call -------------------------------
    public JdbcSourceBuilder<OUT> setResultExtractor(ResultExtractor<OUT> resultExtractor)
    public JdbcSourceBuilder<OUT> setJdbcSqlSplitAssignerProvider(JdbcSqlSplitAssigner.Provider<JdbcSourceSplit> jdbcSqlSplitAssignerProvider);
    public JdbcSourceBuilder<OUT> setUsername(String username);
    public JdbcSourceBuilder<OUT> setPassword(String password);
    public JdbcSourceBuilder<OUT> setDriverName(String driverName);
    public JdbcSourceBuilder<OUT> setDBUrl(String dbURL);
    public JdbcSource<OUT> build();

    // ------ Optional methods to call------------------------------

    public JdbcSourceBuilder<OUT> setDeliveryGuarantee(DeliveryGuarantee deliveryGuarantee);
    public JdbcSourceBuilder<OUT> setTypeInformation(TypeInformation<OUT> typeInformation);
    public JdbcSourceBuilder<OUT> setSplitReaderFetchBatchSize(int splitReaderFetchBatchSize);
    public JdbcSourceBuilder<OUT> setResultSetType(int resultSetType);
    public JdbcSourceBuilder<OUT> setResultSetConcurrency(int resultSetConcurrency);
    public JdbcSourceBuilder<OUT> setAutoCommit(boolean autoCommit);
    public JdbcSourceBuilder<OUT> setResultSetFetchSize(int resultSetFetchSize);
    public JdbcSourceBuilder<OUT> setConnectionProvider(JdbcConnectionProvider connectionProvider);
    public JdbcSourceBuilder<OUT> setJdbcConnectionOptionsBuilder(JdbcConnectionOptionsBuilder jdbcConnectionOptionsBuilder);
}

/** A {@link DynamicTableSource} for New JDBC Source. */
@Internal
public class JdbcDynamicTableSourceV2
        implements ScanTableSource,
                LookupTableSource,
                SupportsProjectionPushDown,
                SupportsLimitPushDown,
                SupportsWatermarkPushDown {

    private static final String JDBC_TRANSFORMATION = "jdbc";

    private final JdbcConnectorOptions options;
    private final JdbcReadOptions readOptions;
    private final JdbcLookupOptions lookupOptions;
    private DataType physicalRowDataType;
    private final String dialectName;
    private long limit = -1;
    private @Nullable WatermarkStrategy<RowData> watermarkStrategy;
    private final String tableIdentifier;

    private final ReadableConfig readableConfig;

    private final DynamicTableFactory.Context context;
                    
    // placeholders.....
    
    // need to use the JdbcSourceBuilder to implement the method.                
    @Override
    public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext);
}

/** A lookup function for {@link JdbcDynamicTableSourceV2}. */
@Internal
public class JdbcRowDataLookupFunction extends TableFunction<RowData> {

    private static final Logger LOG = LoggerFactory.getLogger(JdbcRowDataLookupFunction.class);
    private static final long serialVersionUID = 2L;

    private final String query;
    private final JdbcConnectionProvider connectionProvider;
    private final DataType[] keyTypes;
    private final String[] keyNames;
    private final long cacheMaxSize;
    private final long cacheExpireMs;
    private final int maxRetryTimes;
    private final boolean cacheMissingKey;
    private final JdbcDialect jdbcDialect;
    private final JdbcRowConverter jdbcRowConverter;
    private final JdbcRowConverter lookupKeyRowConverter;

    private transient FieldNamedPreparedStatement statement;
    private transient Cache<RowData, List<RowData>> cache;
}

// Introduce context & config fields
public abstract class AbstractJdbcCatalog extends org.apache.flink.table.catalog.AbtrastCatalog {

    // old fields lines placeholder.......
    
    // Introduced new fileds.
    protected final Context context;
    protected final ReadableConfig config;
    
    public AbstractJdbcCatalog(Context context, ReadableConfig config) {
        //....
    }
    
    // Introduced new methods.
    public Context getContext();
    public ReadableConfig getConfig();
    public Connection getConnection();
    
    // the old code lines placeholder...
}

// Introduce the 'createCatalog' method into package org.apache.flink.connector.jdbc.dialect.JdbcDialect
/**
 * Represents a dialect of SQL implemented by a particular JDBC system. Dialects should be stateless.
 *
 * @see JdbcDialectFactory
 */
@PublicEvolving
public interface JdbcDialect extends Serializable {
    
    // The old methods declaration placeholder.....

    // The new method declaration.
    /**
     * Constructs a catalog.
     *
     * @return A catalog based on the current SQL dialect.
     */
    AbstractJdbcCatalog createCatalog(CatalogFactory.Context context, ReadableConfig config);
}