chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

merge code and fixed issue #2755 #2756 #2735 #2714 #2711 #2710 #2616 #2602 #2585 #2565 #2935 #2434 #2623 #2418 #2387

Browse Source
This commit is contained in:
agapple
2020-08-22 13:07:24 +08:00
parent 90bf9e6de5
commit b79af46054
66 changed files with 459 additions and 477 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/config/KafkaConstants.java
View File

@@ -8,9 +8,9 @@ package com.alibaba.otter.canal.connector.kafka.config;
*/
public class KafkaConstants {
public static final String ROOT = "canal";
public static final String ROOT = "kafka";
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";
public static final String CANAL_MQ_KAFKA_KERBEROS_ENABLE = ROOT + "." + "kerberos.enable";
public static final String CANAL_MQ_KAFKA_KERBEROS_KRB5_FILE = ROOT + "." + "kerberos.krb5.file";
public static final String CANAL_MQ_KAFKA_KERBEROS_JAAS_FILE = ROOT + "." + "kerberos.jaas.file";
}

26
connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/producer/CanalKafkaProducer.java
View File

@@ -59,6 +59,7 @@ public class CanalKafkaProducer extends AbstractMQProducer implements CanalMQPro
Properties kafkaProperties = new Properties();
kafkaProperties.putAll(kafkaProducerConfig.getKafkaProperties());
kafkaProperties.put("max.in.flight.requests.per.connection", 1);
kafkaProperties.put("key.serializer", StringSerializer.class);
if (kafkaProducerConfig.isKerberosEnabled()) {
File krb5File = new File(kafkaProducerConfig.getKrb5File());
@@ -83,6 +84,19 @@ public class CanalKafkaProducer extends AbstractMQProducer implements CanalMQPro
private void loadKafkaProperties(Properties properties) {
KafkaProducerConfig kafkaProducerConfig = (KafkaProducerConfig) this.mqProperties;
Map<String, Object> kafkaProperties = kafkaProducerConfig.getKafkaProperties();
// 兼容下<=1.1.4的mq配置
doMoreCompatibleConvert("canal.mq.servers", "kafka.bootstrap.servers", properties);
doMoreCompatibleConvert("canal.mq.acks", "kafka.acks", properties);
doMoreCompatibleConvert("canal.mq.compressionType", "kafka.compression.type", properties);
doMoreCompatibleConvert("canal.mq.retries", "kafka.retries", properties);
doMoreCompatibleConvert("canal.mq.batchSize", "kafka.batch.size", properties);
doMoreCompatibleConvert("canal.mq.lingerMs", "kafka.linger.ms", properties);
doMoreCompatibleConvert("canal.mq.maxRequestSize", "kafka.max.request.size", properties);
doMoreCompatibleConvert("canal.mq.bufferMemory", "kafka.buffer.memory", properties);
doMoreCompatibleConvert("canal.mq.kafka.kerberos.enable", "kafka.kerberos.enable", properties);
doMoreCompatibleConvert("canal.mq.kafka.kerberos.krb5.file", "kafka.kerberos.krb5.file", properties);
doMoreCompatibleConvert("canal.mq.kafka.kerberos.jaas.file", "kafka.kerberos.jaas.file", properties);
for (Map.Entry<Object, Object> entry : properties.entrySet()) {
String key = (String) entry.getKey();
Object value = entry.getValue();
@@ -91,7 +105,6 @@ public class CanalKafkaProducer extends AbstractMQProducer implements CanalMQPro
kafkaProperties.put(key, value);
}
}
String kerberosEnabled = properties.getProperty(KafkaConstants.CANAL_MQ_KAFKA_KERBEROS_ENABLE);
if (!StringUtils.isEmpty(kerberosEnabled)) {
kafkaProducerConfig.setKerberosEnabled(Boolean.parseBoolean(kerberosEnabled));
@@ -123,7 +136,7 @@ public class CanalKafkaProducer extends AbstractMQProducer implements CanalMQPro
@Override
public void send(MQDestination mqDestination, Message message, Callback callback) {
ExecutorTemplate template = new ExecutorTemplate(executor);
ExecutorTemplate template = new ExecutorTemplate(sendExecutor);
try {
List result;
@@ -186,7 +199,7 @@ public class CanalKafkaProducer extends AbstractMQProducer implements CanalMQPro
if (!flat) {
if (mqDestination.getPartitionHash() != null && !mqDestination.getPartitionHash().isEmpty()) {
// 并发构造
EntryRowData[] datas = MQMessageUtils.buildMessageData(message, executor);
EntryRowData[] datas = MQMessageUtils.buildMessageData(message, buildExecutor);
// 串行分区
Message[] messages = MQMessageUtils.messagePartition(datas,
message.getId(),
@@ -200,7 +213,8 @@ public class CanalKafkaProducer extends AbstractMQProducer implements CanalMQPro
records.add(new ProducerRecord<>(topicName,
i,
null,
CanalMessageSerializerUtil.serializer(messagePartition, true)));
CanalMessageSerializerUtil.serializer(messagePartition,
mqProperties.isFilterTransactionEntry())));
}
}
} else {
@@ -208,12 +222,12 @@ public class CanalKafkaProducer extends AbstractMQProducer implements CanalMQPro
records.add(new ProducerRecord<>(topicName,
partition,
null,
CanalMessageSerializerUtil.serializer(message, true)));
CanalMessageSerializerUtil.serializer(message, mqProperties.isFilterTransactionEntry())));
}
} else {
// 发送扁平数据json
// 并发构造
EntryRowData[] datas = MQMessageUtils.buildMessageData(message, executor);
EntryRowData[] datas = MQMessageUtils.buildMessageData(message, buildExecutor);
// 串行分区
List<FlatMessage> flatMessages = MQMessageUtils.messageConverter(datas, message.getId());
for (FlatMessage flatMessage : flatMessages) {