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

Added functionality to rename tables of the events coming from debezium. #96

Merged
merged 2 commits into from
Oct 30, 2023
+37 −0
Merged

Added functionality to rename tables of the events coming from debezium. #96

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3422b0c
rename table name
makalaaneesh Oct 17, 2023
ea858e0
refactor + handle empty string
makalaaneesh Oct 18, 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
37 changes: 37 additions & 0 deletions ...rver-ybexporter/src/main/java/io/debezium/server/ybexporter/KafkaConnectRecordParser.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
package io.debezium.server.ybexporter;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

Expand All @@ -14,6 +15,8 @@
import org.apache.kafka.connect.json.JsonConverter;
import org.apache.kafka.connect.json.JsonConverterConfig;
import org.apache.kafka.connect.source.SourceRecord;
import org.eclipse.microprofile.config.Config;
import org.eclipse.microprofile.config.ConfigProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand All @@ -24,6 +27,7 @@ class KafkaConnectRecordParser implements RecordParser {
String sourceType;
private Map<String, Table> tableMap;
private JsonConverter jsonConverter;
private Map<String, String> renameTables;
Record r = new Record();

public KafkaConnectRecordParser(String dataDirStr, String sourceType, Map<String, Table> tblMap) {
Expand All @@ -34,6 +38,30 @@ public KafkaConnectRecordParser(String dataDirStr, String sourceType, Map<String
jsonConverter = new JsonConverter();
Map<String, String> jsonConfig = Collections.singletonMap(JsonConverterConfig.SCHEMAS_ENABLE_CONFIG, "false");
jsonConverter.configure(jsonConfig, false);
renameTables = new HashMap<>();
retrieveRenameTablesFromConfig();
}

private void retrieveRenameTablesFromConfig(){
final Config config = ConfigProvider.getConfig();
String renameTablesConfig = config.getOptionalValue("debezium.sink.ybexporter.tables.rename", String.class).orElse("");
if (!renameTablesConfig.isEmpty()){
for (String renameTableConfig: renameTablesConfig.split(",")){
String[] beforeAndAfter = renameTableConfig.split(":");
if (beforeAndAfter.length != 2){
throw new RuntimeException(String.format("Incorrect format for specifying table rename config %s. Provide it as <oldname>:<newname>", renameTableConfig));
}
String before = beforeAndAfter[0];
String after = beforeAndAfter[1];
if ((before.split("\\.").length != 2) && (!sourceType.equals("mysql"))){
throw new RuntimeException(String.format("Incorrect format for specifying table rename config %s. Provide it as <schema>.<tableName>", before));
}
if ((after.split("\\.").length != 2) && (!sourceType.equals("mysql"))){
throw new RuntimeException(String.format("Incorrect format for specifying table rename config %s. Provide it as <schema>.<tableName>", after));
}
renameTables.put(before, after);
}
}
}

/**
Expand Down Expand Up @@ -84,6 +112,15 @@ protected void parseTable(Struct value, Struct sourceNode, Record r) {
schemaName = sourceNode.getString("schema");
}
String tableName = sourceNode.getString("table");
// rename table name
String qualifiedTableName = tableName;
if (!schemaName.equals("")){
qualifiedTableName = schemaName + "." + tableName;
}
if (renameTables.containsKey(qualifiedTableName)){
String[] renamedTableName = renameTables.get(qualifiedTableName).split("\\.");
tableName = renamedTableName[renamedTableName.length - 1];
}
var tableIdentifier = dbName + "-" + schemaName + "-" + tableName;

Table t = tableMap.get(tableIdentifier);
Expand Down