diff --git a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQConstants.java b/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQConstants.java
index 55c8d80b..1c9c1a8e 100644
--- a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQConstants.java
+++ b/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQConstants.java
@@ -1,7 +1,5 @@
package com.alibaba.otter.canal.connector.pulsarmq.config;
-import org.joda.time.Seconds;
-
/**
* PulsarMQ配置
*
@@ -10,19 +8,20 @@ import org.joda.time.Seconds;
* @since 1 by chad at 2021/9/15 新增配置文件
*/
public class PulsarMQConstants {
- public static final String ROOT = "pulsarmq";
+
+ public static final String ROOT = "pulsarmq";
/**
* pulsar服务连接地址
*/
- public static final String PULSARMQ_SERVER_URL = ROOT + "." + "serverUrl";
+ public static final String PULSARMQ_SERVER_URL = ROOT + "." + "serverUrl";
/**
* pulsar服务角色token，需要有对应token的生产者权限
*/
- public static final String PULSARMQ_ROLE_TOKEN = ROOT + "." + "roleToken";
+ public static final String PULSARMQ_ROLE_TOKEN = ROOT + "." + "roleToken";
/**
* topic前缀
*/
- public static final String PULSARMQ_TOPIC_TENANT_PREFIX = ROOT + "." + "topicTenantPrefix";
+ public static final String PULSARMQ_TOPIC_TENANT_PREFIX = ROOT + "." + "topicTenantPrefix";
/**** 消费者 *****/
/**
@@ -32,35 +31,34 @@ public class PulsarMQConstants {
/**
* 批量处理超时时间
*/
- public static final String PULSARMQ_BATCH_PROCESS_TIMEOUT = ROOT + "." + "batchProcessTimeout";
+ public static final String PULSARMQ_BATCH_PROCESS_TIMEOUT = ROOT + "." + "batchProcessTimeout";
/**
* 消费都订阅名称，将以该名称为消费者身份标识，同一个subscriptName，认为是同一个消费实例
*/
- public static final String PULSARMQ_SUBSCRIPT_NAME = ROOT + "." + "subscriptName";
+ public static final String PULSARMQ_SUBSCRIPT_NAME = ROOT + "." + "subscriptName";
/**
* 重试间隔秒数
*/
- public static final String PULSARMQ_REDELIVERY_DELAY_SECONDS = ROOT + "." + "redeliveryDelaySeconds";
+ public static final String PULSARMQ_REDELIVERY_DELAY_SECONDS = ROOT + "." + "redeliveryDelaySeconds";
/**
* ACK超时秒数
*/
- public static final String PULSARMQ_ACK_TIMEOUT_SECONDS = ROOT + "." + "ackTimeoutSeconds";
+ public static final String PULSARMQ_ACK_TIMEOUT_SECONDS = ROOT + "." + "ackTimeoutSeconds";
/**
* 是否开启消费重试
*/
- public static final String PULSARMQ_IS_RETRY = ROOT + "." + "isRetry";
+ public static final String PULSARMQ_IS_RETRY = ROOT + "." + "isRetry";
/**
* 自动生成的 retry dlq队列名称后缀是否大写
*/
- public static final String PULSARMQ_IS_RETRY_DLQ_UPPERCASE = ROOT + "." + "isRetryDLQUpperCase";
+ public static final String PULSARMQ_IS_RETRY_DLQ_UPPERCASE = ROOT + "." + "isRetryDLQUpperCase";
/**
* 最大重试次数
*/
- public static final String PULSARMQ_MAX_REDELIVERY_COUNT = ROOT + "." + "maxRedeliveryCount";
+ public static final String PULSARMQ_MAX_REDELIVERY_COUNT = ROOT + "." + "maxRedeliveryCount";
/**
* Pulsar admin服务器地址
*/
- public static final String PULSARMQ_ADMIN_SERVER_URL = ROOT + "." + "adminServerUrl";
-
+ public static final String PULSARMQ_ADMIN_SERVER_URL = ROOT + "." + "adminServerUrl";
}
diff --git a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQProducerConfig.java b/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQProducerConfig.java
index a984e79e..1ace4aa0 100644
--- a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQProducerConfig.java
+++ b/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQProducerConfig.java
@@ -4,16 +4,18 @@ import com.alibaba.otter.canal.connector.core.config.MQProperties;
/**
* Pulsar生产者配置
+ *
* @author chad
* @date 2021/9/15 11:23
* @since 1 by chad at 2021/9/15 新增
*/
public class PulsarMQProducerConfig extends MQProperties {
+
/**
* pulsar服务连接地址
*
- * broker: pulsar://localhost:6650
- * httpUrl: http://localhost:8080
+ * broker: pulsar://localhost:6650
+ * httpUrl: http://localhost:8080
*
*/
private String serverUrl;
diff --git a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumer.java b/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumer.java
index 5f3e4ee6..0a535468 100644
--- a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumer.java
+++ b/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumer.java
@@ -1,20 +1,24 @@
package com.alibaba.otter.canal.connector.pulsarmq.consumer;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.pulsar.client.api.*;
+
import com.alibaba.fastjson2.JSON;
import com.alibaba.otter.canal.common.utils.MQUtil;
import com.alibaba.otter.canal.connector.core.config.CanalConstants;
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.spi.SPI;
+import com.alibaba.otter.canal.connector.core.util.CanalMessageSerializerUtil;
+import com.alibaba.otter.canal.connector.core.util.MessageUtil;
import com.alibaba.otter.canal.connector.pulsarmq.config.PulsarMQConstants;
+import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import com.google.common.collect.Lists;
-import org.apache.commons.lang.StringUtils;
-import org.apache.pulsar.client.api.*;
-
-import java.util.List;
-import java.util.Properties;
-import java.util.concurrent.TimeUnit;
/**
* Pulsar consumer SPI 实现
@@ -24,19 +28,20 @@ import java.util.concurrent.TimeUnit;
*/
@SPI("pulsarmq")
public class CanalPulsarMQConsumer implements CanalMsgConsumer {
+
/**
* 连接pulsar客户端
*/
- private PulsarClient pulsarClient;
- private Consumer pulsarMQConsumer;
+ private PulsarClient pulsarClient;
+ private Consumer pulsarMQConsumer;
/**
* 是否为扁平消息
*/
- private boolean flatMessage = false;
+ private boolean flatMessage = false;
/**
* 主题名称
*/
- private String topic;
+ private String topic;
/**
* 单线程控制
*/
@@ -44,58 +49,57 @@ public class CanalPulsarMQConsumer implements CanalMsgConsumer {
/**
* 环境连接URL
*/
- private String serviceUrl;
+ private String serviceUrl;
/**
* 角色认证token
*/
- private String roleToken;
+ private String roleToken;
/**
* 订阅客户端名称
*/
- private String subscriptName;
+ private String subscriptName;
/**
* 每次批量获取数据的最大条目数，默认30
*/
- private int batchSize = 30;
+ private int batchSize = 30;
/**
- * 与{@code batchSize}一起决定批量获取的数据大小
- * 当：
+ * 与{@code batchSize}一起决定批量获取的数据大小 当：
*
* 1. {@code batchSize} 条消息未消费时
* 2. 距上一次批量消费时间达到{@code batchTimeoutSeconds}秒时
*
* 任一条件满足，即执行批量消费
*/
- private int getBatchTimeoutSeconds = 30;
+ private int getBatchTimeoutSeconds = 30;
/**
* 批量处理消息时，一次批量处理的超时时间
*
* 该时间应该根据{@code batchSize}和{@code batchTimeoutSeconds}合理设置
*
*/
- private long batchProcessTimeout = 60 * 1000;
+ private long batchProcessTimeout = 60 * 1000;
/**
* 消费失败后的重试秒数，默认60秒
*/
- private int redeliveryDelaySeconds = 60;
+ private int redeliveryDelaySeconds = 60;
/**
* 当客户端接收到消息，30秒还没有返回ack给服务端时，ack超时，会重新消费该消息
*/
- private int ackTimeoutSeconds = 30;
+ private int ackTimeoutSeconds = 30;
/**
* 是否开启消息失败重试功能，默认开启
*/
- private boolean isRetry = true;
+ private boolean isRetry = true;
/**
*
* true重试(-RETRY)和死信队列(-DLQ)后缀为大写，有些地方创建的为小写，需确保正确
*
*/
- private boolean isRetryDLQUpperCase = false;
+ private boolean isRetryDLQUpperCase = false;
/**
* 最大重试次数
*/
- private int maxRedeliveryCount = 128;
+ private int maxRedeliveryCount = 128;
@Override
public void init(Properties properties, String topic, String groupId) {
@@ -106,9 +110,12 @@ public class CanalPulsarMQConsumer implements CanalMsgConsumer {
}
this.serviceUrl = properties.getProperty(PulsarMQConstants.PULSARMQ_SERVER_URL);
this.roleToken = properties.getProperty(PulsarMQConstants.PULSARMQ_ROLE_TOKEN);
- //this.subscriptName = properties.getProperty(PulsarMQConstants.PULSARMQ_SUBSCRIPT_NAME);
+ this.subscriptName = properties.getProperty(PulsarMQConstants.PULSARMQ_SUBSCRIPT_NAME);
// 采用groupId作为subscriptName，避免所有的都是同一个订阅者名称
- this.subscriptName = groupId;
+ if (StringUtils.isEmpty(this.subscriptName)) {
+ this.subscriptName = groupId;
+ }
+
if (StringUtils.isEmpty(this.subscriptName)) {
throw new RuntimeException("Pulsar Consumer subscriptName required");
}
@@ -153,9 +160,9 @@ public class CanalPulsarMQConsumer implements CanalMsgConsumer {
}
// 连接创建客户端
try {
- //AuthenticationDataProvider
+ // AuthenticationDataProvider
ClientBuilder builder = PulsarClient.builder().serviceUrl(serviceUrl);
- if(StringUtils.isNotEmpty(roleToken)) {
+ if (StringUtils.isNotEmpty(roleToken)) {
builder.authentication(AuthenticationFactory.token(roleToken));
}
pulsarClient = builder.build();
@@ -174,15 +181,14 @@ public class CanalPulsarMQConsumer implements CanalMsgConsumer {
builder.subscriptionType(SubscriptionType.Failover);
builder
- // 调用consumer.negativeAcknowledge(message) （即nack）来表示消费失败的消息
- // 在指定的时间进行重新消费，默认是1分钟。
- .negativeAckRedeliveryDelay(this.redeliveryDelaySeconds, TimeUnit.SECONDS)
- .subscriptionName(this.subscriptName)
- ;
+ // 调用consumer.negativeAcknowledge(message) （即nack）来表示消费失败的消息
+ // 在指定的时间进行重新消费，默认是1分钟。
+ .negativeAckRedeliveryDelay(this.redeliveryDelaySeconds, TimeUnit.SECONDS)
+ .subscriptionName(this.subscriptName);
if (this.isRetry) {
DeadLetterPolicy.DeadLetterPolicyBuilder dlqBuilder = DeadLetterPolicy.builder()
- // 最大重试次数
- .maxRedeliverCount(this.maxRedeliveryCount);
+ // 最大重试次数
+ .maxRedeliverCount(this.maxRedeliveryCount);
// 指定重试队列，不是多个或通配符topic才能判断重试队列
if (!MQUtil.isPatternTag(this.topic)) {
String retryTopic = this.topic + (this.isRetryDLQUpperCase ? "-RETRY" : "-retry");
@@ -191,19 +197,17 @@ public class CanalPulsarMQConsumer implements CanalMsgConsumer {
dlqBuilder.deadLetterTopic(dlqTopic);
}
- //默认关闭，如果需要重试则开启
- builder.enableRetry(true)
- .deadLetterPolicy(dlqBuilder.build());
+ // 默认关闭，如果需要重试则开启
+ builder.enableRetry(true).deadLetterPolicy(dlqBuilder.build());
}
// ack超时
builder.ackTimeout(this.ackTimeoutSeconds, TimeUnit.SECONDS);
// pulsar批量获取消息设置
- builder.batchReceivePolicy(new BatchReceivePolicy.Builder()
- .maxNumMessages(this.batchSize)
- .timeout(this.getBatchTimeoutSeconds, TimeUnit.SECONDS)
- .build());
+ builder.batchReceivePolicy(new BatchReceivePolicy.Builder().maxNumMessages(this.batchSize)
+ .timeout(this.getBatchTimeoutSeconds, TimeUnit.SECONDS)
+ .build());
try {
this.pulsarMQConsumer = builder.subscribe();
@@ -225,17 +229,14 @@ public class CanalPulsarMQConsumer implements CanalMsgConsumer {
this.lastGetBatchMessage = messages;
for (org.apache.pulsar.client.api.Message msg : messages) {
byte[] data = msg.getData();
- /*if (!this.flatMessage) {
+ if (!this.flatMessage) {
Message message = CanalMessageSerializerUtil.deserializer(data);
List list = MessageUtil.convert(message);
messageList.addAll(list);
} else {
CommonMessage commonMessage = JSON.parseObject(data, CommonMessage.class);
messageList.add(commonMessage);
- }*/
- // CanalMessageSerializerUtil.deserializer(data) 会转换失败
- CommonMessage commonMessage = JSON.parseObject(data, CommonMessage.class);
- messageList.add(commonMessage);
+ }
}
} catch (PulsarClientException e) {
throw new CanalClientException("Receive pulsar batch message error", e);
diff --git a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/producer/CanalPulsarMQProducer.java b/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/producer/CanalPulsarMQProducer.java
index 631f208a..1b488935 100644
--- a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/producer/CanalPulsarMQProducer.java
+++ b/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/producer/CanalPulsarMQProducer.java
@@ -1,5 +1,17 @@
package com.alibaba.otter.canal.connector.pulsarmq.producer;
+import java.util.*;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONWriter.Feature;
import com.alibaba.otter.canal.common.utils.ExecutorTemplate;
@@ -16,19 +28,6 @@ import com.alibaba.otter.canal.connector.pulsarmq.config.PulsarMQConstants;
import com.alibaba.otter.canal.connector.pulsarmq.config.PulsarMQProducerConfig;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.FlatMessage;
-import org.apache.commons.lang.StringUtils;
-import org.apache.pulsar.client.admin.PulsarAdmin;
-import org.apache.pulsar.client.admin.PulsarAdminException;
-import org.apache.pulsar.client.api.*;
-import org.apache.pulsar.shade.com.google.gson.JsonParser;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.*;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
/**
* PulsarMQ Producer SPI 实现
@@ -39,23 +38,22 @@ import java.util.stream.Collectors;
@SPI("pulsarmq")
public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQProducer {
- private static final Logger logger = LoggerFactory.getLogger(CanalPulsarMQProducer.class);
-
- private static final Map> PRODUCERS = new HashMap<>();
-
- protected ThreadPoolExecutor sendPartitionExecutor;
/**
* 消息体分区属性名称
*/
- public static final String MSG_PROPERTY_PARTITION_NAME = "partitionNum";
+ public static final String MSG_PROPERTY_PARTITION_NAME = "partitionNum";
+ private static final Logger logger = LoggerFactory
+ .getLogger(CanalPulsarMQProducer.class);
+ private static final Map> PRODUCERS = new HashMap<>();
+ protected ThreadPoolExecutor sendPartitionExecutor;
/**
* pulsar客户端，管理连接
*/
- protected PulsarClient client;
+ protected PulsarClient client;
/**
* Pulsar admin 客户端
*/
- protected PulsarAdmin pulsarAdmin;
+ protected PulsarAdmin pulsarAdmin;
@Override
public void init(Properties properties) {
@@ -68,9 +66,9 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
// 初始化连接客户端
try {
ClientBuilder builder = PulsarClient.builder()
- // 填写pulsar的连接地址
- .serviceUrl(pulsarMQProducerConfig.getServerUrl());
- if(StringUtils.isNotEmpty(pulsarMQProducerConfig.getRoleToken())) {
+ // 填写pulsar的连接地址
+ .serviceUrl(pulsarMQProducerConfig.getServerUrl());
+ if (StringUtils.isNotEmpty(pulsarMQProducerConfig.getRoleToken())) {
// 角色权限认证的token
builder.authentication(AuthenticationFactory.token(pulsarMQProducerConfig.getRoleToken()));
}
@@ -80,7 +78,7 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
}
// 初始化Pulsar admin
- if(StringUtils.isNotEmpty(pulsarMQProducerConfig.getAdminServerUrl())) {
+ if (StringUtils.isNotEmpty(pulsarMQProducerConfig.getAdminServerUrl())) {
try {
pulsarAdmin = PulsarAdmin.builder().serviceHttpUrl(pulsarMQProducerConfig.getAdminServerUrl()).build();
} catch (PulsarClientException e) {
@@ -91,12 +89,12 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
// 加载所有生产者 --> topic可能有正则或表名，无法确认所有topic，在使用时再加载
int parallelPartitionSendThreadSize = mqProperties.getParallelSendThreadSize();
sendPartitionExecutor = new ThreadPoolExecutor(parallelPartitionSendThreadSize,
- parallelPartitionSendThreadSize,
- 0,
- TimeUnit.SECONDS,
- new ArrayBlockingQueue<>(parallelPartitionSendThreadSize * 2),
- new NamedThreadFactory("MQ-Parallel-Sender-Partition"),
- new ThreadPoolExecutor.CallerRunsPolicy());
+ parallelPartitionSendThreadSize,
+ 0,
+ TimeUnit.SECONDS,
+ new ArrayBlockingQueue<>(parallelPartitionSendThreadSize * 2),
+ new NamedThreadFactory("MQ-Parallel-Sender-Partition"),
+ new ThreadPoolExecutor.CallerRunsPolicy());
}
/**
@@ -119,12 +117,13 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
if (!StringUtils.isEmpty(roleToken)) {
tmpProperties.setRoleToken(roleToken);
}
- String topicTenantPrefix = PropertiesUtils.getProperty(properties, PulsarMQConstants.PULSARMQ_TOPIC_TENANT_PREFIX);
+ String topicTenantPrefix = PropertiesUtils.getProperty(properties,
+ PulsarMQConstants.PULSARMQ_TOPIC_TENANT_PREFIX);
if (!StringUtils.isEmpty(topicTenantPrefix)) {
tmpProperties.setTopicTenantPrefix(topicTenantPrefix);
}
String adminServerUrl = PropertiesUtils.getProperty(properties, PulsarMQConstants.PULSARMQ_ADMIN_SERVER_URL);
- if(!StringUtils.isEmpty(adminServerUrl)) {
+ if (!StringUtils.isEmpty(adminServerUrl)) {
tmpProperties.setAdminServerUrl(adminServerUrl);
}
if (logger.isDebugEnabled()) {
@@ -140,8 +139,8 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
*
*
* @param destination 消息目标信息
- * @param message 消息
- * @param callback 消息发送结果回调
+ * @param message 消息
+ * @param callback 消息发送结果回调
* @return void
* @date 2021/9/2 22:01
* @author chad
@@ -154,9 +153,8 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
try {
if (!StringUtils.isEmpty(destination.getDynamicTopic())) {
// 动态topic
- Map messageMap = MQMessageUtils.messageTopics(message,
- destination.getTopic(),
- destination.getDynamicTopic());
+ Map messageMap = MQMessageUtils
+ .messageTopics(message, destination.getTopic(), destination.getDynamicTopic());
for (Map.Entry entry : messageMap.entrySet()) {
String topicName = entry.getKey().replace('.', '_');
@@ -195,16 +193,17 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
* @author chad
* @since 1.0.0 by chad at 2021/9/2: 新增
*/
- 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的分区数
Integer partitionNum = MQMessageUtils.parseDynamicTopicPartition(topicName,
- destination.getDynamicTopicPartitionNum());
+ destination.getDynamicTopicPartitionNum());
if (partitionNum == null) {
partitionNum = destination.getPartitionsNum();
}
// 创建多分区topic
- if(pulsarAdmin!=null && partitionNum!=null && partitionNum>0 && PRODUCERS.get(topicName)==null) {
+ if (pulsarAdmin != null && partitionNum != null && partitionNum > 0 && PRODUCERS.get(topicName) == null) {
createMultipleTopic(topicName, partitionNum);
}
@@ -221,10 +220,10 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
}
// 串行分区
com.alibaba.otter.canal.protocol.Message[] messages = MQMessageUtils.messagePartition(datas,
- message.getId(),
- partitionNum,
- destination.getPartitionHash(),
- mqProperties.isDatabaseHash());
+ message.getId(),
+ partitionNum,
+ destination.getPartitionHash(),
+ mqProperties.isDatabaseHash());
// 发送
int len = messages.length;
for (int i = 0; i < len; i++) {
@@ -254,9 +253,9 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
for (FlatMessage flatMessage : flatMessages) {
FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage,
- partitionNum,
- destination.getPartitionHash(),
- mqProperties.isDatabaseHash());
+ partitionNum,
+ destination.getPartitionHash(),
+ mqProperties.isDatabaseHash());
int length = partitionFlatMessage.length;
for (int i = 0; i < length; i++) {
// 增加null判断,issue #3267
@@ -290,9 +289,9 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
/**
* 发送原始消息，需要做分区处理
*
- * @param topic topic
+ * @param topic topic
* @param partitionNum 目标分区
- * @param msg 原始消息内容
+ * @param msg 原始消息内容
* @return void
* @date 2021/9/10 17:55
* @author chad
@@ -303,8 +302,9 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
byte[] msgBytes = CanalMessageSerializerUtil.serializer(msg, mqProperties.isFilterTransactionEntry());
try {
MessageId msgResultId = producer.newMessage()
- .property(MSG_PROPERTY_PARTITION_NAME, String.valueOf(partitionNum))
- .value(msgBytes).send();
+ .property(MSG_PROPERTY_PARTITION_NAME, String.valueOf(partitionNum))
+ .value(msgBytes)
+ .send();
// todo 判断发送结果
if (logger.isDebugEnabled()) {
logger.debug("Send Message to topic:{} Result: {}", topic, msgResultId);
@@ -317,7 +317,7 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
/**
* 发送扁平消息
*
- * @param topic topic主题
+ * @param topic topic主题
* @param flatMessages 扁平消息
* @return void
* @date 2021/9/10 18:22
@@ -328,13 +328,12 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
Producer producer = getProducer(topic);
for (FlatMessage f : flatMessages) {
try {
- MessageId msgResultId = producer
- .newMessage()
- .property(MSG_PROPERTY_PARTITION_NAME, String.valueOf(partition))
- .value(JSON.toJSONBytes(f, Feature.WriteNulls))
- .send()
- //
- ;
+ MessageId msgResultId = producer.newMessage()
+ .property(MSG_PROPERTY_PARTITION_NAME, String.valueOf(partition))
+ .value(JSON.toJSONBytes(f, Feature.WriteNulls))
+ .send()
+ //
+ ;
if (logger.isDebugEnabled()) {
logger.debug("Send Messages to topic:{} Result: {}", topic, msgResultId);
}
@@ -346,10 +345,11 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
/**
* 创建多分区topic
+ *
* @param topic
* @param partitionNum
*/
- private void createMultipleTopic(String topic,Integer partitionNum) {
+ private void createMultipleTopic(String topic, Integer partitionNum) {
// 拼接topic前缀
PulsarMQProducerConfig pulsarMQProperties = (PulsarMQProducerConfig) this.mqProperties;
String prefix = pulsarMQProperties.getTopicTenantPrefix();
@@ -368,8 +368,10 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
// TODO 无论是否报错，都继续后续的操作，此处不进行阻塞
}
}
+
/**
* 获取topic
+ *
* @param topic
* @return
*/
@@ -396,10 +398,10 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
// 创建指定topic的生产者
producer = client.newProducer()
- .topic(fullTopic)
- // 指定路由器
- .messageRouter(new MessageRouterImpl(topic))
- .create();
+ .topic(fullTopic)
+ // 指定路由器
+ .messageRouter(new MessageRouterImpl(topic))
+ .create();
// 放入缓存
PRODUCERS.put(topic, producer);
}
@@ -412,6 +414,26 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
return producer;
}
+ @Override
+ public void stop() {
+ logger.info("## Stop PulsarMQ producer##");
+
+ for (Producer p : PRODUCERS.values()) {
+ try {
+ if (null != p && p.isConnected()) {
+ p.close();
+ }
+ } catch (PulsarClientException e) {
+ logger.warn("close producer name: {}, topic: {}, error: {}",
+ p.getProducerName(),
+ p.getTopic(),
+ e.getMessage());
+ }
+ }
+
+ super.stop();
+ }
+
/**
* Pulsar自定义路由策略
*
@@ -421,9 +443,10 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
* @since 2 by chad at 2021/9/17 修改为msg自带目标分区
*/
private static class MessageRouterImpl implements MessageRouter {
+
private String topicLocal;
- public MessageRouterImpl(String topicLocal) {
+ public MessageRouterImpl(String topicLocal){
this.topicLocal = topicLocal;
}
@@ -435,7 +458,8 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
try {
partition = Integer.parseInt(partitionStr);
} catch (NumberFormatException e) {
- logger.warn("Parse msg {} property failed for value: {}", MSG_PROPERTY_PARTITION_NAME, partitionStr);
+ logger
+ .warn("Parse msg {} property failed for value: {}", MSG_PROPERTY_PARTITION_NAME, partitionStr);
}
}
// topic创建时设置的分区数
@@ -447,21 +471,4 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
return partition;
}
}
-
- @Override
- public void stop() {
- logger.info("## Stop PulsarMQ producer##");
-
- for (Producer p : PRODUCERS.values()) {
- try {
- if (null != p && p.isConnected()) {
- p.close();
- }
- } catch (PulsarClientException e) {
- logger.warn("close producer name: {}, topic: {}, error: {}", p.getProducerName(), p.getTopic(), e.getMessage());
- }
- }
-
- super.stop();
- }
}
diff --git a/connector/pulsarmq-connector/src/test/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumerTest.java b/connector/pulsarmq-connector/src/test/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumerTest.java
index 760d5b1b..b1a8b2ce 100644
--- a/connector/pulsarmq-connector/src/test/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumerTest.java
+++ b/connector/pulsarmq-connector/src/test/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumerTest.java
@@ -1,23 +1,19 @@
package com.alibaba.otter.canal.connector.pulsarmq.consumer;
+import java.time.LocalTime;
+import java.time.format.DateTimeFormatter;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
import com.alibaba.fastjson2.JSON;
import com.alibaba.otter.canal.connector.core.config.CanalConstants;
import com.alibaba.otter.canal.connector.core.consumer.CommonMessage;
import com.alibaba.otter.canal.connector.pulsarmq.config.PulsarMQConstants;
-import org.apache.commons.lang.StringUtils;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-
-import java.time.LocalDateTime;
-import java.time.LocalTime;
-import java.time.format.DateTimeFormatter;
-import java.util.List;
-import java.util.Properties;
-import java.util.UUID;
-import java.util.concurrent.TimeUnit;
-
-import static org.junit.Assert.*;
/**
* @author chad
@@ -26,7 +22,7 @@ import static org.junit.Assert.*;
*/
public class CanalPulsarMQConsumerTest {
- private Properties properties;
+ private Properties properties;
private CanalPulsarMQConsumer consumer;
@@ -73,4 +69,4 @@ public class CanalPulsarMQConsumerTest {
consumer.ack();
}
}
-}
\ No newline at end of file
+}