Skip to content

Commit 97e484c

Browse files
author
guoxuanlin
committed
[oceanus] support pipeline
1 parent 3525b53 commit 97e484c

5 files changed

Lines changed: 142 additions & 1 deletion

File tree

‎flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/CliExecutor.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -108,7 +108,8 @@ public PipelineExecution.ExecutionInfo deployWithNoOpComposer() throws Exception
108108
// The main class for running application mode
109109
public static void main(String[] args) throws Exception {
110110
PipelineDefinitionParser pipelineDefinitionParser = new YamlPipelineDefinitionParser();
111-
PipelineDef pipelineDef = pipelineDefinitionParser.parse(args[0], new Configuration());
111+
PipelineDef pipelineDef =
112+
pipelineDefinitionParser.parse(new Path(args[0]), new Configuration());
112113
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
113114
FlinkPipelineComposer flinkPipelineComposer =
114115
FlinkPipelineComposer.ofApplicationCluster(env);

‎flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/pom.xml‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ limitations under the License.
2929
<artifactId>flink-cdc-pipeline-connector-mysql</artifactId>
3030

3131
<properties>
32+
<mysql.connector.version>8.0.27</mysql.connector.version>
3233
</properties>
3334

3435
<dependencies>
@@ -38,6 +39,15 @@ limitations under the License.
3839
<version>${project.version}</version>
3940
</dependency>
4041

42+
<!-- MySQL JDBC driver, required by Debezium's MySqlConnectorConfig at runtime.
43+
In standard pipeline deployment, users download this JAR separately.
44+
For Application Mode, it must be shaded into the connector JAR. -->
45+
<dependency>
46+
<groupId>mysql</groupId>
47+
<artifactId>mysql-connector-java</artifactId>
48+
<version>${mysql.connector.version}</version>
49+
</dependency>
50+
4151
<dependency>
4252
<groupId>org.apache.flink</groupId>
4353
<artifactId>flink-connector-mysql-cdc</artifactId>
@@ -198,6 +208,8 @@ limitations under the License.
198208
<include>com.zaxxer:HikariCP</include>
199209
<!-- Include fixed version 30.1.1-jre-14.0 of flink shaded guava -->
200210
<include>org.apache.flink:flink-shaded-guava</include>
211+
<!-- MySQL JDBC driver for Application Mode -->
212+
<include>mysql:mysql-connector-java</include>
201213
</includes>
202214
</artifactSet>
203215
<filters>
@@ -249,6 +261,12 @@ limitations under the License.
249261
org.apache.flink.cdc.connectors.shaded.com.zaxxer
250262
</shadedPattern>
251263
</relocation>
264+
<relocation>
265+
<pattern>com.mysql</pattern>
266+
<shadedPattern>
267+
org.apache.flink.cdc.connectors.shaded.com.mysql
268+
</shadedPattern>
269+
</relocation>
252270
</relocations>
253271
</configuration>
254272
</execution>

‎flink-cdc-entry/pom.xml‎

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<!--
3+
Licensed to the Apache Software Foundation (ASF) under one or more
4+
contributor license agreements. See the NOTICE file distributed with
5+
this work for additional information regarding copyright ownership.
6+
The ASF licenses this file to You under the Apache License, Version 2.0
7+
(the "License"); you may not use this file except in compliance with
8+
the License. You may obtain a copy of the License at
9+
10+
http://www.apache.org/licenses/LICENSE-2.0
11+
12+
Unless required by applicable law or agreed to in writing, software
13+
distributed under the License is distributed on an "AS IS" BASIS,
14+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
See the License for the specific language governing permissions and
16+
limitations under the License.
17+
-->
18+
<project xmlns="http://maven.apache.org/POM/4.0.0"
19+
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
20+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
21+
<parent>
22+
<artifactId>flink-cdc-parent</artifactId>
23+
<groupId>org.apache.flink</groupId>
24+
<version>${revision}</version>
25+
</parent>
26+
<modelVersion>4.0.0</modelVersion>
27+
28+
<artifactId>flink-cdc-entry</artifactId>
29+
30+
<dependencies>
31+
<!-- No CDC dependencies needed. This module uses pure reflection to delegate
32+
to CliExecutor.main() which resides in flink-cdc-dist.jar (loaded via AppClassLoader). -->
33+
</dependencies>
34+
35+
<build>
36+
<plugins>
37+
<plugin>
38+
<groupId>org.apache.maven.plugins</groupId>
39+
<artifactId>maven-jar-plugin</artifactId>
40+
<configuration>
41+
<archive>
42+
<manifest>
43+
<mainClass>org.apache.flink.cdc.entry.CdcEntryPoint</mainClass>
44+
</manifest>
45+
</archive>
46+
</configuration>
47+
</plugin>
48+
<!-- Skip shade plugin for this module since we have no dependencies to shade -->
49+
<plugin>
50+
<groupId>org.apache.maven.plugins</groupId>
51+
<artifactId>maven-shade-plugin</artifactId>
52+
<executions>
53+
<execution>
54+
<id>shade-flink</id>
55+
<phase>none</phase>
56+
</execution>
57+
</executions>
58+
</plugin>
59+
</plugins>
60+
</build>
61+
62+
</project>

‎flink-cdc-entry/src/main/java/org/apache/flink/cdc/entry/CdcEntryPoint.java‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.flink.cdc.entry;
19+
20+
import java.lang.reflect.Method;
21+
22+
/**
23+
* A minimal entry point JAR for Oceanus platform deployment.
24+
*
25+
* <p>This class is designed to be placed in {@code pipeline.jars} (which is loaded by
26+
* FlinkUserCodeClassLoader), while the actual Flink CDC dist JAR and connector JARs are placed in
27+
* {@code pipeline.classpaths} (which is loaded by the parent AppClassLoader).
28+
*
29+
* <p>By doing this, all CDC classes (flink-cdc-dist + connectors) reside in the same ClassLoader
30+
* (AppClassLoader), which resolves the ClassNotFoundException issue where connector classes cannot
31+
* find CDC common classes like EventDeserializer.
32+
*
33+
* <p>This class has ZERO dependencies on any CDC module. It uses pure Java reflection to invoke
34+
* {@code CliExecutor.main()} from the parent ClassLoader.
35+
*/
36+
public class CdcEntryPoint {
37+
38+
private static final String CLI_EXECUTOR_CLASS = "org.apache.flink.cdc.cli.CliExecutor";
39+
40+
public static void main(String[] args) throws Exception {
41+
// CliExecutor is in flink-cdc-dist.jar which should be in pipeline.classpaths
42+
// (AppClassLoader). Through parent delegation, we can load it from the child
43+
// FlinkUserCodeClassLoader.
44+
Class<?> cliExecutorClass;
45+
try {
46+
cliExecutorClass = Class.forName(CLI_EXECUTOR_CLASS);
47+
} catch (ClassNotFoundException e) {
48+
throw new RuntimeException(
49+
"Cannot find "
50+
+ CLI_EXECUTOR_CLASS
51+
+ ". "
52+
+ "Please make sure flink-cdc-dist JAR is placed in pipeline.classpaths.",
53+
e);
54+
}
55+
56+
Method mainMethod = cliExecutorClass.getMethod("main", String[].class);
57+
mainMethod.invoke(null, (Object) args);
58+
}
59+
}

‎pom.xml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@ limitations under the License.
4444
<module>flink-cdc-e2e-tests</module>
4545
<module>flink-cdc-pipeline-udf-examples</module>
4646
<module>flink-cdc-pipeline-model</module>
47+
<module>flink-cdc-entry</module>
4748
</modules>
4849

4950
<licenses>

0 commit comments

Comments
 (0)