Skip to content

Commit 037f251

Browse files
Git-YangGit-Yang
committed
[rocketmq-replicator] Add MetaSinkTask
Signed-off-by: zhangyang21 <zhangyang21@xiaomi.com>
1 parent 44ca34d commit 037f251

5 files changed

Lines changed: 243 additions & 5 deletions

File tree

‎rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/MetaSinkTask.java‎

Lines changed: 162 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,162 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.rocketmq.replicator;
19+
20+
import io.openmessaging.KeyValue;
21+
import io.openmessaging.connector.api.common.QueueMetaData;
22+
import io.openmessaging.connector.api.data.Field;
23+
import io.openmessaging.connector.api.data.Schema;
24+
import io.openmessaging.connector.api.data.SinkDataEntry;
25+
import io.openmessaging.connector.api.sink.SinkTask;
26+
import java.util.Collection;
27+
import java.util.HashMap;
28+
import java.util.Map;
29+
import java.util.concurrent.Executors;
30+
import java.util.concurrent.ScheduledExecutorService;
31+
import java.util.concurrent.TimeUnit;
32+
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
33+
import org.apache.rocketmq.client.exception.MQClientException;
34+
import org.apache.rocketmq.common.message.MessageQueue;
35+
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
36+
import org.apache.rocketmq.common.protocol.route.BrokerData;
37+
import org.apache.rocketmq.replicator.common.Utils;
38+
import org.apache.rocketmq.replicator.config.ConfigUtil;
39+
import org.apache.rocketmq.replicator.config.TaskConfig;
40+
import org.apache.rocketmq.replicator.schema.FieldName;
41+
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
42+
import org.slf4j.Logger;
43+
import org.slf4j.LoggerFactory;
44+
45+
public class MetaSinkTask extends SinkTask {
46+
47+
private Logger log = LoggerFactory.getLogger(MetaSinkTask.class);
48+
private final String taskId;
49+
private final TaskConfig taskConfig;
50+
private DefaultMQAdminExt targetMQAdmin;
51+
private volatile boolean started = false;
52+
private final Map<String, String> brokerRoute;
53+
private ScheduledExecutorService executor;
54+
private long refreshInterval = 10 * 1000;
55+
56+
public MetaSinkTask() {
57+
this.taskConfig = new TaskConfig();
58+
this.brokerRoute = new HashMap<>();
59+
this.taskId = Utils.createTaskId(Thread.currentThread().getName());
60+
this.executor = Executors.newSingleThreadScheduledExecutor(new BasicThreadFactory.Builder().namingPattern("MetaSinkTask-SinkWatcher-%d").daemon(true).build());
61+
}
62+
63+
@Override
64+
public void put(Collection<SinkDataEntry> sinkDataEntries) {
65+
for (SinkDataEntry dataEntry : sinkDataEntries) {
66+
if (this.started) {
67+
Schema schema = dataEntry.getSchema();
68+
Object[] payload = dataEntry.getPayload();
69+
70+
String brokerName = (String) getPayloadValue(FieldName.BROKER_NAME.getKey(), schema, payload);
71+
String topic = (String) getPayloadValue(FieldName.TOPIC.getKey(), schema, payload);
72+
Integer queueId = (Integer) getPayloadValue(FieldName.QUEUE_ID.getKey(), schema, payload);
73+
Long offset = (Long) getPayloadValue(FieldName.OFFSET.getKey(), schema, payload);
74+
String consumeGroup = (String) getPayloadValue(FieldName.CONSUME_GROUP.getKey(), schema, payload);
75+
String brokerAddr = this.brokerRoute.get(brokerName);
76+
MessageQueue queue = new MessageQueue(topic, brokerName, queueId);
77+
78+
try {
79+
this.targetMQAdmin.updateConsumeOffset(brokerAddr, consumeGroup, queue, offset);
80+
} catch (Exception e) {
81+
log.error("RocketMQ replicator mata task put error, brokerAddr={}, queue={}, offset={}", brokerAddr, queue, offset, e);
82+
throw new RuntimeException(e);
83+
}
84+
}
85+
}
86+
87+
if (!this.started) {
88+
log.error("RocketMQ replicator mata sink task is not started.");
89+
throw new RuntimeException("RocketMQ replicator mata sink task is not started.");
90+
}
91+
}
92+
93+
private Object getPayloadValue(String fieldName, Schema schema, Object[] payload) {
94+
Field field = schema.getField(fieldName);
95+
if (field == null && payload.length > field.getIndex()) {
96+
return null;
97+
}
98+
99+
return payload[field.getIndex()];
100+
}
101+
102+
@Override
103+
public void commit(Map<QueueMetaData, Long> map) {
104+
}
105+
106+
@Override
107+
public void start(KeyValue config) {
108+
ConfigUtil.load(config, this.taskConfig);
109+
try {
110+
this.targetMQAdmin = Utils.startTargetMQAdminTool(this.taskConfig);
111+
} catch (MQClientException e) {
112+
log.error("RocketMQ mata sink task start error. taskId={}, taskConfig={}", this.taskId, this.taskConfig, e);
113+
throw new RuntimeException(String.format("RocketMQ MetaSinkTask start error. taskId=%s", this.taskId));
114+
}
115+
buildBrokerRoute();
116+
this.executor.scheduleAtFixedRate(this::buildBrokerRoute, refreshInterval, refreshInterval, TimeUnit.MILLISECONDS);
117+
this.started = true;
118+
log.info("RocketMQ mata sink task started");
119+
}
120+
121+
@Override
122+
public void stop() {
123+
if (this.started) {
124+
this.started = false;
125+
this.executor.shutdown();
126+
this.targetMQAdmin.shutdown();
127+
log.info("RocketMQ mata sink task stop. taskId={}", this.taskId);
128+
}
129+
}
130+
131+
@Override
132+
public void pause() {
133+
134+
}
135+
136+
@Override
137+
public void resume() {
138+
139+
}
140+
141+
private void buildBrokerRoute() {
142+
ClusterInfo clusterInfo = null;
143+
try {
144+
clusterInfo = this.targetMQAdmin.examineBrokerClusterInfo();
145+
} catch (Exception e) {
146+
log.error("RocketMQ Replicator build broker route error for `examineBrokerClusterInfo`", e);
147+
}
148+
149+
if (clusterInfo == null) {
150+
return;
151+
}
152+
153+
Map<String, BrokerData> brokerDataTable = clusterInfo.getBrokerAddrTable();
154+
if (brokerDataTable == null) {
155+
return;
156+
}
157+
158+
for (BrokerData brokerData : brokerDataTable.values()) {
159+
this.brokerRoute.put(brokerData.getBrokerName(), brokerData.selectBrokerAddr());
160+
}
161+
}
162+
}

‎rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/MetaSourceTask.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -68,7 +68,7 @@ public void start(KeyValue config) {
6868
ConfigUtil.load(config, this.config);
6969

7070
try {
71-
this.srcMQAdminExt = Utils.startMQAdminTool(this.config);
71+
this.srcMQAdminExt = Utils.startSrcMQAdminTool(this.config);
7272
} catch (MQClientException e) {
7373
log.error("Replicator task start failed for `startMQAdminTool` exception.", e);
7474
throw new IllegalStateException("Replicator task start failed for `startMQAdminTool` exception.");

‎rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/common/Utils.java‎

Lines changed: 19 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -185,15 +185,31 @@ public static DefaultMQAdminExt startTargetMQAdminTool(
185185
return targetMQAdminExt;
186186
}
187187

188-
public static DefaultMQAdminExt startMQAdminTool(TaskConfig taskConfig) throws MQClientException {
188+
public static DefaultMQAdminExt startSrcMQAdminTool(TaskConfig taskConfig) throws MQClientException {
189189
RPCHook rpcHook = null;
190190
if (taskConfig.isSrcAclEnable()) {
191191
rpcHook = new AclClientRPCHook(new SessionCredentials(taskConfig.getSrcAccessKey(), taskConfig.getSrcSecretKey()));
192192
}
193+
DefaultMQAdminExt srcMQAdminExt = new DefaultMQAdminExt(rpcHook);
194+
srcMQAdminExt.setNamesrvAddr(taskConfig.getSourceRocketmq());
195+
srcMQAdminExt.setAdminExtGroup(ConstDefine.REPLICATOR_TASK_ADMIN_GROUP);
196+
srcMQAdminExt.setInstanceName(Utils.createUniqIntanceName(taskConfig.getSourceRocketmq()));
197+
198+
srcMQAdminExt.start();
199+
log.info("SOURCE: RocketMQ srcMQAdminExt started.");
200+
201+
return srcMQAdminExt;
202+
}
203+
204+
public static DefaultMQAdminExt startTargetMQAdminTool(TaskConfig taskConfig) throws MQClientException {
205+
RPCHook rpcHook = null;
206+
if (taskConfig.isTargetAclEnable()) {
207+
rpcHook = new AclClientRPCHook(new SessionCredentials(taskConfig.getTargetAccessKey(), taskConfig.getTargetSecretKey()));
208+
}
193209
DefaultMQAdminExt targetMQAdminExt = new DefaultMQAdminExt(rpcHook);
194-
targetMQAdminExt.setNamesrvAddr(taskConfig.getSourceRocketmq());
210+
targetMQAdminExt.setNamesrvAddr(taskConfig.getTargetRocketmq());
195211
targetMQAdminExt.setAdminExtGroup(ConstDefine.REPLICATOR_TASK_ADMIN_GROUP);
196-
targetMQAdminExt.setInstanceName(Utils.createUniqIntanceName(taskConfig.getSourceRocketmq()));
212+
targetMQAdminExt.setInstanceName(Utils.createUniqIntanceName(taskConfig.getTargetRocketmq()));
197213

198214
targetMQAdminExt.start();
199215
log.info("TARGET: RocketMQ targetMQAdminExt started.");

‎rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/config/TaskConfig.java‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,14 @@ public class TaskConfig {
3131
private String srcAccessKey;
3232
private String srcSecretKey;
3333

34+
private String targetRocketmq;
35+
private boolean targetAclEnable = false;
36+
private String targetAccessKey;
37+
private String targetSecretKey;
38+
private String targetCluster;
39+
40+
private String renamePattern;
41+
3442
public String getSourceGroup() {
3543
return sourceGroup;
3644
}
@@ -130,4 +138,48 @@ public String getSrcSecretKey() {
130138
public void setSrcSecretKey(String srcSecretKey) {
131139
this.srcSecretKey = srcSecretKey;
132140
}
141+
142+
public String getTargetRocketmq() {
143+
return targetRocketmq;
144+
}
145+
146+
public void setTargetRocketmq(String targetRocketmq) {
147+
this.targetRocketmq = targetRocketmq;
148+
}
149+
150+
public boolean isTargetAclEnable() {
151+
return targetAclEnable;
152+
}
153+
154+
public void setTargetAclEnable(boolean targetAclEnable) {
155+
this.targetAclEnable = targetAclEnable;
156+
}
157+
158+
public String getTargetAccessKey() {
159+
return targetAccessKey;
160+
}
161+
162+
public void setTargetAccessKey(String targetAccessKey) {
163+
this.targetAccessKey = targetAccessKey;
164+
}
165+
166+
public String getTargetSecretKey() {
167+
return targetSecretKey;
168+
}
169+
170+
public void setTargetSecretKey(String targetSecretKey) {
171+
this.targetSecretKey = targetSecretKey;
172+
}
173+
174+
public String getTargetCluster() {
175+
return targetCluster;
176+
}
177+
178+
public void setTargetCluster(String targetCluster) {
179+
this.targetCluster = targetCluster;
180+
}
181+
182+
public String getRenamePattern() {
183+
return renamePattern;
184+
}
133185
}

‎rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/schema/FieldName.java‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,15 @@
1818

1919
public enum FieldName {
2020
COMMON_MESSAGE("MessageExt"),
21-
OFFSET("Offset");
21+
OFFSET("Offset"),
22+
BROKER_NAME("BrokerName"),
23+
TOPIC("Topic"),
24+
QUEUE_ID("QueueId"),
25+
OFFSET_MESSAGE_ID("OffsetMessageId"),
26+
CLIENT_MESSAGE_ID("ClientMessageId"),
27+
TAGS("Tags"),
28+
KEYS("Keys"),
29+
CONSUME_GROUP("ConsumeGroup");
2230

2331
private String key;
2432

0 commit comments

Comments
 (0)