[FLINK-38204][pipeline-connector][maxcompute] Use getLatestEvolvedSchema to get Schema in SessionManageOperator in case of using route. apache#4094

proletarians · wuzexian · web-flow · commit 16fa17e06475 · 2025-08-14T12:51:57.000+08:00
Co-authored-by: wuzexian &lt;shanqing.wzx@alibaba-inc.com&gt;
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/coordinator/SessionManageOperator.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/coordinator/SessionManageOperator.java
@@ -194,7 +194,7 @@ public void processElement(StreamRecord<Event> element) throws Exception {
     }
 
     private void emitLatestSchema(TableId tableId) throws Exception {
-        Optional<Schema> schema = schemaEvolutionClient.getLatestOriginalSchema(tableId);
+        Optional<Schema> schema = schemaEvolutionClient.getLatestEvolvedSchema(tableId);
         if (schema.isPresent()) {
             Schema latestSchema = schema.get();
             schemaMaps.put(tableId, latestSchema);

Original file line number	Diff line number	Diff line change
`@@ -194,7 +194,7 @@ public void processElement(StreamRecord<Event> element) throws Exception {`
`194`	`194`	`}`
`195`	`195`
`196`	`196`	`private void emitLatestSchema(TableId tableId) throws Exception {`
`197`		`- Optional<Schema> schema = schemaEvolutionClient.getLatestOriginalSchema(tableId);`
	`197`	`+ Optional<Schema> schema = schemaEvolutionClient.getLatestEvolvedSchema(tableId);`
`198`	`198`	`if (schema.isPresent()) {`
`199`	`199`	`Schema latestSchema = schema.get();`
`200`	`200`	`schemaMaps.put(tableId, latestSchema);`