[jira] [Updated] (KAFKA-13691) Rename target topic to custom name.

["Omnia Ibrahim (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/KAFKA-13691?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Omnia Ibrahim updated KAFKA-13691:
----------------------------------
    Labels: need-kip pull-request-available  (was: pull-request-available)

> Rename target topic to custom name.  
> -------------------------------------
>
>                 Key: KAFKA-13691
>                 URL: https://issues.apache.org/jira/browse/KAFKA-13691
>             Project: Kafka
>          Issue Type: Improvement
>          Components: config, mirrormaker
>    Affects Versions: 2.4.1, 2.7.2
>            Reporter: Eugeniu Cararus
>            Priority: Minor
>              Labels: need-kip, pull-request-available
>   Original Estimate: 1h
>  Remaining Estimate: 1h
>
> When I use mirrormaker it replicates the topics with different name, name of target is built based on
> {code:java}
> sourceClusterAlias + separator + topic  {code}
>  
> implementation is [here|https://github.com/apache/kafka/blob/trunk/connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/DefaultReplicationPolicy.java#L49].
> Proposed improvement consist in:
> Adding new config property key: target.replication.policy.topic_map
> E.I. 
> {code:java}
> source-> target.replication.policy.class= org.apache.kafka.connect.mirror.RenameTopicReplicationPolicy
> source->target.replication.policy.separator = .
> source->target.replication.policy.topic_map= SOURCER_TOPIC_A,DESTINATION_TOPIC_A; SOURCER_TOPIC_B,DESTINATION_TOPIC_B; {code}
> and add new RenameTopicReplicationPolicy which will be responsible to:
> 1. override configure method load target.replication.policy.topic_map into a Map where Key is source topic name and value is target topic name 
>  
>  2. override formatRemoteTopic so the target topic name will be loaded based on config
> {code:java}
> @Override
> public String formatRemoteTopic(String sourceClusterAlias, String topic) {
>     String targetTopic = topicMap.containsKey(topic) ?  topicMap.get(topic) : topic ;
>     return super.formatRemoteTopic(sourceClusterAlias,targetTopic);
> } {code}



--
This message was sent by Atlassian Jira
(v8.20.7#820007)