[Hotfix][Oracle-CDC] Fix state recovery error when switching a single table to multiple tables (#6211)

hailin0 · web-flow · commit 74cfe1995f1a · 2024-01-16T08:56:28.000+08:00
diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSourceFactory.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSourceFactory.java
@@ -103,7 +103,7 @@ TableSource<T, SplitT, StateT> createSource(TableSourceFactoryContext context) {
                     CatalogTableUtil.getCatalogTables(
                             context.getOptions(), context.getClassLoader());
             SeaTunnelDataType<SeaTunnelRow> dataType =
-                    CatalogTableUtil.convertToDataType(catalogTables);
+                    CatalogTableUtil.convertToMultipleRowType(catalogTables);
             return new OracleIncrementalSource(context.getOptions(), dataType, catalogTables);
         };
     }

Original file line number	Diff line number	Diff line change
`@@ -103,7 +103,7 @@ TableSource<T, SplitT, StateT> createSource(TableSourceFactoryContext context) {`
`103`	`103`	`CatalogTableUtil.getCatalogTables(`
`104`	`104`	`context.getOptions(), context.getClassLoader());`
`105`	`105`	`SeaTunnelDataType<SeaTunnelRow> dataType =`
`106`		`- CatalogTableUtil.convertToDataType(catalogTables);`
	`106`	`+ CatalogTableUtil.convertToMultipleRowType(catalogTables);`
`107`	`107`	`return new OracleIncrementalSource(context.getOptions(), dataType, catalogTables);`
`108`	`108`	`};`
`109`	`109`	`}`