You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "loserwang1024 (via GitHub)" <gi...@apache.org> on 2025/07/02 09:40:47 UTC
[PR] [FLINK-37958][cdc-sink-connectors] Flink CDC Sink connector support fluss [flink-cdc]
loserwang1024 opened a new pull request, #4053:
URL: https://github.com/apache/flink-cdc/pull/4053
A continue work for https://github.com/apache/flink-cdc/pull/4047 to close [FLINK-37958](https://issues.apache.org/jira/browse/FLINK-37958)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37958][cdc-sink-connectors] Flink CDC Sink connector support fluss [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4053:
URL: https://github.com/apache/flink-cdc/pull/4053#discussion_r2189376607
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussEventSerializationSchema.java:
##########
@@ -0,0 +1,184 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.flink.cdc.connectors.fluss.sink;
+
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.OperationType;
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.utils.Preconditions;
+import org.apache.flink.cdc.connectors.fluss.sink.v2.FlussEvent;
+import org.apache.flink.cdc.connectors.fluss.sink.v2.FlussRecordSerializer;
+import org.apache.flink.cdc.connectors.fluss.sink.v2.RowWithOp;
+
+import com.alibaba.fluss.client.Connection;
+import com.alibaba.fluss.client.table.Table;
+import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.types.DataType;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.flink.cdc.connectors.fluss.sink.v2.OperationType.APPEND;
+import static org.apache.flink.cdc.connectors.fluss.sink.v2.OperationType.DELETE;
+import static org.apache.flink.cdc.connectors.fluss.sink.v2.OperationType.UPSERT;
+import static org.apache.flink.cdc.connectors.fluss.utils.FlinkConversions.sameCdcColumnsIgnoreCommentAndDefaultValue;
+import static org.apache.flink.cdc.connectors.fluss.utils.FlinkConversions.toFlussSchema;
+
+/** Serialization schema that converts a CDC data record to a Fluss event. */
+public class FlussEventSerializationSchema implements FlussRecordSerializer<Event> {
+ private static final long serialVersionUID = 1L;
+
+ private transient Map<TableId, TableSchemaInfo> tableInfoMap;
+ private transient Connection connection;
+
+ @Override
+ public void open(Connection connection) {
+ this.tableInfoMap = new HashMap<>();
+ this.connection = connection;
+ }
+
+ @Override
+ public FlussEvent serialize(Event record) throws IOException {
+ if (record instanceof SchemaChangeEvent) {
+ applySchemaChangeEvent((SchemaChangeEvent) record);
+ return new FlussEvent(getTablePath(((SchemaChangeEvent) record).tableId()), null, true);
+ } else if (record instanceof DataChangeEvent) {
+ RowWithOp rowWithOp = applyDataChangeEvent((DataChangeEvent) record);
+ return new FlussEvent(
+ getTablePath(((DataChangeEvent) record).tableId()),
+ Collections.singletonList(rowWithOp),
+ false);
Review Comment:
```suggestion
public FlussEvent serialize(Event event) throws IOException {
if (event instanceof SchemaChangeEvent) {
applySchemaChangeEvent((SchemaChangeEvent) event);
return new FlussEvent(getTablePath(((SchemaChangeEvent) event).tableId()), null, true);
} else if (event instanceof DataChangeEvent) {
RowWithOp rowWithOp = applyDataChangeEvent((DataChangeEvent) event);
return new FlussEvent(
getTablePath(((DataChangeEvent) event).tableId()),
Collections.singletonList(rowWithOp),
false);
```
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussEventSerializationSchema.java:
##########
@@ -0,0 +1,184 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.flink.cdc.connectors.fluss.sink;
+
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.OperationType;
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.utils.Preconditions;
+import org.apache.flink.cdc.connectors.fluss.sink.v2.FlussEvent;
+import org.apache.flink.cdc.connectors.fluss.sink.v2.FlussRecordSerializer;
+import org.apache.flink.cdc.connectors.fluss.sink.v2.RowWithOp;
+
+import com.alibaba.fluss.client.Connection;
+import com.alibaba.fluss.client.table.Table;
+import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.types.DataType;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.flink.cdc.connectors.fluss.sink.v2.OperationType.APPEND;
+import static org.apache.flink.cdc.connectors.fluss.sink.v2.OperationType.DELETE;
+import static org.apache.flink.cdc.connectors.fluss.sink.v2.OperationType.UPSERT;
+import static org.apache.flink.cdc.connectors.fluss.utils.FlinkConversions.sameCdcColumnsIgnoreCommentAndDefaultValue;
+import static org.apache.flink.cdc.connectors.fluss.utils.FlinkConversions.toFlussSchema;
+
+/** Serialization schema that converts a CDC data record to a Fluss event. */
+public class FlussEventSerializationSchema implements FlussRecordSerializer<Event> {
Review Comment:
Could we unify the usage of `event` and `record`?
```suggestion
public class FlussEventSerializationSchema implements FlussEventSerializer<Event> {
```
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetaDataApplier.java:
##########
@@ -0,0 +1,181 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.flink.cdc.connectors.fluss.sink;
+
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DropTableEvent;
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.SchemaChangeEventType;
+import org.apache.flink.cdc.common.event.SchemaChangeEventTypeFamily;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.sink.MetadataApplier;
+import org.apache.flink.table.api.ValidationException;
+
+import com.alibaba.fluss.client.Connection;
+import com.alibaba.fluss.client.ConnectionFactory;
+import com.alibaba.fluss.client.admin.Admin;
+import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.metadata.DatabaseDescriptor;
+import com.alibaba.fluss.metadata.TableDescriptor;
+import com.alibaba.fluss.metadata.TableInfo;
+import com.alibaba.fluss.metadata.TablePath;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.cdc.common.event.SchemaChangeEventType.CREATE_TABLE;
+import static org.apache.flink.cdc.common.event.SchemaChangeEventType.DROP_TABLE;
+import static org.apache.flink.cdc.connectors.fluss.utils.FlinkConversions.toFlussTable;
+
+/** {@link MetadataApplier} for fluss. */
+public class FlussMetaDataApplier implements MetadataApplier {
+ private static final Logger LOG = LoggerFactory.getLogger(FlussMetaDataApplier.class);
+ private final Configuration flussClientConfig;
+ private final Map<String, String> tableProperties;
+ private final Map<String, List<String>> bucketKeysMap;
+ private final Map<String, Integer> bucketNumMap;
+ private Set<SchemaChangeEventType> enabledEventTypes =
+ new HashSet<>(Arrays.asList(CREATE_TABLE, DROP_TABLE));
+
+ public FlussMetaDataApplier(
+ Configuration flussClientConfig,
+ Map<String, String> tableProperties,
+ Map<String, List<String>> bucketKeysMap,
+ Map<String, Integer> bucketNumMap) {
+ this.flussClientConfig = flussClientConfig;
+ this.tableProperties = tableProperties;
+ this.bucketKeysMap = bucketKeysMap;
+ this.bucketNumMap = bucketNumMap;
+ }
+
+ @Override
+ public MetadataApplier setAcceptedSchemaEvolutionTypes(
+ Set<SchemaChangeEventType> schemaEvolutionTypes) {
+ enabledEventTypes = schemaEvolutionTypes;
+ return this;
+ }
+
+ @Override
+ public boolean acceptsSchemaEvolutionType(SchemaChangeEventType schemaChangeEventType) {
+ return enabledEventTypes.contains(schemaChangeEventType);
+ }
+
+ @Override
+ public Set<SchemaChangeEventType> getSupportedSchemaEvolutionTypes() {
+ return Arrays.stream(SchemaChangeEventTypeFamily.TABLE).collect(Collectors.toSet());
+ }
+
+ @Override
+ public void applySchemaChange(SchemaChangeEvent schemaChangeEvent) {
+ LOG.info("fluss metadata applier receive schemaChangeEvent {}", schemaChangeEvent);
+ if (schemaChangeEvent instanceof CreateTableEvent) {
+ CreateTableEvent createTableEvent = (CreateTableEvent) schemaChangeEvent;
+ applyCreateTable(createTableEvent);
+ } else if (schemaChangeEvent instanceof DropTableEvent) {
+ DropTableEvent dropTableEvent = (DropTableEvent) schemaChangeEvent;
+ applyDropTable(dropTableEvent);
+ } else {
+ throw new IllegalArgumentException(
+ "fluss metadata applier only support CreateTableEvent now but receives "
+ + schemaChangeEvent);
+ }
+ }
+
+ private void applyCreateTable(CreateTableEvent event) {
+ try (Connection connection = ConnectionFactory.createConnection(flussClientConfig);
+ Admin admin = connection.getAdmin()) {
+ TableId tableId = event.tableId();
+ TablePath tablePath = new TablePath(tableId.getSchemaName(), tableId.getTableName());
+ String tableIdentifier = tablePath.getDatabaseName() + "." + tablePath.getTableName();
+ List<String> bucketKeys = bucketKeysMap.get(tableIdentifier);
+ Integer bucketNum = bucketNumMap.get(tableIdentifier);
+ TableDescriptor inferredFlussTable =
+ toFlussTable(event.getSchema(), bucketKeys, bucketNum, tableProperties);
+ admin.createDatabase(tablePath.getDatabaseName(), DatabaseDescriptor.EMPTY, true);
+ if (!admin.tableExists(tablePath).get()) {
+ admin.createTable(tablePath, inferredFlussTable, false).get();
+ } else {
+ TableInfo currentTableInfo = admin.getTableInfo(tablePath).get();
+ // sanity check to prevent unexpected table schema evolution.
+ sanityCheck(inferredFlussTable, currentTableInfo);
+ }
+ } catch (Exception e) {
+ LOG.error("Failed to apply schema change {}", event, e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void applyDropTable(DropTableEvent event) {
+ try (Connection connection = ConnectionFactory.createConnection(flussClientConfig);
+ Admin admin = connection.getAdmin()) {
+ TableId tableId = event.tableId();
+ TablePath tablePath = new TablePath(tableId.getSchemaName(), tableId.getTableName());
+ admin.dropTable(tablePath, true).get();
+ } catch (Exception e) {
+ LOG.error("Failed to apply schema change {}", event, e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void sanityCheck(TableDescriptor inferredFlussTable, TableInfo currentTableInfo) {
+ List<String> inferredPrimaryKeyColumnNames =
+ inferredFlussTable.getSchema().getPrimaryKeyColumnNames().stream()
+ .sorted()
+ .collect(Collectors.toList());
+ List<String> currentPrimaryKeyColumnNames =
+ currentTableInfo.getSchema().getPrimaryKeyColumnNames().stream()
+ .sorted()
+ .collect(Collectors.toList());
+ if (!inferredPrimaryKeyColumnNames.equals(currentPrimaryKeyColumnNames)) {
+ throw new ValidationException(
+ "The CDC create table event schema is not matched to current Fluss table schema. "
+ + "\n New Fluss table's primary keys : "
+ + inferredPrimaryKeyColumnNames
+ + "\n Current Fluss's primary keys: "
+ + currentPrimaryKeyColumnNames);
+ }
+
+ List<String> inferredBucketKeys = inferredFlussTable.getBucketKeys();
+ List<String> currentBucketKeys = currentTableInfo.getBucketKeys();
+ if (!inferredBucketKeys.equals(currentBucketKeys)) {
+ throw new ValidationException(
+ "The CDC create table event schema is not matched to current Fluss table schema. "
Review Comment:
user would mislead by current exception message, what's CDC？ how about `The table schema inffered by Flink CDC is not matched with current Fluss table schema.`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37958][cdc-sink-connectors] Flink CDC Sink connector support fluss [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4053:
URL: https://github.com/apache/flink-cdc/pull/4053#discussion_r2189417343
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/sink/DefaultDataChangeEventHashFunctionProvider.java:
##########
@@ -63,10 +64,15 @@ public int hashcode(DataChangeEvent event) {
objectsToHash.add(tableId.getTableName());
// Primary key
- RecordData data =
- event.op().equals(OperationType.DELETE) ? event.before() : event.after();
- for (FieldGetter primaryKeyGetter : primaryKeyGetters) {
- objectsToHash.add(primaryKeyGetter.getFieldOrNull(data));
+ if (!primaryKeyGetters.isEmpty()) {
+ RecordData data =
+ event.op().equals(OperationType.DELETE) ? event.before() : event.after();
+ for (FieldGetter primaryKeyGetter : primaryKeyGetters) {
+ objectsToHash.add(primaryKeyGetter.getFieldOrNull(data));
+ }
+ } else {
+ // Avoid sending all events to the same subtask when table has no primary key.
+ objectsToHash.add(ThreadLocalRandom.current().nextInt());
Review Comment:
We should change the COMMON logic in a specific connector implementation, this will lead to data correct issue in the case that changelog comes from mysql table witout primary key
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37958][cdc-sink-connectors] Flink CDC Sink connector support fluss [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #4053:
URL: https://github.com/apache/flink-cdc/pull/4053#issuecomment-3044655501
<img width="1074" alt="image" src="https://github.com/user-attachments/assets/96d7d212-950d-440d-8843-b6891edfb850" />
Fluss tests success.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37958][cdc-sink-connectors] Flink CDC Sink connector support fluss [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on PR #4053:
URL: https://github.com/apache/flink-cdc/pull/4053#issuecomment-3034610768
Could @loserwang1024 please rebase this PR with latest `master` and try applying [this patch](https://github.com/loserwang1024/flink-cdc-connectors/pull/3/commits/1d3ff78d03a3c7bf5166e792be91f5749833a37a)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37958][cdc-sink-connectors] Flink CDC Sink connector support fluss [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #4053:
URL: https://github.com/apache/flink-cdc/pull/4053#discussion_r2189524583
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/sink/DefaultDataChangeEventHashFunctionProvider.java:
##########
@@ -63,10 +64,15 @@ public int hashcode(DataChangeEvent event) {
objectsToHash.add(tableId.getTableName());
// Primary key
- RecordData data =
- event.op().equals(OperationType.DELETE) ? event.before() : event.after();
- for (FieldGetter primaryKeyGetter : primaryKeyGetters) {
- objectsToHash.add(primaryKeyGetter.getFieldOrNull(data));
+ if (!primaryKeyGetters.isEmpty()) {
+ RecordData data =
+ event.op().equals(OperationType.DELETE) ? event.before() : event.after();
+ for (FieldGetter primaryKeyGetter : primaryKeyGetters) {
+ objectsToHash.add(primaryKeyGetter.getFieldOrNull(data));
+ }
+ } else {
+ // Avoid sending all events to the same subtask when table has no primary key.
+ objectsToHash.add(ThreadLocalRandom.current().nextInt());
Review Comment:
I move it to FlussHashFunctionProvider
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37958][cdc-sink-connectors] Flink CDC Sink connector support fluss [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4053:
URL: https://github.com/apache/flink-cdc/pull/4053#discussion_r2189390744
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/FlussPipelineITCase.java:
##########
@@ -0,0 +1,576 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.flink.cdc.connectors.fluss;
+
+import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.pipeline.PipelineOptions;
+import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.common.types.RowType;
+import org.apache.flink.cdc.composer.PipelineExecution;
+import org.apache.flink.cdc.composer.definition.PipelineDef;
+import org.apache.flink.cdc.composer.definition.SinkDef;
+import org.apache.flink.cdc.composer.definition.SourceDef;
+import org.apache.flink.cdc.composer.flink.FlinkPipelineComposer;
+import org.apache.flink.cdc.connectors.values.factory.ValuesDataFactory;
+import org.apache.flink.cdc.connectors.values.source.ValuesDataSourceHelper;
+import org.apache.flink.cdc.connectors.values.source.ValuesDataSourceOptions;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+
+import com.alibaba.fluss.config.ConfigOptions;
+import com.alibaba.fluss.config.MemorySize;
+import com.alibaba.fluss.metadata.DataLakeFormat;
+import com.alibaba.fluss.server.testutils.FlussClusterExtension;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static com.alibaba.fluss.config.ConfigOptions.BOOTSTRAP_SERVERS;
+import static com.alibaba.fluss.flink.source.testutils.FlinkRowAssertionsUtils.assertResultsIgnoreOrder;
+import static com.alibaba.fluss.server.testutils.FlussClusterExtension.BUILTIN_DATABASE;
+import static org.apache.flink.cdc.connectors.values.source.ValuesDataSourceHelper.TABLE_1;
+import static org.apache.flink.cdc.connectors.values.source.ValuesDataSourceHelper.TABLE_2;
+import static org.apache.flink.configuration.CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** ITCase for Fluss Pipeline. */
+public class FlussPipelineITCase {
+ private static final int MAX_PARALLELISM = 4;
+
+ // Always use parent-first classloader for CDC classes.
+ // The reason is that ValuesDatabase uses static field for holding data, we need to make sure
+ // the class is loaded by AppClassloader so that we can verify data in the test case.
+ private static final org.apache.flink.configuration.Configuration MINI_CLUSTER_CONFIG =
+ new org.apache.flink.configuration.Configuration();
+
+ /**
+ * Use {@link MiniClusterExtension} to reduce the overhead of restarting the MiniCluster for
+ * every test case.
+ */
+ @RegisterExtension
+ static final MiniClusterExtension MINI_CLUSTER_RESOURCE =
+ new MiniClusterExtension(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(MAX_PARALLELISM)
+ .build());
+
+ /**
+ * Use {@link FlussClusterExtension} to start a Fluss cluster with sasl authentication for every
+ * test case.
+ */
+ @RegisterExtension
+ public static final FlussClusterExtension FLUSS_CLUSTER_EXTENSION =
+ FlussClusterExtension.builder()
+ .setClusterConf(initConfig())
+ .setCoordinatorServerListeners("FLUSS://localhost:0, CLIENT://localhost:0")
+ .setTabletServerListeners("FLUSS://localhost:0, CLIENT://localhost:0")
+ .setNumOfTabletServers(3)
+ .build();
+
+ static {
+ MINI_CLUSTER_CONFIG.set(
+ ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL,
+ Collections.singletonList("org.apache.flink.cdc"));
+ }
+
+ static final String CATALOG_NAME = "test_catalog";
+ static final String DEFAULT_DB = "default_schema";
+
+ protected TableEnvironment tBatchEnv;
+
+ @BeforeEach
+ void before() {
+ // open a catalog so that we can get table from the catalog
+ String bootstrapServers = FLUSS_CLUSTER_EXTENSION.getBootstrapServers();
+
+ // create batch table environment
+ tBatchEnv =
+ TableEnvironment.create(EnvironmentSettings.newInstance().inBatchMode().build());
+ tBatchEnv.executeSql(
+ String.format(
+ "create catalog %s with ('type' = 'fluss', '%s' = '%s')",
+ CATALOG_NAME, BOOTSTRAP_SERVERS.key(), bootstrapServers));
+ tBatchEnv.executeSql("use catalog " + CATALOG_NAME);
+ tBatchEnv
+ .getConfig()
+ .set(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 2);
+ // create database
+ tBatchEnv.executeSql("create database " + DEFAULT_DB);
+ tBatchEnv.useDatabase(DEFAULT_DB);
+ }
+
+ @AfterEach
+ void after() {
+ tBatchEnv.useDatabase(BUILTIN_DATABASE);
+ tBatchEnv.executeSql(String.format("drop database %s cascade", DEFAULT_DB));
+ }
+
+ @Test
+ void testSinglePrimaryTable() throws Exception {
+ List<List<Event>> eventOfSplits = new ArrayList<>();
+ List<Event> split1 = new ArrayList<>();
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("col1", DataTypes.STRING())
+ .physicalColumn("col2", DataTypes.STRING())
+ .primaryKey("col1")
+ .build();
+ CreateTableEvent createTableEvent = new CreateTableEvent(TABLE_1, schema);
+ split1.add(createTableEvent);
+ BinaryRecordDataGenerator generator =
+ new BinaryRecordDataGenerator(RowType.of(DataTypes.STRING(), DataTypes.STRING()));
+ DataChangeEvent insertEvent1 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("a")
+ }));
+ split1.add(insertEvent1);
+ DataChangeEvent insertEvent2 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("b")
+ }));
+ split1.add(insertEvent2);
+ DataChangeEvent insertEvent3 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("3"),
+ BinaryStringData.fromString("c")
+ }));
+ split1.add(insertEvent3);
+ eventOfSplits.add(split1);
+ DataChangeEvent updateEvent =
+ DataChangeEvent.updateEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("b")
+ }),
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("b2")
+ }));
+ DataChangeEvent deleteEvent =
+ DataChangeEvent.deleteEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("a")
+ }));
+ eventOfSplits.add(Collections.singletonList(updateEvent));
+ eventOfSplits.add(Collections.singletonList(deleteEvent));
+
+ composeAndExecute(eventOfSplits);
+ checkResult(TABLE_1, Arrays.asList("+I[2, b2]", "+I[3, c]"));
+ }
+
+ @Test
+ void testSingleLogTable() throws Exception {
+ List<List<Event>> eventOfSplits = new ArrayList<>();
+ List<Event> split1 = new ArrayList<>();
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("col1", DataTypes.STRING())
+ .physicalColumn("col2", DataTypes.STRING())
+ .build();
+ CreateTableEvent createTableEvent = new CreateTableEvent(TABLE_1, schema);
+ split1.add(createTableEvent);
+ BinaryRecordDataGenerator generator =
+ new BinaryRecordDataGenerator(RowType.of(DataTypes.STRING(), DataTypes.STRING()));
+ DataChangeEvent insertEvent1 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("a")
+ }));
+ split1.add(insertEvent1);
+ DataChangeEvent insertEvent2 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("b")
+ }));
+ split1.add(insertEvent2);
+ DataChangeEvent insertEvent3 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("3"),
+ BinaryStringData.fromString("c")
+ }));
+ split1.add(insertEvent3);
+ eventOfSplits.add(split1);
+
+ composeAndExecute(eventOfSplits);
+ checkResult(TABLE_1, Arrays.asList("+I[1, a]", "+I[2, b]", "+I[3, c]"));
+ }
+
+ @Test
+ void testSingleLogTableWithSchemaChange() {
+ assertThatThrownBy(() -> composeAndExecute(ValuesDataSourceHelper.singleSplitSingleTable()))
+ .rootCause()
+ .hasMessageContaining(
+ "fluss metadata applier only support CreateTableEvent now but receives AddColumnEvent");
+ }
+
+ @Test
+ void testLackUsernameAndPassword() {
+ assertThatThrownBy(
+ () ->
+ composeAndExecute(
+ ValuesDataSourceHelper.singleSplitSingleTable(),
+ Collections.singletonMap(
+ BOOTSTRAP_SERVERS.key(), getBootstrapServers())))
+ .rootCause()
+ .hasMessageContaining(
+ "The connection has not completed authentication yet. This may be caused by a missing or incorrect configuration of 'client.security.protocol' on the client side.");
+ }
+
+ @Test
+ void testWrongTableOptions() {
+ Map<String, String> sinkOption = new HashMap<>();
+ sinkOption.put(BOOTSTRAP_SERVERS.key(), getBootstrapServers());
+ sinkOption.put(BOOTSTRAP_SERVERS.key(), getBootstrapServers());
+ sinkOption.put("properties.client.security.protocol", "sasl");
+ sinkOption.put("properties.client.security.sasl.mechanism", "PLAIN");
+ sinkOption.put("properties.client.security.sasl.username", "guest");
+ sinkOption.put("properties.client.security.sasl.password", "password2");
+ sinkOption.put("properties.table.non-key", "non-key-value");
+ assertThatThrownBy(
+ () ->
+ composeAndExecute(
+ ValuesDataSourceHelper.singleSplitSingleTable(),
+ sinkOption))
+ .rootCause()
+ .hasMessageContaining("'table.non-key' is not a Fluss table property");
+ }
+
+ @Test
+ void testMultiTables() throws Exception {
+ List<List<Event>> eventOfSplits = new ArrayList<>();
+ List<Event> split1 = new ArrayList<>();
+ BinaryRecordDataGenerator generator =
+ new BinaryRecordDataGenerator(RowType.of(DataTypes.STRING(), DataTypes.STRING()));
+
+ // create table
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("col1", DataTypes.STRING())
+ .physicalColumn("col2", DataTypes.STRING())
+ .primaryKey("col1")
+ .build();
+ CreateTableEvent createTableEvent = new CreateTableEvent(TABLE_1, schema);
+ split1.add(createTableEvent);
+ DataChangeEvent insertTable1Event1 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("1")
+ }));
+ split1.add(insertTable1Event1);
+ CreateTableEvent createTableEvent2 = new CreateTableEvent(TABLE_2, schema);
+ split1.add(createTableEvent2);
+ DataChangeEvent insertTable1Event2 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("2")
+ }));
+ split1.add(insertTable1Event2);
+ DataChangeEvent insertTable2Event1 =
+ DataChangeEvent.insertEvent(
+ TABLE_2,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("1")
+ }));
+ split1.add(insertTable2Event1);
+ DataChangeEvent insertTable1Event3 =
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("3"),
+ BinaryStringData.fromString("3")
+ }));
+ split1.add(insertTable1Event3);
+
+ DataChangeEvent insertTable2Event2 =
+ DataChangeEvent.insertEvent(
+ TABLE_2,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("2"),
+ BinaryStringData.fromString("2")
+ }));
+ split1.add(insertTable2Event2);
+ DataChangeEvent insertTabl2Event3 =
+ DataChangeEvent.insertEvent(
+ TABLE_2,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("3"),
+ BinaryStringData.fromString("3")
+ }));
+ split1.add(insertTabl2Event3);
+ eventOfSplits.add(split1);
+
+ composeAndExecute(eventOfSplits);
+ checkResult(TABLE_1, Arrays.asList("+I[1, 1]", "+I[2, 2]", "+I[3, 3]"));
+ checkResult(TABLE_2, Arrays.asList("+I[1, 1]", "+I[2, 2]", "+I[3, 3]"));
+ }
+
+ @Test
+ void testInsertExistTableWithMoreColumns() throws Exception {
+ // create a fluss table with 2 columns but then cdc read a source table with 3 columns
+ tBatchEnv
+ .executeSql(
+ String.format(
+ "CREATE TABLE %s.%s (\n"
+ + " col1 STRING,\n"
+ + " col2 STRING,\n"
Review Comment:
could you change the column order to cover more user case?
```suggestion
+ " col1 STRING,\n"
+ " col1 STRING,\n"
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37958][cdc-sink-connectors] Flink CDC Sink connector support fluss [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4053:
URL: https://github.com/apache/flink-cdc/pull/4053#discussion_r2189417343
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/sink/DefaultDataChangeEventHashFunctionProvider.java:
##########
@@ -63,10 +64,15 @@ public int hashcode(DataChangeEvent event) {
objectsToHash.add(tableId.getTableName());
// Primary key
- RecordData data =
- event.op().equals(OperationType.DELETE) ? event.before() : event.after();
- for (FieldGetter primaryKeyGetter : primaryKeyGetters) {
- objectsToHash.add(primaryKeyGetter.getFieldOrNull(data));
+ if (!primaryKeyGetters.isEmpty()) {
+ RecordData data =
+ event.op().equals(OperationType.DELETE) ? event.before() : event.after();
+ for (FieldGetter primaryKeyGetter : primaryKeyGetters) {
+ objectsToHash.add(primaryKeyGetter.getFieldOrNull(data));
+ }
+ } else {
+ // Avoid sending all events to the same subtask when table has no primary key.
+ objectsToHash.add(ThreadLocalRandom.current().nextInt());
Review Comment:
We should not change the COMMON logic in a specific connector implementation, this will lead to data correct issue in the case that changelog comes from mysql table witout primary key
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37958][cdc-sink-connectors] Flink CDC Sink connector support fluss [flink-cdc]
Posted by "beryllw (via GitHub)" <gi...@apache.org>.
beryllw commented on PR #4053:
URL: https://github.com/apache/flink-cdc/pull/4053#issuecomment-3030248708
Thanks for the great work, LGTM.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37958][cdc-sink-connectors] Flink CDC Sink connector support fluss [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #4053:
URL: https://github.com/apache/flink-cdc/pull/4053#issuecomment-3044219012
> And, a e2e case that synchronize mysql tables to fluss cluster with yaml file is welcome
Can I add it into another PR? Too much content for this pr
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37958][cdc-sink-connectors] Flink CDC Sink connector support fluss [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4053:
URL: https://github.com/apache/flink-cdc/pull/4053#discussion_r2189351013
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/resources/log4j2-test.properties:
##########
@@ -0,0 +1,25 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+################################################################################
+
+# Set root logger level to ERROR to not flood build logs
+# set manually to INFO for debugging purposes
+rootLogger.level=WARN
Review Comment:
change to `OFF ` to match with other modules
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/pom.xml:
##########
@@ -0,0 +1,139 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-pipeline-connectors</artifactId>
+ <version>${revision}</version>
+ </parent>
+
+ <artifactId>flink-cdc-pipeline-connector-fluss</artifactId>
+ <packaging>jar</packaging>
+
+ <name>flink-cdc-pipeline-connector-fluss</name>
+ <url>http://maven.apache.org</url>
+
+
+ <properties>
+ <fluss.version>0.7.0</fluss.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-client</artifactId>
+ <version>${fluss.version}</version>
+ </dependency>
+
+
+ <!-- Test dependencies -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-composer</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-server</artifactId>
+ <version>${fluss.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-server</artifactId>
+ <version>${fluss.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-test-utils</artifactId>
+ <version>${fluss.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <!-- In cdc yaml, we use our own sink. Thus, import fluss-fink just for test -->
Review Comment:
<!-- In Flink CDC project has Pipeline Sink Connector for Fluss. we import fluss-fink for Fluss Sink Connector just for test purpose -->
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/pom.xml:
##########
@@ -0,0 +1,139 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-pipeline-connectors</artifactId>
+ <version>${revision}</version>
+ </parent>
+
+ <artifactId>flink-cdc-pipeline-connector-fluss</artifactId>
+ <packaging>jar</packaging>
+
+ <name>flink-cdc-pipeline-connector-fluss</name>
+ <url>http://maven.apache.org</url>
+
+
+ <properties>
+ <fluss.version>0.7.0</fluss.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-client</artifactId>
+ <version>${fluss.version}</version>
+ </dependency>
+
+
+ <!-- Test dependencies -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-composer</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-server</artifactId>
+ <version>${fluss.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-server</artifactId>
+ <version>${fluss.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-test-utils</artifactId>
+ <version>${fluss.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <!-- In cdc yaml, we use our own sink. Thus, import fluss-fink just for test -->
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-flink-common</artifactId>
+ <version>${fluss.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-flink-1.20</artifactId>
+ <version>${fluss.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-planner_2.12</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-pipeline-connector-values</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+
+
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <version>${maven.shade.plugin.version}</version>
+ <executions>
+ <execution>
+ <id>shade-flink</id>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <shadeTestJar>false</shadeTestJar>
+ <artifactSet>
+ <includes>
+ <include>com.alibaba.fluss:*</include>
+ </includes>
Review Comment:
only include fluss dependency is enough? Do we need to shade common dependencies which are easy to conflict?
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/pom.xml:
##########
@@ -0,0 +1,139 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-pipeline-connectors</artifactId>
+ <version>${revision}</version>
+ </parent>
+
+ <artifactId>flink-cdc-pipeline-connector-fluss</artifactId>
+ <packaging>jar</packaging>
+
+ <name>flink-cdc-pipeline-connector-fluss</name>
+ <url>http://maven.apache.org</url>
+
+
+ <properties>
+ <fluss.version>0.7.0</fluss.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-client</artifactId>
+ <version>${fluss.version}</version>
+ </dependency>
+
+
+ <!-- Test dependencies -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-composer</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-server</artifactId>
+ <version>${fluss.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-server</artifactId>
+ <version>${fluss.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-test-utils</artifactId>
+ <version>${fluss.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <!-- In cdc yaml, we use our own sink. Thus, import fluss-fink just for test -->
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-flink-common</artifactId>
+ <version>${fluss.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-flink-1.20</artifactId>
+ <version>${fluss.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-planner_2.12</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-pipeline-connector-values</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+
+
Review Comment:
minor: remove blank lines
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/metrics/FlinkCounter.java:
##########
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.flink.cdc.connectors.fluss.sink.v2.metrics;
+
+import org.apache.flink.metrics.Counter;
+
+/* This file is based on source code of Apache Fluss Project (https://fluss.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/** An implementation of Flink's {@link Counter} which wraps Fluss's Counter. */
+public class FlinkCounter implements Counter {
+
+ private final com.alibaba.fluss.metrics.Counter wrapped;
+
+ public FlinkCounter(com.alibaba.fluss.metrics.Counter wrapped) {
+ this.wrapped = wrapped;
+ }
Review Comment:
```suggestion
public class WarppedFlussCounter implements Counter {
private final com.alibaba.fluss.metrics.Counter flussCounter;
public WarppedFlussCounter(com.alibaba.fluss.metrics.Counter flussCounter) {
this.flussCounter = flussCounter;
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37958][cdc-sink-connectors] Flink CDC Sink connector support fluss [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #4053:
URL: https://github.com/apache/flink-cdc/pull/4053#discussion_r2189448385
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/pom.xml:
##########
@@ -0,0 +1,139 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-pipeline-connectors</artifactId>
+ <version>${revision}</version>
+ </parent>
+
+ <artifactId>flink-cdc-pipeline-connector-fluss</artifactId>
+ <packaging>jar</packaging>
+
+ <name>flink-cdc-pipeline-connector-fluss</name>
+ <url>http://maven.apache.org</url>
+
+
+ <properties>
+ <fluss.version>0.7.0</fluss.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-client</artifactId>
+ <version>${fluss.version}</version>
+ </dependency>
+
+
+ <!-- Test dependencies -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-composer</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-server</artifactId>
+ <version>${fluss.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-server</artifactId>
+ <version>${fluss.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-test-utils</artifactId>
+ <version>${fluss.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <!-- In cdc yaml, we use our own sink. Thus, import fluss-fink just for test -->
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-flink-common</artifactId>
+ <version>${fluss.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-flink-1.20</artifactId>
+ <version>${fluss.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-planner_2.12</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-pipeline-connector-values</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+
+
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <version>${maven.shade.plugin.version}</version>
+ <executions>
+ <execution>
+ <id>shade-flink</id>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <shadeTestJar>false</shadeTestJar>
+ <artifactSet>
+ <includes>
+ <include>com.alibaba.fluss:*</include>
+ </includes>
Review Comment:
the common depedencies is provided by cdc framework. For example, flink-cdc-pipeline-connector-paimon also only package org.apache.paimon.
```xml
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>${maven.shade.plugin.version}</version>
<executions>
<execution>
<id>shade-flink</id>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<shadeTestJar>false</shadeTestJar>
<artifactSet>
<includes>
<include>org.apache.paimon:*</include>
</includes>
</artifactSet>
<relocations>
<relocation>
<pattern>org.apache.kafka</pattern>
<shadedPattern>shaded.paimon.org.apache.kafka</shadedPattern>
</relocation>
</relocations>
</configuration>
</execution>
</executions>
</plugin>
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37958][cdc-sink-connectors] Flink CDC Sink connector support fluss [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on PR #4053:
URL: https://github.com/apache/flink-cdc/pull/4053#issuecomment-3044600778
Thanks @loserwang1024 for the updating, LGTM
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37958][cdc-sink-connectors] Flink CDC Sink connector support fluss [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4053:
URL: https://github.com/apache/flink-cdc/pull/4053#discussion_r2189288420
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/FlussRecordSerializer.java:
##########
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.flink.cdc.connectors.fluss.sink.v2;
+
+import com.alibaba.fluss.client.Connection;
+
+import java.io.IOException;
+import java.io.Serializable;
+
+/**
+ * Serializer to serialize the input record to a {@link FlussEvent} for {@link FlinkSinkWriter}.
+ *
+ * @param <INPUT>
+ */
+public interface FlussRecordSerializer<INPUT> extends Serializable {
+ void open(Connection connection) throws IOException;
+
+ FlussEvent serialize(INPUT t) throws IOException;
Review Comment:
hint: `INPUT in` or `T t` for generic type
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/RowWithOp.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.flink.cdc.connectors.fluss.sink.v2;
+
+import com.alibaba.fluss.row.InternalRow;
+
+import javax.annotation.Nullable;
+
+import java.util.Objects;
+
+import static com.alibaba.fluss.utils.Preconditions.checkNotNull;
+
+/* This file is based on source code of Apache Fluss Project (https://fluss.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * A wrapper class that associates an {@link InternalRow} with an {@link OperationType} for use in
+ * Fluss-Flink data processing.
+ *
+ * <p>This class is used to represent a row of data along with its corresponding operation type,
+ * such as APPEND, UPSERT, or DELETE, as defined by {@link OperationType}.
+ *
+ * @see InternalRow
+ * @see OperationType
+ */
+public class RowWithOp {
Review Comment:
we can rename to `FlussRowWithOp ` to avoid misleading?
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussConfigUtils.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.flink.cdc.connectors.fluss.sink;
+
+import javax.annotation.Nullable;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/** Utils for parse fluss yaml sink options. */
Review Comment:
hint: `for parsing` or `used to parse`
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/FlussRecordSerializer.java:
##########
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.flink.cdc.connectors.fluss.sink.v2;
+
+import com.alibaba.fluss.client.Connection;
+
+import java.io.IOException;
+import java.io.Serializable;
+
+/**
+ * Serializer to serialize the input record to a {@link FlussEvent} for {@link FlinkSinkWriter}.
+ *
+ * @param <INPUT>
Review Comment:
plz add note for parameters
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/FlinkSinkWriter.java:
##########
@@ -0,0 +1,293 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.flink.cdc.connectors.fluss.sink.v2;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.cdc.connectors.fluss.sink.v2.metrics.FlinkMetricRegistry;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
+import org.apache.flink.table.api.ValidationException;
+
+import com.alibaba.fluss.client.Connection;
+import com.alibaba.fluss.client.ConnectionFactory;
+import com.alibaba.fluss.client.table.Table;
+import com.alibaba.fluss.client.table.writer.AppendWriter;
+import com.alibaba.fluss.client.table.writer.TableWriter;
+import com.alibaba.fluss.client.table.writer.UpsertWriter;
+import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.metadata.TableInfo;
+import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.metrics.Gauge;
+import com.alibaba.fluss.metrics.Metric;
+import com.alibaba.fluss.metrics.MetricNames;
+import com.alibaba.fluss.row.InternalRow;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+/** Base class for Flink {@link SinkWriter} implementations in Fluss. */
+public class FlinkSinkWriter<InputT> implements SinkWriter<InputT> {
Review Comment:
hint: `FlussSinkWriter`
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/utils/FlinkConversions.java:
##########
@@ -0,0 +1,255 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.flink.cdc.connectors.fluss.utils;
+
+import org.apache.flink.cdc.common.types.ArrayType;
+import org.apache.flink.cdc.common.types.BigIntType;
+import org.apache.flink.cdc.common.types.BinaryType;
+import org.apache.flink.cdc.common.types.BooleanType;
+import org.apache.flink.cdc.common.types.CharType;
+import org.apache.flink.cdc.common.types.DateType;
+import org.apache.flink.cdc.common.types.DecimalType;
+import org.apache.flink.cdc.common.types.DoubleType;
+import org.apache.flink.cdc.common.types.FloatType;
+import org.apache.flink.cdc.common.types.IntType;
+import org.apache.flink.cdc.common.types.LocalZonedTimestampType;
+import org.apache.flink.cdc.common.types.MapType;
+import org.apache.flink.cdc.common.types.RowType;
+import org.apache.flink.cdc.common.types.SmallIntType;
+import org.apache.flink.cdc.common.types.TimeType;
+import org.apache.flink.cdc.common.types.TimestampType;
+import org.apache.flink.cdc.common.types.TinyIntType;
+import org.apache.flink.cdc.common.types.VarBinaryType;
+import org.apache.flink.cdc.common.types.VarCharType;
+import org.apache.flink.cdc.common.types.ZonedTimestampType;
+import org.apache.flink.util.CollectionUtil;
+
+import com.alibaba.fluss.annotation.VisibleForTesting;
+import com.alibaba.fluss.metadata.Schema;
+import com.alibaba.fluss.metadata.TableDescriptor;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/** Converter from Flink's type to Fluss's type. */
+public class FlinkConversions {
Review Comment:
minor:FlussConversions
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/FlinkSinkWriter.java:
##########
@@ -0,0 +1,293 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.flink.cdc.connectors.fluss.sink.v2;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.cdc.connectors.fluss.sink.v2.metrics.FlinkMetricRegistry;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
+import org.apache.flink.table.api.ValidationException;
+
+import com.alibaba.fluss.client.Connection;
+import com.alibaba.fluss.client.ConnectionFactory;
+import com.alibaba.fluss.client.table.Table;
+import com.alibaba.fluss.client.table.writer.AppendWriter;
+import com.alibaba.fluss.client.table.writer.TableWriter;
+import com.alibaba.fluss.client.table.writer.UpsertWriter;
+import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.metadata.TableInfo;
+import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.metrics.Gauge;
+import com.alibaba.fluss.metrics.Metric;
+import com.alibaba.fluss.metrics.MetricNames;
+import com.alibaba.fluss.row.InternalRow;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+/** Base class for Flink {@link SinkWriter} implementations in Fluss. */
+public class FlinkSinkWriter<InputT> implements SinkWriter<InputT> {
+
+ protected static final Logger LOG = LoggerFactory.getLogger(FlinkSinkWriter.class);
+
+ private final Configuration flussConfig;
+ private final MailboxExecutor mailboxExecutor;
+ private final FlussRecordSerializer<InputT> flussRecordSerializer;
+
+ private transient Connection connection;
+ protected transient FlinkMetricRegistry flinkMetricRegistry;
+
+ protected transient SinkWriterMetricGroup metricGroup;
+
+ private transient Counter numRecordsOutCounter;
+ private transient Counter numRecordsOutErrorsCounter;
+ private volatile Throwable asyncWriterException;
+
+ private final Map<TablePath, TableWriter> writerMap;
+ private final Map<TablePath, Table> tableMap;
+
+ public FlinkSinkWriter(
+ Configuration flussConfig,
+ MailboxExecutor mailboxExecutor,
+ FlussRecordSerializer<InputT> flussRecordSerializer) {
+ this.flussConfig = flussConfig;
+ this.mailboxExecutor = mailboxExecutor;
+ this.flussRecordSerializer = flussRecordSerializer;
+ this.writerMap = new HashMap<>();
+ this.tableMap = new HashMap<>();
+ }
+
+ public void initialize(SinkWriterMetricGroup metricGroup) throws IOException {
+ LOG.info("Opening Fluss with config {}", flussConfig);
+ this.metricGroup = metricGroup;
+ flinkMetricRegistry =
+ new FlinkMetricRegistry(
+ metricGroup, Collections.singleton(MetricNames.WRITER_SEND_LATENCY_MS));
+ connection = ConnectionFactory.createConnection(flussConfig, flinkMetricRegistry);
+ flussRecordSerializer.open(connection);
+
+ initMetrics();
+ }
+
+ protected void initMetrics() {
+ numRecordsOutCounter = metricGroup.getNumRecordsSendCounter();
+ numRecordsOutErrorsCounter = metricGroup.getNumRecordsOutErrorsCounter();
+ metricGroup.setCurrentSendTimeGauge(this::computeSendTime);
+ }
+
+ @Override
+ public void write(InputT inputValue, Context context) throws IOException {
+ checkAsyncException();
+
+ try {
+ FlussEvent flussEvent = flussRecordSerializer.serialize(inputValue);
+
+ TablePath tablePath = flussEvent.getTablePath();
+
+ if (flussEvent.isShouldRefreshSchema() || !writerMap.containsKey(tablePath)) {
+ // refresh table schema
+ if (tableMap.containsKey(tablePath)) {
+ Table table = tableMap.remove(tablePath);
+ writerMap.remove(tablePath);
+ table.close();
+ }
+
+ Table table = connection.getTable(tablePath);
+ TableWriter writer;
+ if (table.getTableInfo().hasPrimaryKey()) {
+ writer = table.newUpsert().createWriter();
+ } else {
+ writer = table.newAppend().createWriter();
+ }
+ tableMap.put(tablePath, table);
+ writerMap.put(tablePath, writer);
+ }
+
+ List<RowWithOp> rowWithOps = flussEvent.getRowWithOps();
+ if (rowWithOps == null) {
+ return;
+ }
+ for (RowWithOp rowWithOp : rowWithOps) {
+ OperationType opType = rowWithOp.getOperationType();
+ InternalRow row = rowWithOp.getRow();
+ if (opType == OperationType.IGNORE) {
+ // skip writing the row
+ return;
+ }
+ CompletableFuture<?> writeFuture =
+ write(writerMap.get(tablePath), opType, row, tablePath);
+ writeFuture.whenComplete(
+ (ignored, throwable) -> {
+ if (throwable != null) {
+ if (this.asyncWriterException == null) {
+ this.asyncWriterException = throwable;
+ }
+
+ // Checking for exceptions from previous writes
+ mailboxExecutor.execute(
+ this::checkAsyncException, "Update error metric");
+ }
+ });
+
+ numRecordsOutCounter.inc();
+ }
+
+ } catch (Exception e) {
+ throw new IOException(e.getMessage(), e);
+ }
+ }
+
+ private CompletableFuture<?> write(
+ TableWriter writer, OperationType opType, InternalRow row, TablePath tablePath)
+ throws IOException {
+ if (writer instanceof UpsertWriter) {
+ UpsertWriter upsertWriter = (UpsertWriter) writer;
+ if (opType == OperationType.UPSERT) {
+ return upsertWriter.upsert(row);
+ } else if (opType == OperationType.DELETE) {
+ return upsertWriter.delete(row);
+ } else {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Unsupported operation type: %s for primary key table %s",
+ opType, tablePath));
+ }
+ } else if (writer instanceof AppendWriter) {
+ AppendWriter appendWriter = (AppendWriter) writer;
+ if (opType == OperationType.APPEND) {
+ return appendWriter.append(row);
+ } else {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Unsupported operation type: %s for log table %s",
+ opType, tablePath));
+ }
+ } else {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Unsupported writer type: %s for table %s",
+ writer.getClass(), tablePath));
+ }
+ }
+
+ public void flush(boolean endOfInput) throws IOException {
+ for (TableWriter writer : writerMap.values()) {
+ writer.flush();
+ checkAsyncException();
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ LOG.info("Closing Fluss sink function.");
+ try {
+ for (Table table : tableMap.values()) {
+ table.close();
+ }
+
+ tableMap.clear();
+
+ if (connection != null) {
+ connection.close();
+ }
+ } catch (Exception e) {
+ LOG.warn("Exception occurs while closing Fluss Connection.", e);
+ }
+ connection = null;
+
+ if (flinkMetricRegistry != null) {
+ flinkMetricRegistry.close();
+ }
+ flinkMetricRegistry = null;
+
+ // Rethrow exception for the case in which close is called before writer() and flush().
+ checkAsyncException();
+
+ LOG.info("Finished closing Fluss sink function.");
+ }
+
+ // todo: move sanityCheck to fluss.
+ private void sanityCheck(TableInfo flussTableInfo, boolean hasPrimaryKey) {
+ // when it's UpsertSinkWriter, it means it has primary key got from Flink's metadata
+ if (flussTableInfo.hasPrimaryKey() != hasPrimaryKey) {
+ throw new ValidationException(
+ String.format(
+ "Primary key constraint is not matched between metadata in Fluss (%s) and Flink (%s).",
+ flussTableInfo.hasPrimaryKey(), hasPrimaryKey));
+ }
+ // RowType currentTableRowType =
Review Comment:
remove this code piece and you can post a fluss jira link here
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/metrics/FlinkSink.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.flink.cdc.connectors.fluss.sink.v2.metrics;
+
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
+import org.apache.flink.cdc.connectors.fluss.sink.v2.FlinkSinkWriter;
+import org.apache.flink.cdc.connectors.fluss.sink.v2.FlussRecordSerializer;
+import org.apache.flink.runtime.metrics.groups.InternalSinkWriterMetricGroup;
+
+import com.alibaba.fluss.config.Configuration;
+
+import java.io.IOException;
+
+/* This file is based on source code of Apache Fluss Project (https://fluss.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/** Flink sink for Fluss. */
+public class FlinkSink<INPUT> implements Sink<INPUT> {
Review Comment:
Why you put this class under package `xx.metrics` ?
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/metrics/FlinkSink.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.flink.cdc.connectors.fluss.sink.v2.metrics;
+
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
+import org.apache.flink.cdc.connectors.fluss.sink.v2.FlinkSinkWriter;
+import org.apache.flink.cdc.connectors.fluss.sink.v2.FlussRecordSerializer;
+import org.apache.flink.runtime.metrics.groups.InternalSinkWriterMetricGroup;
+
+import com.alibaba.fluss.config.Configuration;
+
+import java.io.IOException;
+
+/* This file is based on source code of Apache Fluss Project (https://fluss.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/** Flink sink for Fluss. */
+public class FlinkSink<INPUT> implements Sink<INPUT> {
Review Comment:
Note, we're in Flink CDC project, thus let's call this class name `FlussSink` instead of `FlinkSink`
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/OperationType.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.flink.cdc.connectors.fluss.sink.v2;
+
+/* This file is based on source code of Apache Fluss Project (https://fluss.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * Enumeration of row operation types used in Fluss-Flink data processing.
+ *
+ * <p>This enum represents the type of operation associated with a row, such as an append (insert),
+ * upsert (update or insert), delete, or ignore. It is used to indicate how a row should be
+ * interpreted or processed in downstream systems.
+ *
+ * <ul>
+ * <li>{@link #APPEND} - Represents an append-only (insert) operation.
+ * <li>{@link #UPSERT} - Represents an upsert operation (update or insert).
+ * <li>{@link #DELETE} - Represents a delete operation.
+ * <li>{@link #IGNORE} - Represents an operation that should be ignored.
+ * </ul>
+ *
+ * @see com.alibaba.fluss.flink.row.RowWithOp
+ */
+public enum OperationType {
Review Comment:
we can rename to `FlussOperationType ` to avoid misleading?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37958][cdc-sink-connectors] Flink CDC Sink connector support fluss [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on PR #4053:
URL: https://github.com/apache/flink-cdc/pull/4053#issuecomment-3044356341
Left minor comments and please take care about the CI status?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37958][cdc-sink-connectors] Flink CDC Sink connector support fluss [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4053:
URL: https://github.com/apache/flink-cdc/pull/4053#discussion_r2189608758
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/metrics/FlinkHistogram.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.flink.cdc.connectors.fluss.sink.v2.metrics;
+
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.metrics.HistogramStatistics;
+
+/* This file is based on source code of Apache Fluss Project (https://fluss.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/** An implementation of Flink's {@link Histogram} which wraps Fluss's Histogram. */
+public class FlinkHistogram implements Histogram {
Review Comment:
ditto ?
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/metrics/FlinkGauge.java:
##########
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.flink.cdc.connectors.fluss.sink.v2.metrics;
+
+import org.apache.flink.metrics.Gauge;
+
+/* This file is based on source code of Apache Fluss Project (https://fluss.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/** An implementation of Flink's {@link Gauge} which wraps Fluss's Gauge. */
+public class FlinkGauge<T> implements Gauge<T> {
Review Comment:
ditto ?
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/metrics/FlinkMeter.java:
##########
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.flink.cdc.connectors.fluss.sink.v2.metrics;
+
+import org.apache.flink.metrics.Meter;
+
+/* This file is based on source code of Apache Fluss Project (https://fluss.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/** An implementation of Flink's {@link Meter} which wraps Fluss's Meter. */
+public class FlinkMeter implements Meter {
Review Comment:
ditto ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37958][cdc-sink-connectors] Flink CDC Sink connector support fluss [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang merged PR #4053:
URL: https://github.com/apache/flink-cdc/pull/4053
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org