【Kafka】MM2同步Kafka集群时如何自定义复制策略(ReplicationPolicy)

需求

使用MM2同步集群数据,topic名称不能变,默认的复制策略为:DefaultReplicationPolicy,这个策略会把同步至目标集群的topic都加上一个源集群别名的前缀,比如源集群别名为A,topic为:bi-log,该topic同步到目标集群后会变成:A.bi-log,为啥这么做呢,就是为了避免双向同步的场景出现死循环。

官方也给出了解释:

这是 MirrorMaker 2.0 中的默认行为,以避免在复杂的镜像拓扑中重写数据。 需要在复制流设计和主题管理方面小心自定义此项,以避免数据丢失。 可以通过对“replication.policy.class”使用自定义复制策略类来完成此操作,所以本文主要记录一下自定义复制策略的流程。

准备工作

下载源码

https://kafka.apache.org/downloads

kafka源码是使用Gradle编译的,需要安装Gradle,具体安装操作不赘述了,可以百度。

源码使用IDEA打开后,在connect模块下找到接口:org.apache.kafka.connect.mirror.ReplicationPolicy

自定义复制策略

ReplicationPolicy这个接口主要有几个方法:

  • formatRemoteTopic:重命名topic名称
  • topicSource:根据topic获取source集群别名
  • upstreamTopic:获取topic在source集群中的名称
  • originalTopic:获取topic原始的名称(针对多次同步过程中,被重命名过多次的topic)
  • isInternalTopic:判断是否为内部topic

根据我们的需求,自定义策略需要满足:

  • 不重命名source集群中topic的名称
  • 能返回source集群别名

实现很简单,就是保证topic原封不动即可,完整代码如下:

package org.apache.kafka.connect.mirror;

import org.apache.kafka.common.Configurable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.regex.Pattern;

/**
 * Defines remote topics like "us-west.topic1". The separator is customizable and defaults to a period.
 */
public class CustomReplicationPolicy implements ReplicationPolicy, Configurable {

    // In order to work with various metrics stores, we allow custom separators.
    public static final String SEPARATOR_CONFIG = MirrorClientConfig.REPLICATION_POLICY_SEPARATOR;
    public static final String SEPARATOR_DEFAULT = ".";
    private static final Logger log = LoggerFactory.getLogger(CustomReplicationPolicy.class);
    private String separator = SEPARATOR_DEFAULT;
    private Pattern separatorPattern = Pattern.compile(Pattern.quote(SEPARATOR_DEFAULT));

    @Override
    public void configure(Map<String, ?> props) {
        if (props.containsKey(SEPARATOR_CONFIG)) {
            separator = (String) props.get(SEPARATOR_CONFIG);
            log.info("Using custom remote topic separator: '{}'", separator);
            separatorPattern = Pattern.compile(Pattern.quote(separator));
        }
    }

    /**
     * 拼接Topic名(if you need)
     *
     * @param sourceClusterAlias 源集群标识
     * @param topic              源Topic名称
     * @return java.lang.String
     * @date 2023/03/03 4:28 下午
     */
    @Override
    public String formatRemoteTopic(String sourceClusterAlias, String topic) {
        return topic;
    }

    /**
     * 获取源集群标(source.cluster.alias)
     *
     * @param topic Topic nameMirrorSourceConnector
     * @return source alias
     */
    @Override
    public String topicSource(String topic) {
        // 和source.cluster.alias配置的一致,可通过读取配置,为了方便直接返回
        return "source";
    }

    /**
     * 截取上游真实Topic名称
     *
     * @param topic Topic name
     * @return java.lang.String
     * @date 2023/03/03 4:22 下午
     */
    @Override
    public String upstreamTopic(String topic) {
        return topic;
    }

    /**
     * 获取原始Topic名,没做过加工,直接返回即可
     *
     * @param topic 源Topic名
     * @return java.lang.String
     * @date 2023/03/03 6:42 下午
     */
    @Override
    public String originalTopic(String topic) {
        return topic;
    }
}

还需要修改一个地方:org.apache.kafka.connect.mirror.MirrorSourceConnector#isCycle

这个方法是判断是否出现循环复制,会递归调用,如果不修改会死循环:

原始代码:

修改为:

    // Recurse upstream to detect cycles, i.e. whether this topic is already on the target cluster
    boolean isCycle(String topic) {
        String source = replicationPolicy.topicSource(topic);
        if (source == null) {
            return false;
        } else {
            return source.equals(sourceAndTarget.target());
        }
    }

不改的话,后果如下:

编译代码

只需要编译connect模块即可,从Gradle视图中找到对应模块的build方法,修改参数,跳过单元测试(不跳过的话电脑得卡死):

build完成之后,在项目目录下找到对应jar文件,用这两个jar文件替换掉你执行脚本所使用kafka的libs目录下的jar即可:(原jar文件记得备份,以防万一)

libs目录示例:

完成上述操作之后,修改MM2配置文件中的:

replication.policy.class=org.apache.kafka.connect.mirror.CustomReplicationPolicy

还有一种方法是直接将class文件上传到classpath下,这种方式我没试。

再次执行脚本,可以看到同步后的topic已经保持原来的名称了,大功告成!