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

[cdc] Added pulsar database cdc synchronization action #2345

Merged
merged 13 commits into from
Nov 30, 2023
Merged

[cdc] Added pulsar database cdc synchronization action #2345

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
1f9352b
Added pulsar database cdc synchronization action
zhuangchong Nov 17, 2023
8ad26e7
add abstract SyncDatabaseActionBase.
zhuangchong Nov 17, 2023
a4a29f1
fix code style
zhuangchong Nov 17, 2023
490f6e4
Merge branch 'master' into cdc-add-pulsar-database
zhuangchong Nov 20, 2023
b52a2d3
fix
zhuangchong Nov 20, 2023
49efca5
fix
zhuangchong Nov 20, 2023
c9a569e
fix code style
zhuangchong Nov 20, 2023
b259cbb
fix
zhuangchong Nov 21, 2023
fe07026
Modify only add pulsar database cdc
zhuangchong Nov 21, 2023
5be28ce
fix
zhuangchong Nov 21, 2023
1571f8f
add docs
zhuangchong Nov 22, 2023
644757f
fix
zhuangchong Nov 30, 2023
4c69beb
fix
zhuangchong Nov 30, 2023
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
Prev Previous commit
Next Next commit
fix
  • Loading branch information
@zhuangchong
zhuangchong committed Nov 21, 2023
commit b259cbb677b69fef51c1c45a69dca1ce1fbd512f
2 changes: 1 addition & 1 deletion ...k-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarActionITCaseBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,7 @@ public class PulsarActionITCaseBase extends CdcActionITCaseBase {

private PulsarAdmin admin;
private PulsarClient client;
protected List<String> topics = new ArrayList<>();

@RegisterExtension
public static final PulsarContainerExtension PULSAR_CONTAINER =
Expand Down Expand Up @@ -191,7 +192,6 @@ private void createPartitionedTopic(String topic, int numberOfPartitions) {
}

private void deleteTopics() throws Exception {
List<String> topics = admin.topics().getList("public/default");
for (String topic : topics) {
String topicName = topicName(topic);
PartitionedTopicMetadata metadata =
Expand Down
5 changes: 3 additions & 2 deletions ...c/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncDatabaseActionITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@ public void testSchemaEvolutionMultiTopic() throws Exception {
final String topic3 = "schema_evolution_2";
boolean writeOne = false;
int fileCount = 3;
List<String> topics = Arrays.asList(topic1, topic2, topic3);
topics = Arrays.asList(topic1, topic2, topic3);
topics.forEach(topic -> createTopic(topic, 1));

// ---------- Write the Canal json into Pulsar -------------------
Expand All @@ -65,6 +65,7 @@ public void testSchemaEvolutionMultiTopic() throws Exception {
}

Map<String, String> pulsarConfig = getBasicPulsarConfig();
pulsarConfig.put(PULSAR_PARTITION_DISCOVERY_INTERVAL_MS.key(), "-1");
pulsarConfig.put(VALUE_FORMAT.key(), "canal-json");
pulsarConfig.put(TOPIC.key(), String.join(",", topics));
PulsarSyncDatabaseAction action =
Expand All @@ -82,7 +83,7 @@ public void testSchemaEvolutionOneTopic() throws Exception {
final String topic = "schema_evolution";
boolean writeOne = true;
int fileCount = 3;
List<String> topics = Collections.singletonList(topic);
topics = Collections.singletonList(topic);
topics.forEach(t -> createTopic(t, 1));

// ---------- Write the Canal json into Pulsar -------------------
Expand Down
2 changes: 2 additions & 0 deletions .../src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncTableActionITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
import org.junit.jupiter.api.Timeout;

import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;

Expand All @@ -45,6 +46,7 @@ public void testSchemaEvolution() throws Exception {

private void runSingleTableSchemaEvolution(String sourceDir) throws Exception {
final String topic = "schema_evolution";
topics = Collections.singletonList(topic);
createTopic(topic, 1);
// ---------- Write the Canal json into Pulsar -------------------
sendMessages(
Expand Down
Loading