[ISSUE #436]BrokerBasedLog supports initialization of compaction topic

README.md

@@ -17,6 +17,7 @@ rocketmq-connect-sample的主要作用是从源文件中读取数据发送到Roc
 2. 64bit JDK 1.8+;
 3. Maven 3.2.x或以上版本;
 4. 启动 [RocketMQ](https://rocketmq.apache.org/docs/quick-start/);
+>  支持 RocketMQ [5.1.0] 版本
 5. 创建测试Topic 
 > sh ${ROCKETMQ_HOME}/bin/mqadmin updateTopic -t fileTopic -n localhost:9876 -c DefaultCluster -r 8 -w 8
 

connectors/rocketmq-connect-activemq/pom.xml

@@ -207,7 +207,7 @@
         <dependency>
             <groupId>org.apache.rocketmq</groupId>
             <artifactId>rocketmq-openmessaging</artifactId>
-            <version>4.3.2</version>
+            <version>4.9.4</version>
         </dependency>
         <dependency>
             <groupId>commons-cli</groupId>

connectors/rocketmq-connect-cassandra/pom.xml

@@ -38,7 +38,7 @@
         <!-- Compiler settings properties -->
         <maven.compiler.source>1.8</maven.compiler.source>
         <maven.compiler.target>1.8</maven.compiler.target>
-        <rocketmq.version>4.5.2</rocketmq.version>
+        <rocketmq.version>4.9.4</rocketmq.version>
     </properties>
 
     <build>
@@ -245,7 +245,7 @@
         <dependency>
             <groupId>org.apache.rocketmq</groupId>
             <artifactId>rocketmq-openmessaging</artifactId>
-            <version>4.3.2</version>
+            <version>4.9.4</version>
         </dependency>
         <dependency>
             <groupId>commons-cli</groupId>

connectors/rocketmq-connect-debezium/pom.xml

@@ -180,8 +180,7 @@
         <debezium.version>1.7.2.Final</debezium.version>
         <debezium.postgresql.version>42.3.3</debezium.postgresql.version>
         <!--rocketmq version-->
-        <rocketmq.version>4.7.1</rocketmq.version>
-        <rocketmq-openmessaging.version>4.3.2</rocketmq-openmessaging.version>
+        <rocketmq.version>5.1.0</rocketmq.version>
 
         <!--rocketmq connect version-->
         <openmessaging-connector.version>0.1.4</openmessaging-connector.version>
@@ -327,6 +326,11 @@
                 <artifactId>logback-core</artifactId>
                 <version>1.2.9</version>
             </dependency>
+            <dependency>
+                <groupId>org.apache.rocketmq</groupId>
+                <artifactId>rocketmq-connect-common</artifactId>
+                <version>0.0.1-SNAPSHOT</version>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 

connectors/rocketmq-connect-debezium/rocketmq-connect-debezium-core/src/main/java/org/apache/rocketmq/connect/debezium/RocketMqAdminUtil.java

@@ -16,6 +16,11 @@
  */
 package org.apache.rocketmq.connect.debezium;
 
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.UUID;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.acl.common.AclClientRPCHook;
 import org.apache.rocketmq.acl.common.SessionCredentials;
@@ -24,28 +29,22 @@
 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.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;
 
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Set;
-import java.util.UUID;
-
 /**
  * Tools for creating RocketMq topic and group
  */
@@ -130,7 +129,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);
@@ -201,7 +200,7 @@ public static String createSubGroup(RocketMqConfig connectConfig, String subGrou
             SubscriptionGroupConfig initConfig = new SubscriptionGroupConfig();
             initConfig.setGroupName(subGroup);
             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);
@@ -231,9 +230,9 @@ public static Map<MessageQueue, TopicOffset> offsets(RocketMqConfig config, Stri
         DefaultMQAdminExt adminClient = null;
         try {
             adminClient = RocketMqAdminUtil.startMQAdminTool(config);
-            TopicStatsTable topicStatsTable = adminClient.examineTopicStats(topic);
+            TopicStatsTable topicStatsTable = examineTopicStats(adminClient,topic);
             return topicStatsTable.getOffsetTable();
-        } catch (MQClientException | MQBrokerException | RemotingException | InterruptedException e) {
+        } catch (Exception e) {
             throw new RuntimeException(e);
         } finally {
             if (adminClient != null) {
@@ -242,4 +241,59 @@ public static Map<MessageQueue, TopicOffset> offsets(RocketMqConfig config, Stri
         }
     }
 
+    /**
+     * Compatible with 4.9.4 and earlier
+     *
+     * @param adminClient
+     * @param topic
+     * @return
+     */
+    private static TopicStatsTable examineTopicStats(DefaultMQAdminExt adminClient, String topic) {
+        try {
+            return adminClient.examineTopicStats(topic);
+        } catch (MQBrokerException e) {
+            // Compatible with 4.9.4 and earlier
+            if (e.getResponseCode() == ResponseCode.REQUEST_CODE_NOT_SUPPORTED) {
+                try {
+                    return overrideExamineTopicStats(adminClient, topic);
+                } catch (Exception ex) {
+                    throw new RuntimeException(ex);
+                }
+            } else {
+                throw new RuntimeException(e);
+            }
+        } catch (Exception ex) {
+            throw new RuntimeException(ex);
+        }
+    }
+
+    /**
+     * Compatible with version 4.9.4
+     *
+     * @param adminClient
+     * @param topic
+     * @return
+     * @throws RemotingException
+     * @throws InterruptedException
+     * @throws MQClientException
+     * @throws MQBrokerException
+     */
+    private static TopicStatsTable overrideExamineTopicStats(DefaultMQAdminExt adminClient,
+        String topic) throws RemotingException, InterruptedException, MQClientException, MQBrokerException {
+        TopicRouteData topicRouteData = adminClient.examineTopicRouteInfo(topic);
+        TopicStatsTable topicStatsTable = new TopicStatsTable();
+        for (BrokerData bd : topicRouteData.getBrokerDatas()) {
+            String addr = bd.selectBrokerAddr();
+            if (addr != null) {
+                TopicStatsTable tst = adminClient
+                    .getDefaultMQAdminExtImpl()
+                    .getMqClientInstance()
+                    .getMQClientAPIImpl()
+                    .getTopicStatsInfo(addr, topic, 5000);
+                topicStatsTable.getOffsetTable().putAll(tst.getOffsetTable());
+            }
+        }
+        return topicStatsTable;
+    }
+
 }

connectors/rocketmq-connect-debezium/rocketmq-connect-debezium-core/src/main/java/org/apache/rocketmq/connect/debezium/RocketMqDatabaseHistory.java

@@ -34,12 +34,12 @@
 import org.apache.rocketmq.client.producer.SendCallback;
 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.message.Message;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.common.message.MessageQueue;
 import org.apache.rocketmq.connect.kafka.connect.adaptor.task.AbstractKafkaConnectSource;
 import org.apache.rocketmq.remoting.exception.RemotingException;
+import org.apache.rocketmq.remoting.protocol.admin.TopicOffset;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

connectors/rocketmq-connect-deltalake/pom.xml

@@ -15,7 +15,7 @@
         <!-- Compiler settings properties -->
         <maven.compiler.source>1.8</maven.compiler.source>
         <maven.compiler.target>1.8</maven.compiler.target>
-        <rocketmq.version>4.5.2</rocketmq.version>
+        <rocketmq.version>4.9.4</rocketmq.version>
         <avro.version>1.10.2</avro.version>
         <parquet.version>1.10.1</parquet.version>
         <scala.version>2.12</scala.version>
@@ -231,7 +231,7 @@
         <dependency>
             <groupId>org.apache.rocketmq</groupId>
             <artifactId>rocketmq-openmessaging</artifactId>
-            <version>4.3.2</version>
+            <version>4.9.4</version>
         </dependency>
 
         <dependency>

connectors/rocketmq-connect-hudi/README.md

@@ -72,7 +72,7 @@ http://${runtime-ip}:${runtime-port}/connectors/${rocketmq-hudi-connector-name}/
 * **spark-submit 启动任务**
 将connect-runtime打包后通过spark-submit提交任务
 ```
-nohup sh spark-submit 	--class org.apache.rocketmq.connect.runtime.DistributedConnectStartup --conf "spark.driver.extraJavaOptions=-Dlogback.configurationFile=logback.xml" --files /xxx/conf/connect.conf,/xxx/conf/log4j.properties  --packages org.apache.hudi:hudi-spark3-bundle_2.12:0.8.0,org.apache.spark:spark-avro_2.12:3.0.1,org.apache.hudi:hudi-java-client:0.8.0,org.apache.parquet:parquet-avro:1.10.1,org.apache.avro:avro:1.10.2,com.alibaba:fastjson:1.2.51,org.reflections:reflections:0.9.11,org.apache.httpcomponents:httpclient:4.5.5,io.openmessaging:openmessaging-connector:0.1.1,commons-cli:commons-cli:1.1,org.apache.rocketmq:rocketmq-client:4.4.0,org.apache.rocketmq:rocketmq-tools:4.4.0,org.apache.rocketmq:rocketmq-remoting:4.4.0,org.apache.rocketmq:rocketmq-openmessaging:4.3.2,org.slf4j:slf4j-api:1.7.7,com.google.guava:guava:20.0,org.apache.hadoop:hadoop-common:3.3.1,org.reflections:reflections:0.9.12,org.apache.hive:hive-exec:2.3.7 --conf 'spark.executor.userClassPathFirst=true'  --conf 'spark.driver.userClassPathFirst=true' --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' /xxx/rocketmq/rocketmq-connect-runtime-0.0.1-SNAPSHOT.jar  &
+nohup sh spark-submit 	--class org.apache.rocketmq.connect.runtime.DistributedConnectStartup --conf "spark.driver.extraJavaOptions=-Dlogback.configurationFile=logback.xml" --files /xxx/conf/connect.conf,/xxx/conf/log4j.properties  --packages org.apache.hudi:hudi-spark3-bundle_2.12:0.8.0,org.apache.spark:spark-avro_2.12:3.0.1,org.apache.hudi:hudi-java-client:0.8.0,org.apache.parquet:parquet-avro:1.10.1,org.apache.avro:avro:1.10.2,com.alibaba:fastjson:1.2.51,org.reflections:reflections:0.9.11,org.apache.httpcomponents:httpclient:4.5.5,io.openmessaging:openmessaging-connector:0.1.1,commons-cli:commons-cli:1.1,org.apache.rocketmq:rocketmq-client:4.4.0,org.apache.rocketmq:rocketmq-tools:4.4.0,org.apache.rocketmq:rocketmq-remoting:4.4.0,org.apache.rocketmq:rocketmq-openmessaging:4.9.4,org.slf4j:slf4j-api:1.7.7,com.google.guava:guava:20.0,org.apache.hadoop:hadoop-common:3.3.1,org.reflections:reflections:0.9.12,org.apache.hive:hive-exec:2.3.7 --conf 'spark.executor.userClassPathFirst=true'  --conf 'spark.driver.userClassPathFirst=true' --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' /xxx/rocketmq/rocketmq-connect-runtime-0.0.1-SNAPSHOT.jar  &
 ```
 后续操作参考rocketmq-connect-hudi启动步骤
 

connectors/rocketmq-connect-hudi/pom.xml

@@ -15,7 +15,7 @@
         <!-- Compiler settings properties -->
         <maven.compiler.source>1.8</maven.compiler.source>
         <maven.compiler.target>1.8</maven.compiler.target>
-        <rocketmq.version>4.7.1</rocketmq.version>
+        <rocketmq.version>4.9.4</rocketmq.version>
 
         <hudi.version>0.8.0</hudi.version>
         <avro.version>1.10.2</avro.version>
@@ -231,7 +231,7 @@
         <dependency>
             <groupId>org.apache.rocketmq</groupId>
             <artifactId>rocketmq-openmessaging</artifactId>
-            <version>4.3.2</version>
+            <version>4.9.4</version>
         </dependency>
         <dependency>
             <groupId>org.apache.hudi</groupId>

connectors/rocketmq-connect-jdbc/pom.xml

@@ -46,7 +46,7 @@
         <!-- Compiler settings properties -->
         <maven.compiler.source>1.8</maven.compiler.source>
         <maven.compiler.target>1.8</maven.compiler.target>
-        <rocketmq.version>4.7.1</rocketmq.version>
+        <rocketmq.version>4.9.4</rocketmq.version>
 
         <!--test jar-->
         <junit.version>4.13.1</junit.version>

connectors/rocketmq-connect-jms/pom.xml

@@ -188,7 +188,7 @@
         <dependency>
             <groupId>org.apache.rocketmq</groupId>
             <artifactId>rocketmq-openmessaging</artifactId>
-            <version>4.3.2</version>
+            <version>4.9.4</version>
         </dependency>
         <dependency>
             <groupId>commons-cli</groupId>

connectors/rocketmq-connect-kafka/pom.xml

@@ -179,7 +179,7 @@
         <dependency>
             <groupId>org.apache.rocketmq</groupId>
             <artifactId>rocketmq-openmessaging</artifactId>
-            <version>4.3.2</version>
+            <version>4.9.4</version>
         </dependency>
         <dependency>
             <groupId>com.alibaba</groupId>

connectors/rocketmq-connect-mongo/pom.xml

@@ -196,7 +196,7 @@
         <dependency>
             <groupId>org.apache.rocketmq</groupId>
             <artifactId>rocketmq-openmessaging</artifactId>
-            <version>5.1.0</version>
+            <version>4.9.4</version>
         </dependency>
         <dependency>
             <groupId>junit</groupId>

connectors/rocketmq-connect-mqtt/pom.xml

@@ -32,7 +32,7 @@
         <!-- Compiler settings properties -->
         <maven.compiler.source>1.8</maven.compiler.source>
         <maven.compiler.target>1.8</maven.compiler.target>
-        <rocketmq.version>4.7.1</rocketmq.version>
+        <rocketmq.version>4.9.4</rocketmq.version>
     </properties>
 
     <build>

connectors/rocketmq-connect-rabbitmq/pom.xml

@@ -206,7 +206,7 @@
 		<dependency>
 			<groupId>org.apache.rocketmq</groupId>
 			<artifactId>rocketmq-openmessaging</artifactId>
-			<version>4.3.2</version>
+			<version>4.9.4</version>
 		</dependency>
 		<dependency>
 			<groupId>commons-cli</groupId>

connectors/rocketmq-connect-redis/pom.xml

@@ -16,7 +16,7 @@
         <!-- Compiler settings properties -->
         <maven.compiler.source>1.8</maven.compiler.source>
         <maven.compiler.target>1.8</maven.compiler.target>
-        <rocketmq.version>4.5.2</rocketmq.version>
+        <rocketmq.version>4.9.4</rocketmq.version>
     </properties>
 
     <build>

connectors/rocketmq-replicator/pom.xml

@@ -67,7 +67,7 @@
     </build>
 
     <properties>
-        <rocketmq.version>4.7.1</rocketmq.version>
+        <rocketmq.version>4.9.4</rocketmq.version>
         <openmessaging.connector.version>0.1.3</openmessaging.connector.version>
         <junit.version>4.13.1</junit.version>
         <mockito.version>3.2.4</mockito.version>

metric-exporter/pom.xml

@@ -29,6 +29,7 @@
             <artifactId>commons-lang3</artifactId>
             <version>3.12.0</version>
         </dependency>
+
         <dependency>
             <groupId>org.apache.rocketmq</groupId>
             <artifactId>rocketmq-client</artifactId>
@@ -37,6 +38,10 @@
             <groupId>org.apache.rocketmq</groupId>
             <artifactId>rocketmq-tools</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-connect-common</artifactId>
+        </dependency>
     </dependencies>
 
     <build>