Resolve conflicts

Author: yuxiqian

flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/SchemaOperatorTranslator.java

@@ -17,11 +17,9 @@
 
 package org.apache.flink.cdc.composer.flink.translator;
 
-import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.cdc.common.annotation.Internal;
 import org.apache.flink.cdc.common.event.Event;
 import org.apache.flink.cdc.common.event.SchemaChangeEvent;
-import org.apache.flink.cdc.common.event.TableId;
 import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
 import org.apache.flink.cdc.common.route.RouteRule;
 import org.apache.flink.cdc.common.sink.MetadataApplier;

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/SchemaOperator.java

@@ -248,7 +248,7 @@ public void processElement(StreamRecord<Event> streamRecord)
     }
 
     private void processSchemaChangeEvents(SchemaChangeEvent event)
-            throws InterruptedException, TimeoutException {
+            throws InterruptedException, TimeoutException, ExecutionException {
         TableId tableId = event.tableId();
         LOG.info("Table {} received SchemaChangeEvent and start to be blocked.", tableId);
         handleSchemaChangeEvent(tableId, event);
@@ -258,7 +258,8 @@ private void processSchemaChangeEvents(SchemaChangeEvent event)
 
         List<TableId> optionalRoutedTable = getRoutedTables(tableId);
         if (!optionalRoutedTable.isEmpty()) {
-            getRoutedTables(tableId)
+            tableIdMappingCache
+                    .get(tableId)
                     .forEach(routed -> evolvedSchema.put(routed, getLatestEvolvedSchema(routed)));
         } else {
             evolvedSchema.put(tableId, getLatestEvolvedSchema(tableId));

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/SchemaOperatorFactory.java

@@ -19,7 +19,6 @@
 
 import org.apache.flink.cdc.common.annotation.Internal;
 import org.apache.flink.cdc.common.event.Event;
-import org.apache.flink.cdc.common.event.TableId;
 import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
 import org.apache.flink.cdc.common.route.RouteRule;
 import org.apache.flink.cdc.common.sink.MetadataApplier;

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaRegistry.java

@@ -19,7 +19,6 @@
 
 import org.apache.flink.cdc.common.event.TableId;
 import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
-import org.apache.flink.cdc.common.schema.Selectors;
 import org.apache.flink.cdc.common.route.RouteRule;
 import org.apache.flink.cdc.common.sink.MetadataApplier;
 import org.apache.flink.cdc.runtime.operators.schema.SchemaOperator;
@@ -269,7 +268,7 @@ public void resetToCheckpoint(long checkpointId, @Nullable byte[] checkpointData
                                         metadataApplier,
                                         schemaManager,
                                         schemaDerivation,
-                                        schemaManager.getBehavior());
+                                        schemaChangeBehavior);
                         break;
                     }
                 default:

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaRegistryProvider.java

@@ -17,18 +17,14 @@
 
 package org.apache.flink.cdc.runtime.operators.schema.coordinator;
 
-import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.cdc.common.annotation.Internal;
-import org.apache.flink.cdc.common.event.TableId;
 import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
-import org.apache.flink.cdc.common.schema.Selectors;
 import org.apache.flink.cdc.common.route.RouteRule;
 import org.apache.flink.cdc.common.sink.MetadataApplier;
 import org.apache.flink.runtime.jobgraph.OperatorID;
 import org.apache.flink.runtime.operators.coordination.OperatorCoordinator;
 
 import java.util.List;
-import java.util.stream.Collectors;
 
 /** Provider of {@link SchemaRegistry}. */
 @Internal
@@ -61,7 +57,7 @@ public OperatorID getOperatorId() {
 
     @Override
     public OperatorCoordinator create(OperatorCoordinator.Context context) throws Exception {
-        return new SchemaRegistry(operatorName, context, metadataApplier, routingRules);
-
+        return new SchemaRegistry(
+                operatorName, context, metadataApplier, routingRules, schemaChangeBehavior);
     }
 }