Skip to content

Commit 0082274

Browse files
Git-YangGit-Yang
committed
[replicator] Consistency detection of consumption offset
Signed-off-by: zhangyang21 <zhangyang21@xiaomi.com>
1 parent 037f251 commit 0082274

8 files changed

Lines changed: 356 additions & 68 deletions

File tree

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

Lines changed: 27 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@
4343
import org.apache.rocketmq.replicator.config.TaskConfig;
4444
import org.apache.rocketmq.replicator.offset.OffsetSyncStore;
4545
import org.apache.rocketmq.replicator.schema.FieldName;
46+
import org.apache.rocketmq.replicator.schema.SchemaEnum;
4647
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
4748
import org.slf4j.Logger;
4849
import org.slf4j.LoggerFactory;
@@ -74,26 +75,32 @@ public void start(KeyValue config) {
7475
throw new IllegalStateException("Replicator task start failed for `startMQAdminTool` exception.");
7576
}
7677

77-
this.store = new OffsetSyncStore(this.srcMQAdminExt, this.config);
78+
this.store = new OffsetSyncStore(this.config);
79+
this.store.start();
7880
this.started = true;
7981
}
8082

81-
@Override public void stop() {
83+
@Override
84+
public void stop() {
8285
if (started) {
8386
started = false;
8487
}
85-
srcMQAdminExt.shutdown();
88+
this.store.shutdown();
89+
this.srcMQAdminExt.shutdown();
8690
}
8791

88-
@Override public void pause() {
92+
@Override
93+
public void pause() {
8994

9095
}
9196

92-
@Override public void resume() {
97+
@Override
98+
public void resume() {
9399

94100
}
95101

96-
@Override public Collection<SourceDataEntry> poll() {
102+
@Override
103+
public Collection<SourceDataEntry> poll() {
97104
log.debug("polling...");
98105
List<String> groups = JSONObject.parseArray(this.config.getTaskGroupList(), String.class);
99106

@@ -121,33 +128,27 @@ public void start(KeyValue config) {
121128
MessageQueue mq = offsetTable.getKey();
122129
long srcOffset = offsetTable.getValue().getConsumerOffset();
123130
long targetOffset = this.store.convertTargetOffset(mq, group, srcOffset);
124-
125-
JSONObject jsonObject = new JSONObject();
126-
jsonObject.put(RmqConstants.NEXT_POSITION, srcOffset);
131+
String targetBrokerName = this.store.getTargetBrokerName(mq.getBrokerName());
132+
String targetTopic = Utils.getTargetTopic(mq.getTopic(), config.getRenamePattern());
127133

128134
Schema schema = new Schema();
129135
schema.setDataSource(this.config.getSourceRocketmq());
130-
schema.setName(mq.getTopic());
136+
schema.setName(SchemaEnum.OFFSET.name());
131137
schema.setFields(new ArrayList<>());
132-
schema.getFields().add(new Field(0,
133-
FieldName.OFFSET.getKey(), FieldType.INT64));
138+
schema.getFields().add(new Field(0, FieldName.OFFSET.getKey(), FieldType.INT64));
139+
schema.getFields().add(new Field(1, FieldName.BROKER_NAME.getKey(), FieldType.STRING));
140+
schema.getFields().add(new Field(2, FieldName.TOPIC.getKey(), FieldType.STRING));
141+
schema.getFields().add(new Field(3, FieldName.QUEUE_ID.getKey(), FieldType.INT64));
142+
schema.getFields().add(new Field(4, FieldName.CONSUME_GROUP.getKey(), FieldType.STRING));
134143

135144
DataEntryBuilder dataEntryBuilder = new DataEntryBuilder(schema);
136-
dataEntryBuilder.timestamp(System.currentTimeMillis())
137-
.queue(this.config.getStoreTopic())
138-
.entryType(EntryType.UPDATE);
145+
dataEntryBuilder.timestamp(System.currentTimeMillis()).entryType(EntryType.UPDATE);
139146
dataEntryBuilder.putFiled(FieldName.OFFSET.getKey(), targetOffset);
140-
141-
SourceDataEntry sourceDataEntry = dataEntryBuilder.buildSourceDataEntry(
142-
ByteBuffer.wrap(RmqConstants.getPartition(
143-
mq.getTopic(),
144-
mq.getBrokerName(),
145-
String.valueOf(mq.getQueueId())).getBytes(StandardCharsets.UTF_8)),
146-
ByteBuffer.wrap(jsonObject.toJSONString().getBytes(StandardCharsets.UTF_8))
147-
);
148-
String targetTopic = new StringBuilder().append(group).append("-").append(mq.getTopic())
149-
.append("-").append(mq.getQueueId()).toString();
150-
sourceDataEntry.setQueueName(targetTopic);
147+
dataEntryBuilder.putFiled(FieldName.BROKER_NAME.getKey(), targetBrokerName);
148+
dataEntryBuilder.putFiled(FieldName.TOPIC.getKey(), targetTopic);
149+
dataEntryBuilder.putFiled(FieldName.QUEUE_ID.getKey(), mq.getQueueId());
150+
dataEntryBuilder.putFiled(FieldName.CONSUME_GROUP.getKey(), group);
151+
SourceDataEntry sourceDataEntry = dataEntryBuilder.buildSourceDataEntry(null, null);
151152
res.add(sourceDataEntry);
152153
}
153154
}

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -91,11 +91,11 @@ public RmqMetaReplicator() {
9191
log.info("verifyAndSetConfig...");
9292
try {
9393
replicatorConfig.validate(config);
94+
this.configValid = true;
9495
} catch (IllegalArgumentException e) {
9596
return e.getMessage();
9697
}
9798
this.prepare();
98-
this.configValid = true;
9999
return "";
100100
}
101101

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

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,4 +26,8 @@ public class ConstDefine {
2626

2727
public static String REPLICATOR_TASK_ADMIN_GROUP = "REPLICATOR-TASK-ADMIN-GROUP";
2828

29+
public static String REPLICATOR_SOURCE_TASK_GROUP = "REPLICATOR-SOURCE-TASK-GROUP";
30+
31+
public static String REPLICATOR_SINK_TASK_GROUP = "REPLICATOR-SINK-TASK-GROUP";
32+
2933
}

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

Lines changed: 68 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,15 +21,21 @@
2121
import io.openmessaging.internal.DefaultKeyValue;
2222
import java.util.ArrayList;
2323
import java.util.Collections;
24+
import java.util.HashMap;
2425
import java.util.List;
26+
import java.util.Map;
2527
import java.util.Set;
2628
import java.util.UUID;
2729
import java.util.concurrent.ThreadLocalRandom;
2830
import org.apache.commons.lang3.StringUtils;
31+
import org.apache.commons.lang3.text.StrSubstitutor;
2932
import org.apache.rocketmq.acl.common.AclClientRPCHook;
3033
import org.apache.rocketmq.acl.common.SessionCredentials;
34+
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
3135
import org.apache.rocketmq.client.exception.MQClientException;
3236
import org.apache.rocketmq.common.TopicConfig;
37+
import org.apache.rocketmq.common.message.MessageClientExt;
38+
import org.apache.rocketmq.common.message.MessageExt;
3339
import org.apache.rocketmq.common.protocol.route.BrokerData;
3440
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
3541
import org.apache.rocketmq.remoting.RPCHook;
@@ -38,6 +44,7 @@
3844
import org.apache.rocketmq.replicator.config.RmqConnectorConfig;
3945
import org.apache.rocketmq.replicator.config.TaskConfig;
4046
import org.apache.rocketmq.replicator.config.TaskConfigEnum;
47+
import org.apache.rocketmq.replicator.config.TaskDivideConfig;
4148
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
4249
import org.apache.rocketmq.tools.command.CommandUtil;
4350
import org.slf4j.Logger;
@@ -118,8 +125,9 @@ public static void createTopic(DefaultMQAdminExt defaultMQAdminExt, TopicConfig
118125
}
119126

120127
public static List<KeyValue> groupPartitions(List<String> elements, int numGroups, RmqConnectorConfig tdc) {
121-
if (numGroups <= 0)
128+
if (numGroups <= 0) {
122129
throw new IllegalArgumentException("Number of groups must be positive.");
130+
}
123131

124132
List<KeyValue> result = new ArrayList<>(numGroups);
125133

@@ -143,6 +151,17 @@ public static List<KeyValue> groupPartitions(List<String> elements, int numGroup
143151
keyValue.put(TaskConfigEnum.TASK_DATA_TYPE.getKey(), DataType.OFFSET.ordinal());
144152
keyValue.put(TaskConfigEnum.TASK_GROUP_INFO.getKey(), JSONObject.toJSONString(groupList));
145153
keyValue.put(TaskConfigEnum.TASK_SOURCE_RECORD_CONVERTER.getKey(), tdc.getConverter());
154+
155+
keyValue.put(TaskConfigEnum.TASK_SOURCE_ACL_ENABLE.getKey(), String.valueOf(tdc.isSrcAclEnable()));
156+
keyValue.put(TaskConfigEnum.TASK_SOURCE_ACCESS_KEY.getKey(), tdc.getSrcAccessKey());
157+
keyValue.put(TaskConfigEnum.TASK_SOURCE_SECRET_KEY.getKey(), tdc.getSrcSecretKey());
158+
159+
keyValue.put(TaskConfigEnum.TASK_TARGET_ROCKETMQ.getKey(), tdc.getTargetNamesrvs());
160+
keyValue.put(TaskConfigEnum.TASK_TARGET_CLUSTER.getKey(), tdc.getTargetCluster());
161+
keyValue.put(TaskConfigEnum.TASK_TARGET_ACL_ENABLE.getKey(), String.valueOf(tdc.isTargetAclEnable()));
162+
keyValue.put(TaskConfigEnum.TASK_TARGET_ACCESS_KEY.getKey(), tdc.getTargetAccessKey());
163+
keyValue.put(TaskConfigEnum.TASK_TARGET_SECRET_KEY.getKey(), tdc.getTargetSecretKey());
164+
keyValue.put(TaskConfigEnum.TASK_RENAME_PATTERN.getKey(), tdc.getRenamePattern());
146165
result.add(keyValue);
147166

148167
log.debug("allocate group partition: {}", keyValue);
@@ -216,4 +235,52 @@ public static DefaultMQAdminExt startTargetMQAdminTool(TaskConfig taskConfig) th
216235

217236
return targetMQAdminExt;
218237
}
238+
239+
public static DefaultMQPullConsumer startSrcMQPullConsumer(TaskConfig taskConfig) throws MQClientException {
240+
RPCHook rpcHook = null;
241+
if (taskConfig.isSrcAclEnable()) {
242+
rpcHook = new AclClientRPCHook(new SessionCredentials(taskConfig.getSrcAccessKey(), taskConfig.getSrcSecretKey()));
243+
}
244+
DefaultMQPullConsumer srcConsumer = new DefaultMQPullConsumer(rpcHook);
245+
srcConsumer.setNamesrvAddr(taskConfig.getSourceRocketmq());
246+
srcConsumer.setConsumerGroup(ConstDefine.REPLICATOR_SOURCE_TASK_GROUP);
247+
srcConsumer.setInstanceName(Utils.createUniqIntanceName(taskConfig.getSourceRocketmq()));
248+
249+
srcConsumer.start();
250+
log.info("SOURCE: RocketMQ srcConsumer started.");
251+
252+
return srcConsumer;
253+
}
254+
255+
public static DefaultMQPullConsumer startTargetMQPullConsumer(TaskConfig taskConfig) throws MQClientException {
256+
RPCHook rpcHook = null;
257+
if (taskConfig.isTargetAclEnable()) {
258+
rpcHook = new AclClientRPCHook(new SessionCredentials(taskConfig.getTargetAccessKey(), taskConfig.getTargetSecretKey()));
259+
}
260+
DefaultMQPullConsumer targetConsumer = new DefaultMQPullConsumer(rpcHook);
261+
targetConsumer.setNamesrvAddr(taskConfig.getTargetRocketmq());
262+
targetConsumer.setConsumerGroup(ConstDefine.REPLICATOR_SINK_TASK_GROUP);
263+
targetConsumer.setInstanceName(Utils.createUniqIntanceName(taskConfig.getTargetRocketmq()));
264+
265+
targetConsumer.start();
266+
log.info("TARGET: RocketMQ targetConsumer started.");
267+
268+
return targetConsumer;
269+
}
270+
271+
public static String getOffsetMsgId(MessageExt msg) {
272+
if (msg instanceof MessageClientExt) {
273+
return ((MessageClientExt) msg).getOffsetMsgId();
274+
}
275+
return msg.getMsgId();
276+
}
277+
278+
public static String getTargetTopic(String topic, String renamePattern) {
279+
if (StringUtils.isNotEmpty(renamePattern)) {
280+
Map<String, String> params = new HashMap<>();
281+
params.put("topic", topic);
282+
return StrSubstitutor.replace(renamePattern, params);
283+
}
284+
return topic;
285+
}
219286
}

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

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,16 @@ public enum TaskConfigEnum {
3434
TASK_SOURCE_RECORD_CONVERTER("source-record-converter"),
3535
TASK_SOURCE_ACL_ENABLE("srcAclEnable"),
3636
TASK_SOURCE_ACCESS_KEY("srcAccessKey"),
37-
TASK_SOURCE_SECRET_KEY("srcSecretKey");
37+
TASK_SOURCE_SECRET_KEY("srcSecretKey"),
38+
39+
TASK_TARGET_ROCKETMQ("targetRocketmq"),
40+
TASK_TARGET_CLUSTER("targetCluster"),
41+
TASK_TARGET_ACL_ENABLE("targetAclEnable"),
42+
TASK_TARGET_ACCESS_KEY("targetAccessKey"),
43+
TASK_TARGET_SECRET_KEY("targetSecretKey"),
44+
45+
TASK_RENAME_PATTERN("renamePattern"),
46+
;
3847

3948
private String key;
4049

‎rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/offset/OffsetSync.java‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,11 +25,11 @@ public class OffsetSync {
2525
private long srcOffset;
2626
private long targetOffset;
2727

28-
public OffsetSync(MessageQueue mq, String group, long srcOffset, long targetOffset) {
28+
public OffsetSync(MessageQueue mq, String group, long srcOffset, long targtOffset) {
2929
this.mq = mq;
3030
this.group = group;
3131
this.srcOffset = srcOffset;
32-
this.targetOffset = targetOffset;
32+
this.targetOffset = targtOffset;
3333
}
3434

3535
public void setMq(MessageQueue mq) {
@@ -72,4 +72,9 @@ public static OffsetSync decode(byte[] body) {
7272
OffsetSync sync = JSON.parseObject(body, OffsetSync.class);
7373
return sync;
7474
}
75+
76+
public void updateOffset(long srcOffset, long targtOffset) {
77+
this.srcOffset = srcOffset;
78+
this.targetOffset = targtOffset;
79+
}
7580
}

0 commit comments

Comments
 (0)