alexey.rassokhin/tet123
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

DBZ-7525 Update to RocketMQ 5.1.4

Browse Source
This commit is contained in:
Vojtech Juranek 2024-02-20 17:01:20 +01:00 committed by Jiri Pechanec
parent 0c34c6ac68
commit 09f3746e56
5 changed files with 23 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
debezium-bom/pom.xml
View File

@ -479,6 +479,12 @@
<version>${version.rocketmq}</version>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-remoting</artifactId>
<version>${version.rocketmq}</version>
</dependency>
<!-- S3 Storage -->
<dependency>
<groupId>software.amazon.awssdk</groupId>

5
debezium-storage/debezium-storage-rocketmq/pom.xml
View File

@ -42,6 +42,11 @@
<artifactId>rocketmq-tools</artifactId>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-remoting</artifactId>
</dependency>
<!-- Testing -->
<dependency>
<groupId>junit</groupId>

21
debezium-storage/debezium-storage-rocketmq/src/main/java/io/debezium/storage/rocketmq/RocketMqAdminUtil.java
View File

@ -5,7 +5,6 @@
*/
package io.debezium.storage.rocketmq;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Objects;
@ -20,19 +19,19 @@
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.admin.TopicOffset;
import org.apache.rocketmq.common.admin.TopicStatsTable;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.body.SubscriptionGroupWrapper;
import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.remoting.protocol.LanguageCode;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.admin.TopicOffset;
import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.apache.rocketmq.tools.command.CommandUtil;
@ -108,7 +107,7 @@ public static void createTopic(RocketMqConfig config, TopicConfig topicConfig) {
try {
defaultMQAdminExt = startMqAdminTool(config);
ClusterInfo clusterInfo = defaultMQAdminExt.examineBrokerClusterInfo();
HashMap<String, Set<String>> clusterAddrTable = clusterInfo.getClusterAddrTable();
Map<String, Set<String>> clusterAddrTable = clusterInfo.getClusterAddrTable();
Set<String> clusterNameSet = clusterAddrTable.keySet();
for (String clusterName : clusterNameSet) {
Set<String> masterSet = CommandUtil.fetchMasterAddrByClusterName(defaultMQAdminExt, clusterName);
@ -188,7 +187,7 @@ public static String createGroup(RocketMqConfig connectConfig, String group) {
SubscriptionGroupConfig initConfig = new SubscriptionGroupConfig();
initConfig.setGroupName(group);
ClusterInfo clusterInfo = defaultMQAdminExt.examineBrokerClusterInfo();
HashMap<String, Set<String>> clusterAddrTable = clusterInfo.getClusterAddrTable();
Map<String, Set<String>> clusterAddrTable = clusterInfo.getClusterAddrTable();
Set<String> clusterNameSet = clusterAddrTable.keySet();
for (String clusterName : clusterNameSet) {
Set<String> masterSet = CommandUtil.fetchMasterAddrByClusterName(defaultMQAdminExt, clusterName);

2
debezium-storage/debezium-storage-rocketmq/src/main/java/io/debezium/storage/rocketmq/history/RocketMqSchemaHistory.java
View File

@ -21,12 +21,12 @@
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.admin.TopicOffset;
import org.apache.rocketmq.common.constant.PermName;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.remoting.protocol.admin.TopicOffset;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

2
pom.xml
View File

@ -157,7 +157,7 @@
<version.infinispan.protostream>4.6.2.Final</version.infinispan.protostream>
<!-- RocketMQ version for RocketMQ storage and Server sink -->
<version.rocketmq>4.9.4</version.rocketmq>
<version.rocketmq>5.1.4</version.rocketmq>
<!-- MySQL server image name -->
<mysql.server.image.source>container-registry.oracle.com/mysql/community-server</mysql.server.image.source>