address comment

Author: GOODBOY008

docs/content/connectors/sqlserver-cdc.md

@@ -138,19 +138,12 @@ Connector Options
       <td>String</td>
       <td>Database name of the SQLServer database to monitor.</td>
     </tr> 
-    <tr>
-      <td>schema-name</td>
-      <td>required</td>
-      <td style="word-wrap: break-word;">(none)</td>
-      <td>String</td>
-      <td>Schema name of the SQLServer database to monitor.</td>
-    </tr> 
     <tr>
       <td>table-name</td>
       <td>required</td>
       <td style="word-wrap: break-word;">(none)</td>
       <td>String</td>
-      <td>Table name of the SQLServer database to monitor.</td>
+      <td>Table name of the SQLServer database to monitor, e.g.: "db1.table1"</td>
     </tr>
     <tr>
       <td>port</td>

flink-connector-debezium/src/main/java/com/ververica/cdc/debezium/internal/FlinkDatabaseHistory.java

@@ -35,7 +35,7 @@
  * Inspired from {@link io.debezium.relational.history.MemoryDatabaseHistory} but we will store the
  * HistoryRecords in Flink's state for persistence.
  *
- * <p>Note: This is not a clean solution because we depend on a global variable and all the history
+ * <p>Note: This is not a clean solution because we depends on a global variable and all the history
  * records will be stored in state (grow infinitely). We may need to come up with a
  * FileSystemDatabaseHistory in the future to store history in HDFS.
  */

flink-connector-oracle-cdc/src/main/java/com/ververica/cdc/connectors/oracle/source/reader/fetch/OracleScanFetchTask.java

@@ -30,6 +30,7 @@
 import io.debezium.connector.oracle.OracleDatabaseSchema;
 import io.debezium.connector.oracle.OracleOffsetContext;
 import io.debezium.connector.oracle.OracleValueConverters;
+import io.debezium.connector.oracle.logminer.LogMinerOracleOffsetContextLoader;
 import io.debezium.heartbeat.Heartbeat;
 import io.debezium.pipeline.EventDispatcher;
 import io.debezium.pipeline.source.AbstractSnapshotChangeEventSource;
@@ -152,6 +153,12 @@ private StreamSplit createBackfillRedoLogSplit(
 
     private RedoLogSplitReadTask createBackfillRedoLogReadTask(
             StreamSplit backfillBinlogSplit, OracleSourceFetchTaskContext context) {
+        OracleConnectorConfig oracleConnectorConfig =
+                context.getSourceConfig().getDbzConnectorConfig();
+        final OffsetContext.Loader<OracleOffsetContext> loader =
+                new LogMinerOracleOffsetContextLoader(oracleConnectorConfig);
+        final OracleOffsetContext oracleOffsetContext =
+                loader.load(backfillBinlogSplit.getStartingOffset().getOffset());
         // we should only capture events for the current table,
         // otherwise, we may can't find corresponding schema
         Configuration dezConf =
@@ -287,13 +294,13 @@ protected SnapshottingTask getSnapshottingTask(OffsetContext previousOffset) {
         @Override
         protected SnapshotContext prepare(ChangeEventSourceContext changeEventSourceContext)
                 throws Exception {
-            return new OracleSnapshotContext();
+            return new MySqlSnapshotContext();
         }
 
-        private static class OracleSnapshotContext
+        private static class MySqlSnapshotContext
                 extends RelationalSnapshotChangeEventSource.RelationalSnapshotContext {
 
-            public OracleSnapshotContext() throws SQLException {
+            public MySqlSnapshotContext() throws SQLException {
                 super("");
             }
         }

flink-connector-oracle-cdc/src/test/java/com/ververica/cdc/connectors/oracle/source/OracleSourceITCase.java

@@ -129,7 +129,7 @@ private void testOracleParallelSource(
 
         String sourceDDL =
                 format(
-                        "CREATE TABLE customers ("
+                        "CREATE TABLE products ("
                                 + " ID INT NOT NULL,"
                                 + " NAME STRING,"
                                 + " ADDRESS STRING,"
@@ -183,7 +183,7 @@ private void testOracleParallelSource(
                     "+I[2000, user_21, Shanghai, 123567891234]"
                 };
         tEnv.executeSql(sourceDDL);
-        TableResult tableResult = tEnv.executeSql("select * from customers");
+        TableResult tableResult = tEnv.executeSql("select * from products");
         CloseableIterator<Row> iterator = tableResult.collect();
         JobID jobId = tableResult.getJobClient().get().getJobID();
         List<String> expectedSnapshotData = new ArrayList<>();

flink-connector-sqlserver-cdc/src/main/java/com/ververica/cdc/connectors/sqlserver/source/utils/SqlServerTypeUtils.java

@@ -65,6 +65,9 @@ private static DataType convertFromColumn(Column column) {
                 return DataTypes.DATE();
             case Types.TIMESTAMP:
             case Types.TIMESTAMP_WITH_TIMEZONE:
+                return column.length() >= 0
+                        ? DataTypes.TIMESTAMP(column.length())
+                        : DataTypes.TIMESTAMP();
             case Types.BOOLEAN:
                 return DataTypes.BOOLEAN();
             default:

flink-connector-sqlserver-cdc/src/main/java/io/debezium/connector/sqlserver/SqlServerStreamingChangeEventSource.java

@@ -46,8 +46,10 @@
 import java.util.stream.Collectors;
 
 /**
- * A {@link StreamingChangeEventSource} based on SQL Server change data capture functionality. A
- * main loop polls database DDL change and change data tables and turns them into change events.
+ * Copied from Debezium project(1.6.4.final) to add method {@link
+ * SqlServerStreamingChangeEventSource#afterHandleLsn(SqlServerOffsetContext)}. A {@link
+ * StreamingChangeEventSource} based on SQL Server change data capture functionality. A main loop
+ * polls database DDL change and change data tables and turns them into change events.
  *
  * <p>The connector uses CDC functionality of SQL Server that is implemented as as a process that
  * monitors source table and write changes from the table into the change table.