模块化mq connector (#2562)

* add connector module, add kafka producer connector module for server * add rocketMQ producer connector module for server * add rabbitMQ producer connector module for server * optimize code, add annotation * add canal message consumer connector for canal-adapter module
2026-11-04 00:30:15 +08:00 · 2020-02-28 11:00:08 +08:00
parent 4fc7e5f7d9
commit 265a8e8594
82 changed files with 4932 additions and 2412 deletions
--- a/connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/config/KafkaConstants.java
+++ b/connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/config/KafkaConstants.java
@@ -0,0 +1,16 @@
+package com.alibaba.otter.canal.connector.kafka.config;
+
+/**
+ * Kafka 配置常量类
+ *
+ * @author rewerma 2020-01-27
+ * @version 1.0.0
+ */
+public class KafkaConstants {
+
+    public static final String ROOT                              = "canal";
+
+    public static final String CANAL_MQ_KAFKA_KERBEROS_ENABLE    = ROOT + "." + "mq.kafka.kerberos.enable";
+    public static final String CANAL_MQ_KAFKA_KERBEROS_KRB5_FILE = ROOT + "." + "mq.kafka.kerberos.krb5.file";
+    public static final String CANAL_MQ_KAFKA_KERBEROS_JAAS_FILE = ROOT + "." + "mq.kafka.kerberos.jaas.file";
+}
--- a/connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/config/KafkaProducerConfig.java
+++ b/connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/config/KafkaProducerConfig.java
@@ -0,0 +1,53 @@
+package com.alibaba.otter.canal.connector.kafka.config;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import com.alibaba.otter.canal.connector.core.config.MQProperties;
+
+/**
+ * Kafka producer 配置类
+ *
+ * @author rewerma 2020-01-27
+ * @version 1.0.0
+ */
+public class KafkaProducerConfig extends MQProperties {
+
+    private Map<String, Object> kafkaProperties = new LinkedHashMap<>();
+
+    private boolean             kerberosEnabled         = false;
+    private String              krb5File;
+    private String              jaasFile;
+
+    public Map<String, Object> getKafkaProperties() {
+        return kafkaProperties;
+    }
+
+    public void setKafkaProperties(Map<String, Object> kafkaProperties) {
+        this.kafkaProperties = kafkaProperties;
+    }
+
+    public boolean isKerberosEnabled() {
+        return kerberosEnabled;
+    }
+
+    public void setKerberosEnabled(boolean kerberosEnabled) {
+        this.kerberosEnabled = kerberosEnabled;
+    }
+
+    public String getKrb5File() {
+        return krb5File;
+    }
+
+    public void setKrb5File(String krb5File) {
+        this.krb5File = krb5File;
+    }
+
+    public String getJaasFile() {
+        return jaasFile;
+    }
+
+    public void setJaasFile(String jaasFile) {
+        this.jaasFile = jaasFile;
+    }
+}
--- a/connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/consumer/CanalKafkaConsumer.java
+++ b/connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/consumer/CanalKafkaConsumer.java
@@ -0,0 +1,137 @@
+package com.alibaba.otter.canal.connector.kafka.consumer;
+
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.otter.canal.connector.core.config.CanalConstants;
+import com.alibaba.otter.canal.connector.core.spi.SPI;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.StringDeserializer;
+
+import com.alibaba.otter.canal.connector.core.consumer.CommonMessage;
+import com.alibaba.otter.canal.connector.core.spi.CanalMsgConsumer;
+import com.alibaba.otter.canal.connector.core.util.MessageUtil;
+import com.alibaba.otter.canal.protocol.Message;
+
+/**
+ * Kafka consumer SPI 实现
+ *
+ * @author rewerma @ 2020-02-01
+ * @version 1.0.0
+ */
+@SPI("kafka")
+public class CanalKafkaConsumer implements CanalMsgConsumer {
+
+    private static final String      PREFIX_KAFKA_CONFIG = "kafka.";
+
+    private KafkaConsumer<String, ?> kafkaConsumer;
+    private boolean                  flatMessage         = true;
+    private String                   topic;
+
+    private Map<Integer, Long>       currentOffsets      = new ConcurrentHashMap<>();
+    private Properties               kafkaProperties     = new Properties();
+
+    @Override
+    public void init(Properties properties, String topic, String groupId) {
+        this.topic = topic;
+
+        Boolean flatMessage = (Boolean) properties.get(CanalConstants.CANAL_MQ_FLAT_MESSAGE);
+        if (flatMessage != null) {
+            this.flatMessage = flatMessage;
+        }
+        for (Map.Entry<Object, Object> entry : properties.entrySet()) {
+            String k = (String) entry.getKey();
+            Object v = entry.getValue();
+            if (k.startsWith(PREFIX_KAFKA_CONFIG) && v != null) {
+                kafkaProperties.put(k.substring(PREFIX_KAFKA_CONFIG.length()), v);
+            }
+        }
+        kafkaProperties.put("group.id", groupId);
+        kafkaProperties.put("key.deserializer", StringDeserializer.class);
+        kafkaProperties.put("client.id", UUID.randomUUID().toString().substring(0, 6));
+    }
+
+    @Override
+    public void connect() {
+        if (this.flatMessage) {
+            kafkaProperties.put("value.deserializer", StringDeserializer.class);
+            this.kafkaConsumer = new KafkaConsumer<String, String>(kafkaProperties);
+        } else {
+            kafkaProperties.put("value.deserializer", KafkaMessageDeserializer.class);
+            this.kafkaConsumer = new KafkaConsumer<String, Message>(kafkaProperties);
+        }
+        kafkaConsumer.subscribe(Collections.singletonList(topic));
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public List<CommonMessage> getMessage(Long timeout, TimeUnit unit) {
+        if (!flatMessage) {
+            ConsumerRecords<String, Message> records = (ConsumerRecords<String, Message>) kafkaConsumer
+                .poll(unit.toMillis(timeout));
+            if (!records.isEmpty()) {
+                currentOffsets.clear();
+                List<CommonMessage> messages = new ArrayList<>();
+                for (ConsumerRecord<String, Message> record : records) {
+                    if (currentOffsets.get(record.partition()) == null) {
+                        currentOffsets.put(record.partition(), record.offset());
+                    }
+                    messages.addAll(MessageUtil.convert(record.value()));
+                }
+                return messages;
+            }
+        } else {
+            ConsumerRecords<String, String> records = (ConsumerRecords<String, String>) kafkaConsumer
+                .poll(unit.toMillis(timeout));
+
+            if (!records.isEmpty()) {
+                List<CommonMessage> messages = new ArrayList<>();
+                currentOffsets.clear();
+                for (ConsumerRecord<String, String> record : records) {
+                    if (currentOffsets.get(record.partition()) == null) {
+                        currentOffsets.put(record.partition(), record.offset());
+                    }
+                    String flatMessageJson = record.value();
+                    CommonMessage flatMessages = JSON.parseObject(flatMessageJson, CommonMessage.class);
+                    messages.add(flatMessages);
+                }
+                return messages;
+            }
+        }
+        return null;
+    }
+
+    @Override
+    public void rollback() {
+        // 回滚所有分区
+        if (kafkaConsumer != null) {
+            for (Map.Entry<Integer, Long> entry : currentOffsets.entrySet()) {
+                kafkaConsumer.seek(new TopicPartition(topic, entry.getKey()), currentOffsets.get(entry.getKey()));
+                kafkaConsumer.commitSync();
+            }
+        }
+    }
+
+    @Override
+    public void ack() {
+        if (kafkaConsumer != null) {
+            kafkaConsumer.commitSync();
+        }
+    }
+
+    @Override
+    public void disconnect() {
+        if (kafkaConsumer != null) {
+            kafkaConsumer.unsubscribe();
+        }
+        if (kafkaConsumer != null) {
+            kafkaConsumer.close();
+            kafkaConsumer = null;
+        }
+    }
+}
--- a/connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/consumer/KafkaMessageDeserializer.java
+++ b/connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/consumer/KafkaMessageDeserializer.java
@@ -0,0 +1,31 @@
+package com.alibaba.otter.canal.connector.kafka.consumer;
+
+import java.util.Map;
+
+import com.alibaba.otter.canal.connector.core.util.CanalMessageSerializerUtil;
+import org.apache.kafka.common.serialization.Deserializer;
+
+import com.alibaba.otter.canal.protocol.Message;
+
+/**
+ * Kafka Message类的反序列化
+ *
+ * @author rewerma @ 2018-6-12
+ * @version 1.0.0
+ */
+public class KafkaMessageDeserializer implements Deserializer<Message> {
+
+    @Override
+    public void configure(Map<String, ?> configs, boolean isKey) {
+    }
+
+    @Override
+    public Message deserialize(String topic1, byte[] data) {
+        return CanalMessageSerializerUtil.deserializer(data);
+    }
+
+    @Override
+    public void close() {
+        // nothing to do
+    }
+}
--- a/connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/producer/CanalKafkaProducer.java
+++ b/connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/producer/CanalKafkaProducer.java
@@ -0,0 +1,254 @@
+package com.alibaba.otter.canal.connector.kafka.producer;
+
+import java.io.File;
+import java.util.*;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+import com.alibaba.otter.canal.connector.kafka.config.KafkaConstants;
+import org.apache.commons.lang.StringUtils;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.serializer.SerializerFeature;
+import com.alibaba.otter.canal.common.utils.ExecutorTemplate;
+import com.alibaba.otter.canal.connector.core.producer.AbstractMQProducer;
+import com.alibaba.otter.canal.connector.core.util.Callback;
+import com.alibaba.otter.canal.connector.core.producer.MQDestination;
+import com.alibaba.otter.canal.connector.core.producer.MQMessageUtils;
+import com.alibaba.otter.canal.connector.core.producer.MQMessageUtils.EntryRowData;
+import com.alibaba.otter.canal.connector.core.spi.CanalMQProducer;
+import com.alibaba.otter.canal.connector.core.spi.SPI;
+import com.alibaba.otter.canal.connector.core.util.CanalMessageSerializerUtil;
+import com.alibaba.otter.canal.connector.kafka.config.KafkaProducerConfig;
+import com.alibaba.otter.canal.protocol.FlatMessage;
+import com.alibaba.otter.canal.protocol.Message;
+
+/**
+ * kafka producer SPI 实现
+ *
+ * @author rewerma 2018-6-11 下午05:30:49
+ * @version 1.0.0
+ */
+@SuppressWarnings({ "rawtypes", "unchecked" })
+@SPI("kafka")
+public class CanalKafkaProducer extends AbstractMQProducer implements CanalMQProducer {
+
+    private static final Logger      logger              = LoggerFactory.getLogger(CanalKafkaProducer.class);
+
+    private static final String      PREFIX_KAFKA_CONFIG = "kafka.";
+
+    private Producer<String, byte[]> producer;
+
+    @Override
+    public void init(Properties properties) {
+        KafkaProducerConfig kafkaProducerConfig = new KafkaProducerConfig();
+        this.mqProperties = kafkaProducerConfig;
+        super.init(properties);
+        // load properties
+        this.loadKafkaProperties(properties);
+
+        Properties kafkaProperties = new Properties();
+        kafkaProperties.putAll(kafkaProducerConfig.getKafkaProperties());
+        kafkaProperties.put("key.serializer", StringSerializer.class);
+        if (kafkaProducerConfig.isKerberosEnabled()) {
+            File krb5File = new File(kafkaProducerConfig.getKrb5File());
+            File jaasFile = new File(kafkaProducerConfig.getJaasFile());
+            if (krb5File.exists() && jaasFile.exists()) {
+                // 配置kerberos认证，需要使用绝对路径
+                System.setProperty("java.security.krb5.conf", krb5File.getAbsolutePath());
+                System.setProperty("java.security.auth.login.config", jaasFile.getAbsolutePath());
+                System.setProperty("javax.security.auth.useSubjectCredsOnly", "false");
+                kafkaProperties.put("security.protocol", "SASL_PLAINTEXT");
+                kafkaProperties.put("sasl.kerberos.service.name", "kafka");
+            } else {
+                String errorMsg = "ERROR # The kafka kerberos configuration file does not exist! please check it";
+                logger.error(errorMsg);
+                throw new RuntimeException(errorMsg);
+            }
+        }
+        kafkaProperties.put("value.serializer", KafkaMessageSerializer.class);
+        producer = new KafkaProducer<>(kafkaProperties);
+    }
+
+    private void loadKafkaProperties(Properties properties) {
+        KafkaProducerConfig kafkaProducerConfig = (KafkaProducerConfig) this.mqProperties;
+        Map<String, Object> kafkaProperties = kafkaProducerConfig.getKafkaProperties();
+        for (Map.Entry<Object, Object> entry : properties.entrySet()) {
+            String key = (String) entry.getKey();
+            Object value = entry.getValue();
+            if (key.startsWith(PREFIX_KAFKA_CONFIG) && value != null) {
+                key = key.substring(PREFIX_KAFKA_CONFIG.length());
+                kafkaProperties.put(key, value);
+            }
+        }
+
+        String kerberosEnabled = properties.getProperty(KafkaConstants.CANAL_MQ_KAFKA_KERBEROS_ENABLE);
+        if (!StringUtils.isEmpty(kerberosEnabled)) {
+            kafkaProducerConfig.setKerberosEnabled(Boolean.parseBoolean(kerberosEnabled));
+        }
+        String krb5File = properties.getProperty(KafkaConstants.CANAL_MQ_KAFKA_KERBEROS_KRB5_FILE);
+        if (!StringUtils.isEmpty(krb5File)) {
+            kafkaProducerConfig.setKrb5File(krb5File);
+        }
+        String jaasFile = properties.getProperty(KafkaConstants.CANAL_MQ_KAFKA_KERBEROS_JAAS_FILE);
+        if (!StringUtils.isEmpty(jaasFile)) {
+            kafkaProducerConfig.setJaasFile(jaasFile);
+        }
+    }
+
+    @Override
+    public void stop() {
+        try {
+            logger.info("## stop the kafka producer");
+            if (producer != null) {
+                producer.close();
+            }
+            super.stop();
+        } catch (Throwable e) {
+            logger.warn("##something goes wrong when stopping kafka producer:", e);
+        } finally {
+            logger.info("## kafka producer is down.");
+        }
+    }
+
+    @Override
+    public void send(MQDestination mqDestination, Message message, Callback callback) {
+        ExecutorTemplate template = new ExecutorTemplate(executor);
+
+        try {
+            List result;
+            if (!StringUtils.isEmpty(mqDestination.getDynamicTopic())) {
+                // 动态topic路由计算,只是基于schema/table,不涉及proto数据反序列化
+                Map<String, Message> messageMap = MQMessageUtils
+                    .messageTopics(message, mqDestination.getTopic(), mqDestination.getDynamicTopic());
+
+                // 针对不同的topic,引入多线程提升效率
+                for (Map.Entry<String, Message> entry : messageMap.entrySet()) {
+                    final String topicName = entry.getKey().replace('.', '_');
+                    final Message messageSub = entry.getValue();
+                    template.submit((Callable) () -> {
+                        try {
+                            return send(mqDestination, topicName, messageSub, mqProperties.isFlatMessage());
+                        } catch (Exception e) {
+                            throw new RuntimeException(e);
+                        }
+                    });
+                }
+
+                result = template.waitForResult();
+            } else {
+                result = new ArrayList();
+                List<Future> futures = send(mqDestination,
+                    mqDestination.getTopic(),
+                    message,
+                    mqProperties.isFlatMessage());
+                result.add(futures);
+            }
+
+            // 一个批次的所有topic和分区的队列，都采用异步的模式进行多线程批量发送
+            // 最后在集结点进行flush等待，确保所有数据都写出成功
+            // 注意：kafka的异步模式如果要保证顺序性，需要设置max.in.flight.requests.per.connection=1，确保在网络异常重试时有排他性
+            producer.flush();
+            // flush操作也有可能是发送失败,这里需要异步关注一下发送结果,针对有异常的直接出发rollback
+            for (Object obj : result) {
+                List<Future> futures = (List<Future>) obj;
+                for (Future future : futures) {
+                    try {
+                        future.get();
+                    } catch (InterruptedException | ExecutionException e) {
+                        throw new RuntimeException(e);
+                    }
+                }
+            }
+
+            callback.commit();
+        } catch (Throwable e) {
+            logger.error(e.getMessage(), e);
+            callback.rollback();
+        } finally {
+            template.clear();
+        }
+    }
+
+    private List<Future> send(MQDestination mqDestination, String topicName, Message message, boolean flat) {
+        List<ProducerRecord<String, byte[]>> records = new ArrayList<>();
+        if (!flat) {
+            if (mqDestination.getPartitionHash() != null && !mqDestination.getPartitionHash().isEmpty()) {
+                // 并发构造
+                EntryRowData[] datas = MQMessageUtils.buildMessageData(message, executor);
+                // 串行分区
+                Message[] messages = MQMessageUtils.messagePartition(datas,
+                    message.getId(),
+                    mqDestination.getPartitionsNum(),
+                    mqDestination.getPartitionHash(),
+                    this.mqProperties.isDatabaseHash());
+                int length = messages.length;
+                for (int i = 0; i < length; i++) {
+                    Message messagePartition = messages[i];
+                    if (messagePartition != null) {
+                        records.add(new ProducerRecord<>(topicName,
+                            i,
+                            null,
+                            CanalMessageSerializerUtil.serializer(messagePartition, true)));
+                    }
+                }
+            } else {
+                final int partition = mqDestination.getPartition() != null ? mqDestination.getPartition() : 0;
+                records.add(new ProducerRecord<>(topicName,
+                    partition,
+                    null,
+                    CanalMessageSerializerUtil.serializer(message, true)));
+            }
+        } else {
+            // 发送扁平数据json
+            // 并发构造
+            EntryRowData[] datas = MQMessageUtils.buildMessageData(message, executor);
+            // 串行分区
+            List<FlatMessage> flatMessages = MQMessageUtils.messageConverter(datas, message.getId());
+            for (FlatMessage flatMessage : flatMessages) {
+                if (mqDestination.getPartitionHash() != null && !mqDestination.getPartitionHash().isEmpty()) {
+                    FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage,
+                        mqDestination.getPartitionsNum(),
+                        mqDestination.getPartitionHash(),
+                        this.mqProperties.isDatabaseHash());
+                    int length = partitionFlatMessage.length;
+                    for (int i = 0; i < length; i++) {
+                        FlatMessage flatMessagePart = partitionFlatMessage[i];
+                        if (flatMessagePart != null) {
+                            records.add(new ProducerRecord<>(topicName,
+                                i,
+                                null,
+                                JSON.toJSONBytes(flatMessagePart, SerializerFeature.WriteMapNullValue)));
+                        }
+                    }
+                } else {
+                    final int partition = mqDestination.getPartition() != null ? mqDestination.getPartition() : 0;
+                    records.add(new ProducerRecord<>(topicName,
+                        partition,
+                        null,
+                        JSON.toJSONBytes(flatMessage, SerializerFeature.WriteMapNullValue)));
+                }
+            }
+        }
+
+        return produce(records);
+    }
+
+    private List<Future> produce(List<ProducerRecord<String, byte[]>> records) {
+        List<Future> futures = new ArrayList<>();
+        // 异步发送，因为在partition hash的时候已经按照每个分区合并了消息，走到这一步不需要考虑单个分区内的顺序问题
+        for (ProducerRecord record : records) {
+            futures.add(producer.send(record));
+        }
+
+        return futures;
+    }
+
+}
--- a/connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/producer/KafkaMessageSerializer.java
+++ b/connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/producer/KafkaMessageSerializer.java
@@ -0,0 +1,28 @@
+package com.alibaba.otter.canal.connector.kafka.producer;
+
+import org.apache.kafka.common.serialization.Serializer;
+
+import java.util.Map;
+
+/**
+ * Kafka Message类的序列化
+ *
+ * @author rewerma 2018-6-11 下午05:30:49
+ * @version 1.0.0
+ */
+public class KafkaMessageSerializer implements Serializer<byte[]> {
+
+    @Override
+    public void configure(Map<String, ?> configs, boolean isKey) {
+    }
+
+    @Override
+    public byte[] serialize(String topic, byte[] data) {
+        return data;
+    }
+
+    @Override
+    public void close() {
+        // nothing to do
+    }
+}
--- a/connector/kafka-connector/src/main/resources/META-INF/canal/com.alibaba.otter.canal.connector.core.spi.CanalMQProducer
+++ b/connector/kafka-connector/src/main/resources/META-INF/canal/com.alibaba.otter.canal.connector.core.spi.CanalMQProducer
@@ -0,0 +1 @@
+kafka=com.alibaba.otter.canal.connector.kafka.producer.CanalKafkaProducer
--- a/connector/kafka-connector/src/main/resources/META-INF/canal/com.alibaba.otter.canal.connector.core.spi.CanalMsgConsumer
+++ b/connector/kafka-connector/src/main/resources/META-INF/canal/com.alibaba.otter.canal.connector.core.spi.CanalMsgConsumer
@@ -0,0 +1 @@
+kafka=com.alibaba.otter.canal.connector.kafka.consumer.CanalKafkaConsumer
--- a/connector/kafka-connector/src/test/java/com/alibaba/otter/canal/connector/kafka/test/CanalKafkaProducerTest.java
+++ b/connector/kafka-connector/src/test/java/com/alibaba/otter/canal/connector/kafka/test/CanalKafkaProducerTest.java
@@ -0,0 +1,29 @@
+package com.alibaba.otter.canal.connector.kafka.test;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.util.Properties;
+
+import org.junit.Ignore;
+import org.junit.Test;
+
+import com.alibaba.otter.canal.connector.core.spi.CanalMQProducer;
+import com.alibaba.otter.canal.connector.core.spi.ExtensionLoader;
+
+@Ignore
+public class CanalKafkaProducerTest {
+
+    @Test
+    public void testLoadKafkaProducer() throws IOException {
+        Properties pro = new Properties();
+        FileInputStream in = new FileInputStream("../../deployer/src/main/resources/canal.properties");
+        pro.load(in);
+
+        ExtensionLoader<CanalMQProducer> loader = ExtensionLoader.getExtensionLoader(CanalMQProducer.class);
+        CanalMQProducer canalMQProducer = loader
+            .getExtension("kafka", "/../../deployer/target/canal/plugin", "/../../deployer/target/canal/plugin");
+        canalMQProducer.init(pro);
+
+        in.close();
+    }
+}
				`@@ -0,0 +1 @@`
				`kafka=com.alibaba.otter.canal.connector.kafka.producer.CanalKafkaProducer`