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

fixed MR

Browse Source
This commit is contained in:
jianghang.loujh
2021-10-09 15:59:12 +08:00
parent 5b6fd8094a
commit 40d840375f
4 changed files with 66 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
client/pom.xml
View File

@@ -132,6 +132,12 @@
<version>1.1.1</version> <version>1.1.1</version>
<scope>provided</scope> <scope>provided</scope>
</dependency> </dependency>
<!-- Pulsar -->
<dependency>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-client</artifactId>
<scope>provided</scope>
</dependency>
<!-- junit --> <!-- junit -->
<dependency> <dependency>
@@ -139,12 +145,6 @@
<artifactId>junit</artifactId> <artifactId>junit</artifactId>
<scope>test</scope> <scope>test</scope>
</dependency> </dependency>
<!-- Pulsar -->
<dependency>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-client</artifactId>
</dependency>
</dependencies> </dependencies>
<build> <build>

6
connector/pom.xml
View File

@@ -92,12 +92,12 @@
</repositories> </repositories>
<modules> <modules>
<module>kafka-connector</module>
<module>core</module> <module>core</module>
<module>kafka-connector</module>
<module>rocketmq-connector</module> <module>rocketmq-connector</module>
<module>rabbitmq-connector</module> <module>rabbitmq-connector</module>
<module>tcp-connector</module>
<module>pulsarmq-connector</module> <module>pulsarmq-connector</module>
<module>tcp-connector</module>
</modules> </modules>
<build> <build>
@@ -127,4 +127,4 @@
<url>https://oss.sonatype.org/service/local/staging/deploy/maven2/</url> <url>https://oss.sonatype.org/service/local/staging/deploy/maven2/</url>
</repository> </repository>
</distributionManagement> </distributionManagement>
</project> </project>

113
connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/producer/CanalRocketMQProducer.java
View File

@@ -46,17 +46,15 @@ import java.util.stream.Collectors;
* @author rewerma 2020-01-27 * @author rewerma 2020-01-27
* @version 1.0.0 * @version 1.0.0
*/ */
@SPI("rocketmq") @SPI("rocketmq") public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQProducer {
public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQProducer {
private static final Logger logger = LoggerFactory.getLogger(CanalRocketMQProducer.class); private static final Logger logger = LoggerFactory.getLogger(CanalRocketMQProducer.class);
private DefaultMQProducer defaultMQProducer; private DefaultMQProducer defaultMQProducer;
private static final String CLOUD_ACCESS_CHANNEL = "cloud"; private static final String CLOUD_ACCESS_CHANNEL = "cloud";
protected ThreadPoolExecutor sendPartitionExecutor; protected ThreadPoolExecutor sendPartitionExecutor;
@Override @Override public void init(Properties properties) {
public void init(Properties properties) {
RocketMQProducerConfig rocketMQProperties = new RocketMQProducerConfig(); RocketMQProducerConfig rocketMQProperties = new RocketMQProducerConfig();
this.mqProperties = rocketMQProperties; this.mqProperties = rocketMQProperties;
super.init(properties); super.init(properties);
@@ -71,9 +69,9 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
} }
defaultMQProducer = new DefaultMQProducer(rocketMQProperties.getProducerGroup(), defaultMQProducer = new DefaultMQProducer(rocketMQProperties.getProducerGroup(),
rpcHook, rpcHook,
rocketMQProperties.isEnableMessageTrace(), rocketMQProperties.isEnableMessageTrace(),
rocketMQProperties.getCustomizedTraceTopic()); rocketMQProperties.getCustomizedTraceTopic());
if (CLOUD_ACCESS_CHANNEL.equals(rocketMQProperties.getAccessChannel())) { if (CLOUD_ACCESS_CHANNEL.equals(rocketMQProperties.getAccessChannel())) {
defaultMQProducer.setAccessChannel(AccessChannel.CLOUD); defaultMQProducer.setAccessChannel(AccessChannel.CLOUD);
} }
@@ -92,12 +90,12 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
int parallelPartitionSendThreadSize = mqProperties.getParallelSendThreadSize(); int parallelPartitionSendThreadSize = mqProperties.getParallelSendThreadSize();
sendPartitionExecutor = new ThreadPoolExecutor(parallelPartitionSendThreadSize, sendPartitionExecutor = new ThreadPoolExecutor(parallelPartitionSendThreadSize,
parallelPartitionSendThreadSize, parallelPartitionSendThreadSize,
0, 0,
TimeUnit.SECONDS, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(parallelPartitionSendThreadSize * 2), new ArrayBlockingQueue<>(parallelPartitionSendThreadSize * 2),
new NamedThreadFactory("MQ-Parallel-Sender-Partition"), new NamedThreadFactory("MQ-Parallel-Sender-Partition"),
new ThreadPoolExecutor.CallerRunsPolicy()); new ThreadPoolExecutor.CallerRunsPolicy());
} }
private void loadRocketMQProperties(Properties properties) { private void loadRocketMQProperties(Properties properties) {
@@ -113,12 +111,12 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
rocketMQProperties.setProducerGroup(producerGroup); rocketMQProperties.setProducerGroup(producerGroup);
} }
String enableMessageTrace = PropertiesUtils.getProperty(properties, String enableMessageTrace = PropertiesUtils.getProperty(properties,
RocketMQConstants.ROCKETMQ_ENABLE_MESSAGE_TRACE); RocketMQConstants.ROCKETMQ_ENABLE_MESSAGE_TRACE);
if (!StringUtils.isEmpty(enableMessageTrace)) { if (!StringUtils.isEmpty(enableMessageTrace)) {
rocketMQProperties.setEnableMessageTrace(Boolean.parseBoolean(enableMessageTrace)); rocketMQProperties.setEnableMessageTrace(Boolean.parseBoolean(enableMessageTrace));
} }
String customizedTraceTopic = PropertiesUtils.getProperty(properties, String customizedTraceTopic = PropertiesUtils.getProperty(properties,
RocketMQConstants.ROCKETMQ_CUSTOMIZED_TRACE_TOPIC); RocketMQConstants.ROCKETMQ_CUSTOMIZED_TRACE_TOPIC);
if (!StringUtils.isEmpty(customizedTraceTopic)) { if (!StringUtils.isEmpty(customizedTraceTopic)) {
rocketMQProperties.setCustomizedTraceTopic(customizedTraceTopic); rocketMQProperties.setCustomizedTraceTopic(customizedTraceTopic);
} }
@@ -135,7 +133,7 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
rocketMQProperties.setRetryTimesWhenSendFailed(Integer.parseInt(retry)); rocketMQProperties.setRetryTimesWhenSendFailed(Integer.parseInt(retry));
} }
String vipChannelEnabled = PropertiesUtils.getProperty(properties, String vipChannelEnabled = PropertiesUtils.getProperty(properties,
RocketMQConstants.ROCKETMQ_VIP_CHANNEL_ENABLED); RocketMQConstants.ROCKETMQ_VIP_CHANNEL_ENABLED);
if (!StringUtils.isEmpty(vipChannelEnabled)) { if (!StringUtils.isEmpty(vipChannelEnabled)) {
rocketMQProperties.setVipChannelEnabled(Boolean.parseBoolean(vipChannelEnabled)); rocketMQProperties.setVipChannelEnabled(Boolean.parseBoolean(vipChannelEnabled));
} }
@@ -145,15 +143,15 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
} }
} }
@Override @Override public void send(MQDestination destination, com.alibaba.otter.canal.protocol.Message message,
public void send(MQDestination destination, com.alibaba.otter.canal.protocol.Message message, Callback callback) { Callback callback) {
ExecutorTemplate template = new ExecutorTemplate(sendExecutor); ExecutorTemplate template = new ExecutorTemplate(sendExecutor);
try { try {
if (!StringUtils.isEmpty(destination.getDynamicTopic())) { if (!StringUtils.isEmpty(destination.getDynamicTopic())) {
// 动态topic // 动态topic
Map<String, com.alibaba.otter.canal.protocol.Message> messageMap = MQMessageUtils.messageTopics(message, Map<String, com.alibaba.otter.canal.protocol.Message> messageMap = MQMessageUtils.messageTopics(message,
destination.getTopic(), destination.getTopic(),
destination.getDynamicTopic()); destination.getDynamicTopic());
for (Map.Entry<String, com.alibaba.otter.canal.protocol.Message> entry : messageMap.entrySet()) { for (Map.Entry<String, com.alibaba.otter.canal.protocol.Message> entry : messageMap.entrySet()) {
String topicName = entry.getKey().replace('.', '_'); String topicName = entry.getKey().replace('.', '_');
@@ -181,16 +179,17 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
} }
} }
public void send(final MQDestination destination, String topicName, com.alibaba.otter.canal.protocol.Message message) { public void send(final MQDestination destination, String topicName,
com.alibaba.otter.canal.protocol.Message message) {
// 获取当前topic的分区数 // 获取当前topic的分区数
Integer partitionNum = MQMessageUtils.parseDynamicTopicPartition(topicName, Integer partitionNum = MQMessageUtils.parseDynamicTopicPartition(topicName,
destination.getDynamicTopicPartitionNum()); destination.getDynamicTopicPartitionNum());
// 获取topic的队列数为分区数 // 获取topic的队列数为分区数
if(partitionNum == null){ if (partitionNum == null) {
partitionNum = getTopicDynamicQueuesSize(destination.getEnableDynamicQueuePartition(), topicName); partitionNum = getTopicDynamicQueuesSize(destination.getEnableDynamicQueuePartition(), topicName);
} }
if (partitionNum == null) { if (partitionNum == null) {
partitionNum = destination.getPartitionsNum(); partitionNum = destination.getPartitionsNum();
} }
@@ -200,10 +199,10 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
MQMessageUtils.EntryRowData[] datas = MQMessageUtils.buildMessageData(message, buildExecutor); MQMessageUtils.EntryRowData[] datas = MQMessageUtils.buildMessageData(message, buildExecutor);
// 串行分区 // 串行分区
com.alibaba.otter.canal.protocol.Message[] messages = MQMessageUtils.messagePartition(datas, com.alibaba.otter.canal.protocol.Message[] messages = MQMessageUtils.messagePartition(datas,
message.getId(), message.getId(),
partitionNum, partitionNum,
destination.getPartitionHash(), destination.getPartitionHash(),
mqProperties.isDatabaseHash()); mqProperties.isDatabaseHash());
int length = messages.length; int length = messages.length;
ExecutorTemplate template = new ExecutorTemplate(sendPartitionExecutor); ExecutorTemplate template = new ExecutorTemplate(sendPartitionExecutor);
@@ -213,9 +212,9 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
final int index = i; final int index = i;
template.submit(() -> { template.submit(() -> {
Message data = new Message(topicName, Message data = new Message(topicName,
((RocketMQProducerConfig) this.mqProperties).getTag(), ((RocketMQProducerConfig) this.mqProperties).getTag(),
CanalMessageSerializerUtil.serializer(dataPartition, CanalMessageSerializerUtil.serializer(dataPartition,
mqProperties.isFilterTransactionEntry())); mqProperties.isFilterTransactionEntry()));
sendMessage(data, index); sendMessage(data, index);
}); });
} }
@@ -225,8 +224,8 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
} else { } else {
final int partition = destination.getPartition() != null ? destination.getPartition() : 0; final int partition = destination.getPartition() != null ? destination.getPartition() : 0;
Message data = new Message(topicName, Message data = new Message(topicName,
((RocketMQProducerConfig) this.mqProperties).getTag(), ((RocketMQProducerConfig) this.mqProperties).getTag(),
CanalMessageSerializerUtil.serializer(message, mqProperties.isFilterTransactionEntry())); CanalMessageSerializerUtil.serializer(message, mqProperties.isFilterTransactionEntry()));
sendMessage(data, partition); sendMessage(data, partition);
} }
} else { } else {
@@ -243,9 +242,9 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
for (FlatMessage flatMessage : flatMessages) { for (FlatMessage flatMessage : flatMessages) {
FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage, FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage,
partitionNum, partitionNum,
destination.getPartitionHash(), destination.getPartitionHash(),
mqProperties.isDatabaseHash()); mqProperties.isDatabaseHash());
int length = partitionFlatMessage.length; int length = partitionFlatMessage.length;
for (int i = 0; i < length; i++) { for (int i = 0; i < length; i++) {
// 增加null判断,issue #3267 // 增加null判断,issue #3267
@@ -262,10 +261,10 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
final int index = i; final int index = i;
template.submit(() -> { template.submit(() -> {
List<Message> messages = flatMessagePart.stream() List<Message> messages = flatMessagePart.stream()
.map(flatMessage -> new Message(topicName, .map(flatMessage -> new Message(topicName,
((RocketMQProducerConfig) this.mqProperties).getTag(), ((RocketMQProducerConfig) this.mqProperties).getTag(),
JSON.toJSONBytes(flatMessage, SerializerFeature.WriteMapNullValue))) JSON.toJSONBytes(flatMessage, SerializerFeature.WriteMapNullValue)))
.collect(Collectors.toList()); .collect(Collectors.toList());
// 批量发送 // 批量发送
sendMessage(messages, index); sendMessage(messages, index);
}); });
@@ -277,10 +276,10 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
} else { } else {
final int partition = destination.getPartition() != null ? destination.getPartition() : 0; final int partition = destination.getPartition() != null ? destination.getPartition() : 0;
List<Message> messages = flatMessages.stream() List<Message> messages = flatMessages.stream()
.map(flatMessage -> new Message(topicName, .map(flatMessage -> new Message(topicName,
((RocketMQProducerConfig) this.mqProperties).getTag(), ((RocketMQProducerConfig) this.mqProperties).getTag(),
JSON.toJSONBytes(flatMessage, SerializerFeature.WriteMapNullValue))) JSON.toJSONBytes(flatMessage, SerializerFeature.WriteMapNullValue)))
.collect(Collectors.toList()); .collect(Collectors.toList());
// 批量发送 // 批量发送
sendMessage(messages, partition); sendMessage(messages, partition);
} }
@@ -305,8 +304,7 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
} }
} }
@SuppressWarnings("deprecation") @SuppressWarnings("deprecation") private void sendMessage(List<Message> messages, int partition) {
private void sendMessage(List<Message> messages, int partition) {
if (messages.isEmpty()) { if (messages.isEmpty()) {
return; return;
} }
@@ -348,8 +346,7 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
} }
} }
@Override @Override public void stop() {
public void stop() {
logger.info("## Stop RocketMQ producer##"); logger.info("## Stop RocketMQ producer##");
this.defaultMQProducer.shutdown(); this.defaultMQProducer.shutdown();
if (sendPartitionExecutor != null) { if (sendPartitionExecutor != null) {
@@ -358,15 +355,15 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
super.stop(); super.stop();
} }
private Integer getTopicDynamicQueuesSize(Boolean enable, String topicName){ private Integer getTopicDynamicQueuesSize(Boolean enable, String topicName) {
if(enable!=null && enable){ if (enable != null && enable) {
topicName = this.defaultMQProducer.withNamespace(topicName); topicName = this.defaultMQProducer.withNamespace(topicName);
DefaultMQProducerImpl innerProducer = this.defaultMQProducer.getDefaultMQProducerImpl(); DefaultMQProducerImpl innerProducer = this.defaultMQProducer.getDefaultMQProducerImpl();
TopicPublishInfo topicInfo = innerProducer.getTopicPublishInfoTable().get(topicName); TopicPublishInfo topicInfo = innerProducer.getTopicPublishInfoTable().get(topicName);
if(topicInfo == null){ if (topicInfo == null) {
return null; return null;
}else{ } else {
return topicInfo.getMessageQueueList().size(); return topicInfo.getMessageQueueList().size();
} }
} }

4
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java
View File

@@ -223,7 +223,7 @@ public class BinlogDownloadQueue {
TarArchiveEntry tarArchiveEntry = null; TarArchiveEntry tarArchiveEntry = null;
while ((tarArchiveEntry = tais.getNextTarEntry()) != null) { while ((tarArchiveEntry = tais.getNextTarEntry()) != null) {
String name = tarArchiveEntry.getName(); String name = tarArchiveEntry.getName();
File tarFile = new File(parentFile, name); File tarFile = new File(parentFile, name + ".tmp");
logger.info("start to download file " + tarFile.getName()); logger.info("start to download file " + tarFile.getName());
if (tarFile.exists()) { if (tarFile.exists()) {
tarFile.delete(); tarFile.delete();
@@ -244,7 +244,7 @@ public class BinlogDownloadQueue {
} }
tais.close(); tais.close();
} else { } else {
File file = new File(parentFile, fileName); File file = new File(parentFile, fileName + ".tmp");
if (file.exists()) { if (file.exists()) {
file.delete(); file.delete();
} }