Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[FLINK-35757][mysql] support dynamically capture new tables #3467

Open
wants to merge 14 commits into
base: master
Choose a base branch
from
Open

[FLINK-35757][mysql] support dynamically capture new tables #3467

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
e1d2517
support add tables
Karl-WangSK Jun 30, 2024
74a6aa6
add logic
Jul 2, 2024
a936aa0
support add new table logic
Karl-WangSK Jul 2, 2024
c8efe72
support dynamically add new tables
Jul 4, 2024
c682522
support new add table
Jul 4, 2024
320fff3
support add tables
Jul 8, 2024
c66e1a2
support add tables
Jul 8, 2024
5389541
support new add tables
Jul 9, 2024
ecc7042
add test case
Jul 10, 2024
a0ae444
update
Jul 10, 2024
557f4a6
update
Jul 10, 2024
0dca9c7
fix checkstyle
Jul 15, 2024
29cc5ec
add CdcDebeziumDeserializationSchema
Aug 1, 2024
cbe43c5
update
Aug 2, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions ...tor-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,9 +51,12 @@ public EventSourceProvider getEventSourceProvider() {
new MySqlSource<>(
configFactory,
deserializer,
(sourceReaderMetrics, sourceConfig) ->
(sourceReaderContext, sourceReaderMetrics, sourceConfig) ->
new MySqlPipelineRecordEmitter(
deserializer, sourceReaderMetrics, sourceConfig));
sourceReaderContext,
deserializer,
sourceReaderMetrics,
sourceConfig));

return FlinkSourceProvider.of(source);
}
Expand Down
3 changes: 3 additions & 0 deletions .../java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.flink.cdc.connectors.mysql.source.reader;

import org.apache.flink.api.connector.source.SourceOutput;
import org.apache.flink.api.connector.source.SourceReaderContext;
import org.apache.flink.cdc.common.event.CreateTableEvent;
import org.apache.flink.cdc.common.event.Event;
import org.apache.flink.cdc.common.schema.Schema;
Expand Down Expand Up @@ -71,10 +72,12 @@ public class MySqlPipelineRecordEmitter extends MySqlRecordEmitter<Event> {
private final List<CreateTableEvent> createTableEventCache;

public MySqlPipelineRecordEmitter(
SourceReaderContext context,
DebeziumDeserializationSchema<Event> debeziumDeserializationSchema,
MySqlSourceReaderMetrics sourceReaderMetrics,
MySqlSourceConfig sourceConfig) {
super(
context,
debeziumDeserializationSchema,
sourceReaderMetrics,
sourceConfig.isIncludeSchemaChanges());
Expand Down
4 changes: 4 additions & 0 deletions ...m/src/main/java/org/apache/flink/cdc/debezium/table/RowDataDebeziumDeserializeSchema.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,6 +120,10 @@ public static Builder newBuilder() {
@Override
public void deserialize(SourceRecord record, Collector<RowData> out) throws Exception {
Envelope.Operation op = Envelope.operationFor(record);
// filter ddl statement to avoid NPE
if (op == null) {
return;
}
Struct value = (Struct) record.value();
Schema valueSchema = record.valueSchema();
if (op == Envelope.Operation.CREATE || op == Envelope.Operation.READ) {
Expand Down
11 changes: 2 additions & 9 deletions ...rc/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -264,15 +264,8 @@ private boolean hasEnterPureBinlogPhase(TableId tableId, BinlogOffset position)
return true;
}

// Use still need to capture new sharding table if user disable scan new added table,
// The history records for all new added tables(including sharding table and normal table)
// will be capture after restore from a savepoint if user enable scan new added table
if (!statefulTaskContext.getSourceConfig().isScanNewlyAddedTableEnabled()) {
// the new added sharding table without history records
return !maxSplitHighWatermarkMap.containsKey(tableId)
&& capturedTableFilter.isIncluded(tableId);
}
return false;
return !maxSplitHighWatermarkMap.containsKey(tableId)
&& capturedTableFilter.isIncluded(tableId);
}

private void configureFilter() {
Expand Down
9 changes: 6 additions & 3 deletions ...tor-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,8 +127,9 @@ public static <T> MySqlSourceBuilder<T> builder() {
this(
configFactory,
deserializationSchema,
(sourceReaderMetrics, sourceConfig) ->
(sourceReaderContext, sourceReaderMetrics, sourceConfig) ->
new MySqlRecordEmitter<>(
sourceReaderContext,
deserializationSchema,
sourceReaderMetrics,
sourceConfig.isIncludeSchemaChanges()));
Expand Down Expand Up @@ -185,7 +186,7 @@ public SourceReader<T, MySqlSplit> createReader(SourceReaderContext readerContex
return new MySqlSourceReader<>(
elementsQueue,
splitReaderSupplier,
recordEmitterSupplier.get(sourceReaderMetrics, sourceConfig),
recordEmitterSupplier.get(readerContext, sourceReaderMetrics, sourceConfig),
readerContext.getConfiguration(),
mySqlSourceReaderContext,
sourceConfig);
Expand Down Expand Up @@ -272,6 +273,8 @@ public void setSnapshotHooks(SnapshotPhaseHooks snapshotHooks) {
interface RecordEmitterSupplier<T> extends Serializable {

RecordEmitter<SourceRecords, T, MySqlSplitState> get(
MySqlSourceReaderMetrics metrics, MySqlSourceConfig sourceConfig);
SourceReaderContext sourceReaderContext,
MySqlSourceReaderMetrics metrics,
MySqlSourceConfig sourceConfig);
}
}
8 changes: 7 additions & 1 deletion ...java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlBinlogSplitAssigner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,8 @@
import org.apache.flink.cdc.connectors.mysql.source.split.MySqlSplit;
import org.apache.flink.util.CollectionUtil;

import io.debezium.relational.TableId;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
Expand Down Expand Up @@ -77,7 +79,8 @@ public boolean waitingForFinishedSplits() {
}

@Override
public List<FinishedSnapshotSplitInfo> getFinishedSplitInfos() {
public List<FinishedSnapshotSplitInfo> getFinishedSplitInfos(
boolean isScanNewlyAddedTableEnabled) {
return Collections.emptyList();
}

Expand Down Expand Up @@ -120,6 +123,9 @@ public void startAssignNewlyAddedTables() {}
@Override
public void onBinlogSplitUpdated() {}

@Override
public void addAlreadyProcessedTables(TableId tableId) {}

@Override
public void close() {}

Expand Down
10 changes: 8 additions & 2 deletions ...java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlHybridSplitAssigner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,8 +135,9 @@ public boolean waitingForFinishedSplits() {
}

@Override
public List<FinishedSnapshotSplitInfo> getFinishedSplitInfos() {
return snapshotSplitAssigner.getFinishedSplitInfos();
public List<FinishedSnapshotSplitInfo> getFinishedSplitInfos(
boolean isScanNewlyAddedTableEnabled) {
return snapshotSplitAssigner.getFinishedSplitInfos(isScanNewlyAddedTableEnabled);
}

@Override
Expand All @@ -158,6 +159,11 @@ public void addSplits(Collection<MySqlSplit> splits) {
snapshotSplitAssigner.addSplits(snapshotSplits);
}

@Override
public void addAlreadyProcessedTables(TableId tableId) {
snapshotSplitAssigner.addAlreadyProcessedTables(tableId);
}

@Override
public PendingSplitsState snapshotState(long checkpointId) {
return new HybridPendingSplitsState(
Expand Down
20 changes: 18 additions & 2 deletions ...va/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,7 @@ public class MySqlSnapshotSplitAssigner implements MySqlSplitAssigner {

private final List<TableId> alreadyProcessedTables;
private final List<MySqlSchemalessSnapshotSplit> remainingSplits;
private final List<MySqlSchemalessSnapshotSplit> newAddedTableRemainingSplits;
private final Map<String, MySqlSchemalessSnapshotSplit> assignedSplits;
private final Map<TableId, TableChanges.TableChange> tableSchemas;
private final Map<String, BinlogOffset> splitFinishedOffsets;
Expand Down Expand Up @@ -146,6 +147,7 @@ private MySqlSnapshotSplitAssigner(
this.currentParallelism = currentParallelism;
this.alreadyProcessedTables = alreadyProcessedTables;
this.remainingSplits = new CopyOnWriteArrayList<>(remainingSplits);
this.newAddedTableRemainingSplits = new CopyOnWriteArrayList<>();
// When job restore from savepoint, sort the existing tables and newly added tables
// to let enumerator only send newly added tables' BinlogSplitMetaEvent
this.assignedSplits =
Expand Down Expand Up @@ -319,6 +321,7 @@ private void splitTable(TableId nextTable) {
.collect(Collectors.toList());
chunkNum += splits.size();
remainingSplits.addAll(schemaLessSnapshotSplits);
newAddedTableRemainingSplits.addAll(schemaLessSnapshotSplits);
if (!chunkSplitter.hasNextChunk()) {
remainingTables.remove(nextTable);
}
Expand Down Expand Up @@ -369,15 +372,23 @@ public boolean waitingForFinishedSplits() {
}

@Override
public List<FinishedSnapshotSplitInfo> getFinishedSplitInfos() {
public List<FinishedSnapshotSplitInfo> getFinishedSplitInfos(
boolean isScanNewlyAddedTableEnabled) {
if (waitingForFinishedSplits()) {
LOG.error(
"The assigner is not ready to offer finished split information, this should not be called");
throw new FlinkRuntimeException(
"The assigner is not ready to offer finished split information, this should not be called");
}
Collection<MySqlSchemalessSnapshotSplit> snapshotSplit;
if (isScanNewlyAddedTableEnabled && !newAddedTableRemainingSplits.isEmpty()) {
snapshotSplit = newAddedTableRemainingSplits;
} else {
snapshotSplit = assignedSplits.values();
}

final List<MySqlSchemalessSnapshotSplit> assignedSnapshotSplit =
new ArrayList<>(assignedSplits.values());
new ArrayList<>(snapshotSplit);
List<FinishedSnapshotSplitInfo> finishedSnapshotSplitInfos = new ArrayList<>();
for (MySqlSchemalessSnapshotSplit split : assignedSnapshotSplit) {
BinlogOffset binlogOffset = splitFinishedOffsets.get(split.splitId());
Expand Down Expand Up @@ -607,4 +618,9 @@ private static MySqlChunkSplitter createChunkSplitter(
}
return new MySqlChunkSplitter(mySqlSchema, sourceConfig);
}

@Override
public void addAlreadyProcessedTables(TableId tableId) {
alreadyProcessedTables.add(tableId);
}
}
6 changes: 5 additions & 1 deletion .../main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSplitAssigner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,8 @@
import org.apache.flink.cdc.connectors.mysql.source.split.FinishedSnapshotSplitInfo;
import org.apache.flink.cdc.connectors.mysql.source.split.MySqlSplit;

import io.debezium.relational.TableId;

import java.util.Collection;
import java.util.List;
import java.util.Map;
Expand Down Expand Up @@ -59,7 +61,7 @@ public interface MySqlSplitAssigner {
* Gets the finished splits' information. This is useful metadata to generate a binlog split
* that considering finished snapshot splits.
*/
List<FinishedSnapshotSplitInfo> getFinishedSplitInfos();
List<FinishedSnapshotSplitInfo> getFinishedSplitInfos(boolean isScanNewlyAddedTableEnabled);

/**
* Callback to handle the finished splits with finished binlog offset. This is useful for
Expand Down Expand Up @@ -116,6 +118,8 @@ public interface MySqlSplitAssigner {
*/
void onBinlogSplitUpdated();

void addAlreadyProcessedTables(TableId tableId);

/**
* Called to close the assigner, in case it holds on to any resources, like threads or network
* connections.
Expand Down
Loading
Loading