DBZ-6641: Schema name changed with Custom topic naming strategy
This commit is contained in:
parent
55650fac7d
commit
a4c0bf5614
@ -31,6 +31,10 @@ public interface TopicNamingStrategy<I extends DataCollectionId> {
|
|||||||
|
|
||||||
String transactionTopic();
|
String transactionTopic();
|
||||||
|
|
||||||
|
default String recordSchemaPrefix(I id) {
|
||||||
|
return dataChangeTopic(id);
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Sanitize the given character whether is a legal character of a Kafka topic name.
|
* Sanitize the given character whether is a legal character of a Kafka topic name.
|
||||||
* Legal characters are {@code [a-zA-Z0-9._-]}.
|
* Legal characters are {@code [a-zA-Z0-9._-]}.
|
||||||
|
@ -124,8 +124,9 @@ public TableSchemaBuilder(ValueConverterProvider valueConverterProvider,
|
|||||||
public TableSchema create(TopicNamingStrategy topicNamingStrategy, Table table, ColumnNameFilter filter, ColumnMappers mappers, KeyMapper keysMapper) {
|
public TableSchema create(TopicNamingStrategy topicNamingStrategy, Table table, ColumnNameFilter filter, ColumnMappers mappers, KeyMapper keysMapper) {
|
||||||
// Build the schemas ...
|
// Build the schemas ...
|
||||||
final TableId tableId = table.id();
|
final TableId tableId = table.id();
|
||||||
final String schemaNamePrefix = topicNamingStrategy.dataChangeTopic(tableId);
|
final String schemaNamePrefix = topicNamingStrategy.recordSchemaPrefix(tableId);
|
||||||
final String envelopSchemaName = Envelope.schemaName(schemaNamePrefix);
|
final String envelopeSchemaPrefix = topicNamingStrategy.dataChangeTopic(tableId);
|
||||||
|
final String envelopSchemaName = Envelope.schemaName(envelopeSchemaPrefix);
|
||||||
LOGGER.debug("Mapping table '{}' to schemas under '{}'", tableId, schemaNamePrefix);
|
LOGGER.debug("Mapping table '{}' to schemas under '{}'", tableId, schemaNamePrefix);
|
||||||
SchemaBuilder valSchemaBuilder = SchemaBuilder.struct().name(schemaNameAdjuster.adjust(schemaNamePrefix + ".Value"));
|
SchemaBuilder valSchemaBuilder = SchemaBuilder.struct().name(schemaNameAdjuster.adjust(schemaNamePrefix + ".Value"));
|
||||||
SchemaBuilder keySchemaBuilder = SchemaBuilder.struct().name(schemaNameAdjuster.adjust(schemaNamePrefix + ".Key"));
|
SchemaBuilder keySchemaBuilder = SchemaBuilder.struct().name(schemaNameAdjuster.adjust(schemaNamePrefix + ".Key"));
|
||||||
|
Loading…
Reference in New Issue
Block a user