You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by yu...@apache.org on 2024/01/08 06:54:37 UTC
(incubator-paimon) branch master updated: [hotfix] Clean redundant dynamic option copy code in cdc action (#2642)
This is an automated email from the ASF dual-hosted git repository.
yuzelin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new d59f119c9 [hotfix] Clean redundant dynamic option copy code in cdc action (#2642)
d59f119c9 is described below
commit d59f119c94733b027234aa7b606dd43a723de067
Author: Kerwin <37...@users.noreply.github.com>
AuthorDate: Mon Jan 8 14:54:32 2024 +0800
[hotfix] Clean redundant dynamic option copy code in cdc action (#2642)
---
.../sink/cdc/CdcMultiplexRecordChannelComputer.java | 6 +-----
.../sink/cdc/CdcRecordStoreMultiWriteOperator.java | 6 +-----
.../paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java | 17 ++++++-----------
.../flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java | 6 ++----
.../sink/cdc/CdcMultiplexRecordChannelComputerTest.java | 2 +-
.../sink/cdc/CdcRecordStoreMultiWriteOperatorTest.java | 3 +--
.../flink/sink/cdc/FlinkCdcMultiTableSinkTest.java | 3 +--
7 files changed, 13 insertions(+), 30 deletions(-)
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcMultiplexRecordChannelComputer.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcMultiplexRecordChannelComputer.java
index 86a6c398f..dd83c1a0e 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcMultiplexRecordChannelComputer.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcMultiplexRecordChannelComputer.java
@@ -43,12 +43,9 @@ public class CdcMultiplexRecordChannelComputer implements ChannelComputer<CdcMul
private Map<Identifier, CdcRecordChannelComputer> channelComputers;
private Catalog catalog;
- private final Map<String, String> dynamicOptions;
- public CdcMultiplexRecordChannelComputer(
- Catalog.Loader catalogLoader, Map<String, String> dynamicOptions) {
+ public CdcMultiplexRecordChannelComputer(Catalog.Loader catalogLoader) {
this.catalogLoader = catalogLoader;
- this.dynamicOptions = dynamicOptions;
}
@Override
@@ -76,7 +73,6 @@ public class CdcMultiplexRecordChannelComputer implements ChannelComputer<CdcMul
FileStoreTable table;
try {
table = (FileStoreTable) catalog.getTable(id);
- table.copy(dynamicOptions);
} catch (Catalog.TableNotExistException e) {
LOG.error("Failed to get table " + id.getFullName());
return null;
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperator.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperator.java
index 7330e63c5..af544def0 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperator.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperator.java
@@ -72,19 +72,16 @@ public class CdcRecordStoreMultiWriteOperator
private Map<Identifier, StoreSinkWrite> writes;
private String commitUser;
private ExecutorService compactExecutor;
- private final Map<String, String> dynamicOptions;
public CdcRecordStoreMultiWriteOperator(
Catalog.Loader catalogLoader,
StoreSinkWrite.WithWriteBufferProvider storeSinkWriteProvider,
String initialCommitUser,
- Options options,
- Map<String, String> dynamicOptions) {
+ Options options) {
super(options);
this.catalogLoader = catalogLoader;
this.storeSinkWriteProvider = storeSinkWriteProvider;
this.initialCommitUser = initialCommitUser;
- this.dynamicOptions = dynamicOptions;
}
@Override
@@ -181,7 +178,6 @@ public class CdcRecordStoreMultiWriteOperator
while (true) {
try {
table = (FileStoreTable) catalog.getTable(tableId);
- table.copy(dynamicOptions);
tables.put(tableId, table);
break;
} catch (Catalog.TableNotExistException e) {
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java
index b3a2ad287..99d933d89 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java
@@ -47,7 +47,6 @@ import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import javax.annotation.Nullable;
import java.io.Serializable;
-import java.util.Map;
import java.util.UUID;
import static org.apache.paimon.flink.sink.FlinkSink.assertStreamingConfiguration;
@@ -91,22 +90,20 @@ public class FlinkCdcMultiTableSink implements Serializable {
metricGroup);
}
- public DataStreamSink<?> sinkFrom(
- DataStream<CdcMultiplexRecord> input, Map<String, String> dynamicOptions) {
+ public DataStreamSink<?> sinkFrom(DataStream<CdcMultiplexRecord> input) {
// This commitUser is valid only for new jobs.
// After the job starts, this commitUser will be recorded into the states of write and
// commit operators.
// When the job restarts, commitUser will be recovered from states and this value is
// ignored.
String initialCommitUser = UUID.randomUUID().toString();
- return sinkFrom(input, initialCommitUser, createWriteProvider(), dynamicOptions);
+ return sinkFrom(input, initialCommitUser, createWriteProvider());
}
public DataStreamSink<?> sinkFrom(
DataStream<CdcMultiplexRecord> input,
String commitUser,
- StoreSinkWrite.WithWriteBufferProvider sinkProvider,
- Map<String, String> dynamicOptions) {
+ StoreSinkWrite.WithWriteBufferProvider sinkProvider) {
StreamExecutionEnvironment env = input.getExecutionEnvironment();
assertStreamingConfiguration(env);
MultiTableCommittableTypeInfo typeInfo = new MultiTableCommittableTypeInfo();
@@ -114,7 +111,7 @@ public class FlinkCdcMultiTableSink implements Serializable {
input.transform(
WRITER_NAME,
typeInfo,
- createWriteOperator(sinkProvider, commitUser, dynamicOptions))
+ createWriteOperator(sinkProvider, commitUser))
.setParallelism(input.getParallelism());
// shuffle committables by table
@@ -144,11 +141,9 @@ public class FlinkCdcMultiTableSink implements Serializable {
}
protected OneInputStreamOperator<CdcMultiplexRecord, MultiTableCommittable> createWriteOperator(
- StoreSinkWrite.WithWriteBufferProvider writeProvider,
- String commitUser,
- Map<String, String> dynamicOptions) {
+ StoreSinkWrite.WithWriteBufferProvider writeProvider, String commitUser) {
return new CdcRecordStoreMultiWriteOperator(
- catalogLoader, writeProvider, commitUser, new Options(), dynamicOptions);
+ catalogLoader, writeProvider, commitUser, new Options());
}
// Table committers are dynamically created at runtime
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
index b09d09ad9..9d675e332 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
@@ -65,7 +65,6 @@ public class FlinkCdcSyncDatabaseSinkBuilder<T> {
@Nullable private Integer parallelism;
private double committerCpu;
@Nullable private MemorySize committerMemory;
- private Map<String, String> dynamicOptions;
// Paimon catalog used to check and create tables. There will be two
// places where this catalog is used. 1) in processing function,
@@ -98,7 +97,6 @@ public class FlinkCdcSyncDatabaseSinkBuilder<T> {
}
public FlinkCdcSyncDatabaseSinkBuilder<T> withTableOptions(Options options) {
- this.dynamicOptions = options.toMap();
this.parallelism = options.get(FlinkConnectorOptions.SINK_PARALLELISM);
this.committerCpu = options.get(FlinkConnectorOptions.SINK_COMMITTER_CPU);
this.committerMemory = options.get(FlinkConnectorOptions.SINK_COMMITTER_MEMORY);
@@ -157,12 +155,12 @@ public class FlinkCdcSyncDatabaseSinkBuilder<T> {
DataStream<CdcMultiplexRecord> partitioned =
partition(
newlyAddedTableStream,
- new CdcMultiplexRecordChannelComputer(catalogLoader, dynamicOptions),
+ new CdcMultiplexRecordChannelComputer(catalogLoader),
parallelism);
FlinkCdcMultiTableSink sink =
new FlinkCdcMultiTableSink(catalogLoader, committerCpu, committerMemory);
- sink.sinkFrom(partitioned, dynamicOptions);
+ sink.sinkFrom(partitioned);
}
private void buildForFixedBucket(FileStoreTable table, DataStream<CdcRecord> parsed) {
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcMultiplexRecordChannelComputerTest.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcMultiplexRecordChannelComputerTest.java
index c20cc4440..2ae627552 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcMultiplexRecordChannelComputerTest.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcMultiplexRecordChannelComputerTest.java
@@ -149,7 +149,7 @@ public class CdcMultiplexRecordChannelComputerTest {
int numChannels = random.nextInt(10) + 1;
CdcMultiplexRecordChannelComputer channelComputer =
- new CdcMultiplexRecordChannelComputer(catalogLoader, new HashMap<>());
+ new CdcMultiplexRecordChannelComputer(catalogLoader);
channelComputer.setup(numChannels);
// assert that insert and delete records are routed into same channel
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperatorTest.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperatorTest.java
index e44c0ddca..23ca641c3 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperatorTest.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperatorTest.java
@@ -810,8 +810,7 @@ public class CdcRecordStoreMultiWriteOperatorTest {
memoryPoolFactory,
metricGroup),
commitUser,
- Options.fromMap(new HashMap<>()),
- new HashMap<>());
+ Options.fromMap(new HashMap<>()));
TypeSerializer<CdcMultiplexRecord> inputSerializer = new JavaSerializer<>();
TypeSerializer<MultiTableCommittable> outputSerializer =
new MultiTableCommittableTypeInfo().createSerializer(new ExecutionConfig());
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSinkTest.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSinkTest.java
index 585372c03..56235ca95 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSinkTest.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSinkTest.java
@@ -31,7 +31,6 @@ import org.apache.flink.streaming.api.transformations.OneInputTransformation;
import org.apache.flink.streaming.api.transformations.PartitionTransformation;
import org.junit.jupiter.api.Test;
-import java.util.Collections;
import java.util.concurrent.ThreadLocalRandom;
import static org.assertj.core.api.Assertions.assertThat;
@@ -60,7 +59,7 @@ public class FlinkCdcMultiTableSinkTest {
() -> FlinkCatalogFactory.createPaimonCatalog(new Options()),
FlinkConnectorOptions.SINK_COMMITTER_CPU.defaultValue(),
null);
- DataStreamSink<?> dataStreamSink = sink.sinkFrom(input, Collections.emptyMap());
+ DataStreamSink<?> dataStreamSink = sink.sinkFrom(input);
// check the transformation graph
LegacySinkTransformation<?> end =