Skip to content

Commit 3d73038

Browse files
loserwang1024loserwang1024
committed
[FLINK-37958][cdc-sink-connectors] Cdc yaml sink pipeline support meta applier
1 parent 246bddd commit 3d73038

28 files changed

Lines changed: 2997 additions & 593 deletions

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

Lines changed: 38 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@ limitations under the License.
2222
<parent>
2323
<groupId>org.apache.flink</groupId>
2424
<artifactId>flink-cdc-pipeline-connectors</artifactId>
25-
<version>3.5-SNAPSHOT</version>
25+
<version>${revision}</version>
2626
</parent>
2727

2828
<artifactId>flink-cdc-pipeline-connector-fluss</artifactId>
@@ -34,16 +34,16 @@ limitations under the License.
3434

3535
<properties>
3636
<fluss.version>0.7.0</fluss.version>
37-
3837
</properties>
3938

4039
<dependencies>
4140
<dependency>
4241
<groupId>com.alibaba.fluss</groupId>
43-
<artifactId>fluss-flink-1.20</artifactId>
42+
<artifactId>fluss-client</artifactId>
4443
<version>${fluss.version}</version>
4544
</dependency>
4645

46+
4747
<!-- Test dependencies -->
4848
<dependency>
4949
<groupId>org.apache.flink</groupId>
@@ -78,19 +78,54 @@ limitations under the License.
7878
<version>${fluss.version}</version>
7979
<scope>test</scope>
8080
</dependency>
81+
<!-- In cdc yaml, we use our own sink. Thus, import fluss-fink just for test -->
8182
<dependency>
8283
<groupId>com.alibaba.fluss</groupId>
8384
<artifactId>fluss-flink-common</artifactId>
8485
<version>${fluss.version}</version>
8586
<type>test-jar</type>
8687
<scope>test</scope>
8788
</dependency>
89+
<dependency>
90+
<groupId>com.alibaba.fluss</groupId>
91+
<artifactId>fluss-flink-1.20</artifactId>
92+
<version>${fluss.version}</version>
93+
<scope>test</scope>
94+
</dependency>
8895
<dependency>
8996
<groupId>org.apache.flink</groupId>
9097
<artifactId>flink-table-planner_2.12</artifactId>
9198
<version>${flink.version}</version>
9299
<scope>test</scope>
93100
</dependency>
101+
94102
</dependencies>
95103

104+
<build>
105+
<plugins>
106+
<plugin>
107+
<groupId>org.apache.maven.plugins</groupId>
108+
<artifactId>maven-shade-plugin</artifactId>
109+
<version>${maven.shade.plugin.version}</version>
110+
<executions>
111+
<execution>
112+
<id>shade-flink</id>
113+
<phase>package</phase>
114+
<goals>
115+
<goal>shade</goal>
116+
</goals>
117+
<configuration>
118+
<shadeTestJar>false</shadeTestJar>
119+
<artifactSet>
120+
<includes>
121+
<include>com.alibaba.fluss:*</include>
122+
</includes>
123+
</artifactSet>
124+
</configuration>
125+
</execution>
126+
</executions>
127+
</plugin>
128+
</plugins>
129+
</build>
130+
96131
</project>

‎flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/factory/FlussDataSinkFactory.java‎

Lines changed: 16 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -20,32 +20,37 @@
2020
import org.apache.flink.cdc.common.configuration.ConfigOption;
2121
import org.apache.flink.cdc.common.factories.DataSinkFactory;
2222
import org.apache.flink.cdc.common.factories.FactoryHelper;
23-
import org.apache.flink.cdc.common.pipeline.PipelineOptions;
2423
import org.apache.flink.cdc.common.sink.DataSink;
2524
import org.apache.flink.cdc.connectors.fluss.sink.FlussDataSink;
2625

2726
import com.alibaba.fluss.config.Configuration;
2827

29-
import java.time.ZoneId;
3028
import java.util.HashMap;
3129
import java.util.HashSet;
30+
import java.util.List;
3231
import java.util.Map;
33-
import java.util.Objects;
3432
import java.util.Set;
3533

34+
import static org.apache.flink.cdc.connectors.fluss.sink.FlussConfigUtils.parseBucketKeys;
35+
import static org.apache.flink.cdc.connectors.fluss.sink.FlussConfigUtils.parseBucketNumber;
3636
import static org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.BOOTSTRAP_SERVERS;
37+
import static org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.BUCKET_KEY;
38+
import static org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.BUCKET_NUMBER;
3739
import static org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.CLIENT_ID;
3840
import static org.apache.flink.cdc.connectors.fluss.sink.FlussDataSinkOptions.PREFIX_FLUSS_PROPERTIES;
3941

42+
/** Factory for creating configured instances of {@link FlussDataSink}. */
4043
public class FlussDataSinkFactory implements DataSinkFactory {
4144
public static final String IDENTIFIER = "fluss";
4245

4346
@Override
4447
public DataSink createDataSink(Context context) {
4548
FactoryHelper.createFactoryHelper(this, context).validateExcept(PREFIX_FLUSS_PROPERTIES);
49+
org.apache.flink.cdc.common.configuration.Configuration factoryConfiguration =
50+
context.getFactoryConfiguration();
4651

4752
Map<String, String> flussOptions = new HashMap<>();
48-
context.getFactoryConfiguration()
53+
factoryConfiguration
4954
.toMap()
5055
.forEach(
5156
(key, value) -> {
@@ -56,17 +61,12 @@ public DataSink createDataSink(Context context) {
5661
flussOptions.put(key, value);
5762
}
5863
});
59-
ZoneId zoneId = ZoneId.systemDefault();
60-
if (!Objects.equals(
61-
context.getPipelineConfiguration().get(PipelineOptions.PIPELINE_LOCAL_TIME_ZONE),
62-
PipelineOptions.PIPELINE_LOCAL_TIME_ZONE.defaultValue())) {
63-
zoneId =
64-
ZoneId.of(
65-
context.getPipelineConfiguration()
66-
.get(PipelineOptions.PIPELINE_LOCAL_TIME_ZONE));
67-
}
6864
Configuration config = Configuration.fromMap(flussOptions);
69-
return new FlussDataSink(config, zoneId);
65+
Map<String, List<String>> bucketKeysMap =
66+
parseBucketKeys(factoryConfiguration.get(BUCKET_KEY));
67+
Map<String, Integer> bucketNumMap =
68+
parseBucketNumber(factoryConfiguration.get(BUCKET_NUMBER));
69+
return new FlussDataSink(config, bucketKeysMap, bucketNumMap);
7070
}
7171

7272
@Override
@@ -84,6 +84,8 @@ public Set<ConfigOption<?>> requiredOptions() {
8484
@Override
8585
public Set<ConfigOption<?>> optionalOptions() {
8686
Set<ConfigOption<?>> options = new HashSet<>();
87+
options.add(BUCKET_KEY);
88+
options.add(BUCKET_NUMBER);
8789
options.add(CLIENT_ID);
8890
return options;
8991
}

‎flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/CdcAsFlussRow.java‎

Lines changed: 153 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,153 @@
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.connectors.fluss.sink;
19+
20+
import org.apache.flink.cdc.common.data.DecimalData;
21+
import org.apache.flink.cdc.common.data.LocalZonedTimestampData;
22+
import org.apache.flink.cdc.common.data.RecordData;
23+
import org.apache.flink.cdc.common.data.TimestampData;
24+
25+
import com.alibaba.fluss.row.BinaryString;
26+
import com.alibaba.fluss.row.Decimal;
27+
import com.alibaba.fluss.row.InternalRow;
28+
import com.alibaba.fluss.row.TimestampLtz;
29+
import com.alibaba.fluss.row.TimestampNtz;
30+
31+
import java.util.Objects;
32+
33+
/** Wraps a CDC {@link RecordData} as a Fluss {@link InternalRow}. */
34+
public class CdcAsFlussRow implements InternalRow {
35+
36+
private final RecordData cdcRecord;
37+
38+
private CdcAsFlussRow(RecordData cdcRecord) {
39+
this.cdcRecord = cdcRecord;
40+
}
41+
42+
public static CdcAsFlussRow replace(RecordData cdcRecord) {
43+
return new CdcAsFlussRow(cdcRecord);
44+
}
45+
46+
@Override
47+
public int getFieldCount() {
48+
return cdcRecord.getArity();
49+
}
50+
51+
@Override
52+
public boolean isNullAt(int pos) {
53+
return cdcRecord.isNullAt(pos);
54+
}
55+
56+
@Override
57+
public boolean getBoolean(int pos) {
58+
return cdcRecord.getBoolean(pos);
59+
}
60+
61+
@Override
62+
public byte getByte(int pos) {
63+
return cdcRecord.getByte(pos);
64+
}
65+
66+
@Override
67+
public short getShort(int pos) {
68+
return cdcRecord.getShort(pos);
69+
}
70+
71+
@Override
72+
public int getInt(int pos) {
73+
return cdcRecord.getInt(pos);
74+
}
75+
76+
@Override
77+
public long getLong(int pos) {
78+
return cdcRecord.getLong(pos);
79+
}
80+
81+
@Override
82+
public float getFloat(int pos) {
83+
return cdcRecord.getFloat(pos);
84+
}
85+
86+
@Override
87+
public double getDouble(int pos) {
88+
return cdcRecord.getDouble(pos);
89+
}
90+
91+
@Override
92+
public BinaryString getChar(int pos, int length) {
93+
return BinaryString.fromBytes(cdcRecord.getString(pos).toBytes());
94+
}
95+
96+
@Override
97+
public BinaryString getString(int pos) {
98+
return BinaryString.fromBytes(cdcRecord.getString(pos).toBytes());
99+
}
100+
101+
@Override
102+
public Decimal getDecimal(int pos, int precision, int scale) {
103+
return fromFlinkDecimal(cdcRecord.getDecimal(pos, precision, scale));
104+
}
105+
106+
public static Decimal fromFlinkDecimal(DecimalData decimal) {
107+
if (decimal.isCompact()) {
108+
return Decimal.fromUnscaledLong(
109+
decimal.toUnscaledLong(), decimal.precision(), decimal.scale());
110+
} else {
111+
return Decimal.fromBigDecimal(
112+
decimal.toBigDecimal(), decimal.precision(), decimal.scale());
113+
}
114+
}
115+
116+
@Override
117+
public TimestampNtz getTimestampNtz(int pos, int precision) {
118+
TimestampData timestamp = cdcRecord.getTimestamp(pos, precision);
119+
return TimestampNtz.fromMillis(
120+
timestamp.getMillisecond(), timestamp.getNanoOfMillisecond());
121+
}
122+
123+
@Override
124+
public TimestampLtz getTimestampLtz(int pos, int precision) {
125+
// Fluss ltz maybe mapping from CDC ltz or
126+
LocalZonedTimestampData timestamp = cdcRecord.getLocalZonedTimestampData(pos, precision);
127+
return TimestampLtz.fromInstant(timestamp.toInstant());
128+
}
129+
130+
@Override
131+
public byte[] getBinary(int pos, int length) {
132+
return cdcRecord.getBinary(pos);
133+
}
134+
135+
@Override
136+
public byte[] getBytes(int pos) {
137+
return cdcRecord.getBinary(pos);
138+
}
139+
140+
@Override
141+
public boolean equals(Object o) {
142+
if (o == null || getClass() != o.getClass()) {
143+
return false;
144+
}
145+
CdcAsFlussRow that = (CdcAsFlussRow) o;
146+
return Objects.equals(cdcRecord, that.cdcRecord);
147+
}
148+
149+
@Override
150+
public int hashCode() {
151+
return Objects.hashCode(cdcRecord);
152+
}
153+
}

‎flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussConfigUtils.java‎

Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,81 @@
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.connectors.fluss.sink;
19+
20+
import javax.annotation.Nullable;
21+
22+
import java.util.Arrays;
23+
import java.util.HashMap;
24+
import java.util.List;
25+
import java.util.Map;
26+
27+
/** Utils for parse fluss yaml sink options. */
28+
public class FlussConfigUtils {
29+
public static Map<String, List<String>> parseBucketKeys(@Nullable String rawValue)
30+
throws IllegalArgumentException {
31+
Map<String, List<String>> result = new HashMap<>();
32+
if (rawValue == null || rawValue.isEmpty()) {
33+
return result;
34+
}
35+
36+
for (String tableBucketKeyStr : rawValue.split(";")) {
37+
if (tableBucketKeyStr.trim().isEmpty()) {
38+
continue;
39+
}
40+
41+
String[] tableAndBucketKeys = tableBucketKeyStr.trim().split(":", 2);
42+
if (tableAndBucketKeys.length != 2) {
43+
throw new IllegalArgumentException("Invalid bucket key configuration: " + rawValue);
44+
}
45+
46+
String table = tableAndBucketKeys[0].trim();
47+
List<String> keys = Arrays.asList(tableAndBucketKeys[1].trim().split(","));
48+
result.put(table, keys);
49+
}
50+
return result;
51+
}
52+
53+
public static Map<String, Integer> parseBucketNumber(@Nullable String rawValue)
54+
throws IllegalArgumentException {
55+
Map<String, Integer> result = new HashMap<>();
56+
if (rawValue == null || rawValue.isEmpty()) {
57+
return result;
58+
}
59+
60+
for (String tableBucketNumStr : rawValue.split(";")) {
61+
if (tableBucketNumStr.trim().isEmpty()) {
62+
continue;
63+
}
64+
String[] kv = tableBucketNumStr.trim().split(":", 2);
65+
if (kv.length != 2) {
66+
throw new IllegalArgumentException(
67+
"Invalid bucket number configuration: " + rawValue);
68+
}
69+
70+
String table = kv[0].trim();
71+
try {
72+
int value = Integer.parseInt(kv[1].trim());
73+
result.put(table, value);
74+
} catch (NumberFormatException ignored) {
75+
throw new IllegalArgumentException(
76+
"Invalid bucket number configuration: " + rawValue);
77+
}
78+
}
79+
return result;
80+
}
81+
}

0 commit comments

Comments
 (0)