[cdc-composer] Add routing logic into the Flink composer

flink-cdc-composer/src/main/java/com/ververica/cdc/composer/flink/FlinkPipelineComposer.java

@@ -33,6 +33,7 @@ import com.ververica.cdc.composer.flink.coordination.OperatorIDGenerator;
 import com.ververica.cdc.composer.flink.translator.DataSinkTranslator;
 import com.ververica.cdc.composer.flink.translator.DataSourceTranslator;
 import com.ververica.cdc.composer.flink.translator.PartitioningTranslator;
+import com.ververica.cdc.composer.flink.translator.RouteTranslator;
 import com.ververica.cdc.composer.flink.translator.SchemaOperatorTranslator;
 import com.ververica.cdc.composer.utils.FactoryDiscoveryUtils;
 
@@ -72,6 +73,10 @@ public class FlinkPipelineComposer implements PipelineComposer {
         DataStream<Event> stream =
                 sourceTranslator.translate(pipelineDef.getSource(), env, parallelism);
 
+        // Route
+        RouteTranslator routeTranslator = new RouteTranslator();
+        stream = routeTranslator.translate(stream, pipelineDef.getRoute());
+
         // Create sink in advance as schema operator requires MetadataApplier
         DataSink dataSink = createDataSink(pipelineDef.getSink());
 

flink-cdc-composer/src/main/java/com/ververica/cdc/composer/flink/translator/RouteTranslator.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2023 Ververica Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.ververica.cdc.composer.flink.translator;
+
+import org.apache.flink.streaming.api.datastream.DataStream;
+
+import com.ververica.cdc.common.event.Event;
+import com.ververica.cdc.common.event.TableId;
+import com.ververica.cdc.composer.definition.RouteDef;
+import com.ververica.cdc.runtime.operators.route.RouteFunction;
+import com.ververica.cdc.runtime.typeutils.EventTypeInfo;
+
+import java.util.List;
+
+/** Translator for router. */
+public class RouteTranslator {
+    public DataStream<Event> translate(DataStream<Event> input, List<RouteDef> routes) {
+        if (routes.isEmpty()) {
+            return input;
+        }
+        RouteFunction.Builder routeFunctionBuilder = RouteFunction.newBuilder();
+        for (RouteDef route : routes) {
+            routeFunctionBuilder.addRoute(
+                    route.getSourceTable(), TableId.parse(route.getSinkTable()));
+        }
+        return input.map(routeFunctionBuilder.build(), new EventTypeInfo());
+    }
+}