Skip to content

Commit 44ca34d

Browse files
Git-YangGit-Yang
authored
[ISSUE #821] Support auto create subGroup (#828)
* [connect] Support auto create subGroup Signed-off-by: zhangyang21 <zhangyang21@xiaomi.com> * [connect] Use a unique instance name Signed-off-by: zhangyang21 <zhangyang21@xiaomi.com>
1 parent ef45628 commit 44ca34d

9 files changed

Lines changed: 137 additions & 21 deletions

File tree

‎rocketmq-connect/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/config/ConnectConfig.java‎

Lines changed: 49 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,11 @@
2424
*/
2525
public class ConnectConfig {
2626

27+
/**
28+
* The unique ID of each worker instance in the cluster
29+
*/
30+
private String workerId;
31+
2732
/**
2833
* Storage directory for file store.
2934
*/
@@ -109,6 +114,20 @@ public void setBrokerSuspendMaxTimeMillis(int brokerSuspendMaxTimeMillis) {
109114

110115
private String secretKey;
111116

117+
private boolean autoCreateGroupEnable = false;
118+
119+
private String clusterName;
120+
121+
private String adminExtGroup = "connector-admin-group";
122+
123+
public String getWorkerId() {
124+
return workerId;
125+
}
126+
127+
public void setWorkerId(String workerId) {
128+
this.workerId = workerId;
129+
}
130+
112131
public String getNamesrvAddr() {
113132
return namesrvAddr;
114133
}
@@ -301,10 +320,34 @@ public void setSecretKey(String secretKey) {
301320
this.secretKey = secretKey;
302321
}
303322

304-
@Override
305-
public String toString() {
323+
public boolean isAutoCreateGroupEnable() {
324+
return autoCreateGroupEnable;
325+
}
326+
327+
public void setAutoCreateGroupEnable(boolean autoCreateGroupEnable) {
328+
this.autoCreateGroupEnable = autoCreateGroupEnable;
329+
}
330+
331+
public String getClusterName() {
332+
return clusterName;
333+
}
334+
335+
public void setClusterName(String clusterName) {
336+
this.clusterName = clusterName;
337+
}
338+
339+
public String getAdminExtGroup() {
340+
return adminExtGroup;
341+
}
342+
343+
public void setAdminExtGroup(String adminExtGroup) {
344+
this.adminExtGroup = adminExtGroup;
345+
}
346+
347+
@Override public String toString() {
306348
return "ConnectConfig{" +
307-
"storePathRootDir='" + storePathRootDir + '\'' +
349+
"workerId='" + workerId + '\'' +
350+
", storePathRootDir='" + storePathRootDir + '\'' +
308351
", namesrvAddr='" + namesrvAddr + '\'' +
309352
", rmqProducerGroup='" + rmqProducerGroup + '\'' +
310353
", maxMessageSize=" + maxMessageSize +
@@ -329,6 +372,9 @@ public String toString() {
329372
", aclEnable=" + aclEnable +
330373
", accessKey='" + accessKey + '\'' +
331374
", secretKey='" + secretKey + '\'' +
375+
", autoCreateGroupEnable=" + autoCreateGroupEnable +
376+
", clusterName='" + clusterName + '\'' +
377+
", adminExtGroup='" + adminExtGroup + '\'' +
332378
'}';
333379
}
334380
}

‎rocketmq-connect/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/Worker.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -426,6 +426,9 @@ public void maintainTaskState() throws Exception {
426426
this.pendingTasks.put(workerSourceTask, System.currentTimeMillis());
427427
} else if (task instanceof SinkTask) {
428428
DefaultMQPullConsumer consumer = ConnectUtil.initDefaultMQPullConsumer(connectConfig);
429+
if (connectConfig.isAutoCreateGroupEnable()) {
430+
ConnectUtil.createSubGroup(connectConfig, consumer.getConsumerGroup());
431+
}
429432

430433
WorkerSinkTask workerSinkTask = new WorkerSinkTask(connectorName,
431434
(SinkTask) task, keyValue, offsetManagementService, recordConverter, consumer, workerState);

‎rocketmq-connect/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/ClusterManagementServiceImpl.java‎

Lines changed: 15 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import org.apache.rocketmq.common.protocol.header.NotifyConsumerIdsChangedRequestHeader;
2929
import org.apache.rocketmq.connect.runtime.common.LoggerName;
3030
import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
31+
import org.apache.rocketmq.connect.runtime.utils.ConnectUtil;
3132
import org.apache.rocketmq.remoting.common.RemotingHelper;
3233
import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
3334
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
@@ -53,8 +54,20 @@ public class ClusterManagementServiceImpl implements ClusterManagementService {
5354
public ClusterManagementServiceImpl(ConnectConfig connectConfig) {
5455
this.connectConfig = connectConfig;
5556
this.workerStatusListeners = new HashSet<>();
56-
this.defaultMQPullConsumer = new DefaultMQPullConsumer(connectConfig.getConnectClusterId());
57-
this.defaultMQPullConsumer.setNamesrvAddr(connectConfig.getNamesrvAddr());
57+
this.defaultMQPullConsumer = ConnectUtil.initDefaultMQPullConsumer(connectConfig);
58+
this.defaultMQPullConsumer.setConsumerGroup(connectConfig.getConnectClusterId());
59+
this.prepare(connectConfig);
60+
}
61+
62+
/**
63+
* Preparation before startup
64+
*
65+
* @param connectConfig
66+
*/
67+
private void prepare(ConnectConfig connectConfig) {
68+
if (connectConfig.isAutoCreateGroupEnable()) {
69+
ConnectUtil.createSubGroup(connectConfig, this.defaultMQPullConsumer.getConsumerGroup());
70+
}
5871
}
5972

6073
@Override

‎rocketmq-connect/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/ConfigManagementServiceImpl.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -77,7 +77,7 @@ public ConfigManagementServiceImpl(ConnectConfig connectConfig, Plugin plugin) {
7777
this.connectorConfigUpdateListener = new HashSet<>();
7878
this.dataSynchronizer = new BrokerBasedLog<>(connectConfig,
7979
connectConfig.getConfigStoreTopic(),
80-
ConnectUtil.createGroupName(configManagePrefix),
80+
ConnectUtil.createGroupName(configManagePrefix, connectConfig.getWorkerId()),
8181
new ConfigChangeCallback(),
8282
new JsonConverter(),
8383
new ConnAndTaskConfigConverter());

‎rocketmq-connect/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/OffsetManagementServiceImpl.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -60,7 +60,7 @@ public OffsetManagementServiceImpl(ConnectConfig connectConfig) {
6060
new ByteBufferConverter());
6161
this.dataSynchronizer = new BrokerBasedLog(connectConfig,
6262
connectConfig.getOffsetStoreTopic(),
63-
ConnectUtil.createGroupName(offsetManagePrefix),
63+
ConnectUtil.createGroupName(offsetManagePrefix, connectConfig.getWorkerId()),
6464
new OffsetChangeCallback(),
6565
new JsonConverter(),
6666
new ByteMapConverter());

‎rocketmq-connect/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/PositionManagementServiceImpl.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -60,7 +60,7 @@ public PositionManagementServiceImpl(ConnectConfig connectConfig) {
6060
new ByteBufferConverter());
6161
this.dataSynchronizer = new BrokerBasedLog(connectConfig,
6262
connectConfig.getPositionStoreTopic(),
63-
ConnectUtil.createGroupName(positionManagePrefix),
63+
ConnectUtil.createGroupName(positionManagePrefix, connectConfig.getWorkerId()),
6464
new PositionChangeCallback(),
6565
new JsonConverter(),
6666
new ByteMapConverter());

‎rocketmq-connect/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/utils/ConnectUtil.java‎

Lines changed: 52 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -20,25 +20,28 @@
2020
import java.util.ArrayList;
2121
import java.util.Collections;
2222
import java.util.List;
23-
import java.util.concurrent.atomic.AtomicLong;
23+
import java.util.Set;
24+
import java.util.UUID;
2425
import org.apache.rocketmq.acl.common.AclClientRPCHook;
2526
import org.apache.rocketmq.acl.common.SessionCredentials;
2627
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
2728
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
29+
import org.apache.rocketmq.client.exception.MQClientException;
2830
import org.apache.rocketmq.client.producer.DefaultMQProducer;
2931
import org.apache.rocketmq.common.UtilAll;
3032
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
33+
import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
3134
import org.apache.rocketmq.connect.runtime.config.ConnectConfig;
3235
import org.apache.rocketmq.connect.runtime.config.RuntimeConfigDefine;
3336
import org.apache.rocketmq.connect.runtime.service.strategy.AllocateConnAndTaskStrategy;
3437
import org.apache.rocketmq.remoting.RPCHook;
3538
import org.apache.rocketmq.remoting.common.RemotingUtil;
3639
import org.apache.rocketmq.remoting.protocol.LanguageCode;
40+
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
41+
import org.apache.rocketmq.tools.command.CommandUtil;
3742

3843
public class ConnectUtil {
3944

40-
private final static AtomicLong GROUP_POSTFIX_ID = new AtomicLong(0);
41-
4245
public static String createGroupName(String prefix) {
4346
StringBuilder sb = new StringBuilder();
4447
sb.append(prefix).append("-");
@@ -48,10 +51,8 @@ public static String createGroupName(String prefix) {
4851
return sb.toString().replace(".", "-");
4952
}
5053

51-
public static String createGroupNameV2(String prefix) {
52-
StringBuilder sb = new StringBuilder();
53-
sb.append(prefix).append("-").append(GROUP_POSTFIX_ID.getAndIncrement());
54-
return sb.toString();
54+
public static String createGroupName(String prefix, String postfix) {
55+
return new StringBuilder().append(prefix).append("-").append(postfix).toString();
5556
}
5657

5758
public static String createInstance(String servers) {
@@ -66,6 +67,10 @@ public static String createInstance(String servers) {
6667
return String.valueOf(serversList.toString().hashCode());
6768
}
6869

70+
public static String createUniqInstance(String prefix) {
71+
return new StringBuffer(prefix).append("-").append(UUID.randomUUID().toString()).toString();
72+
}
73+
6974
public static AllocateConnAndTaskStrategy initAllocateConnAndTaskStrategy(ConnectConfig connectConfig) {
7075
try {
7176
return (AllocateConnAndTaskStrategy) Thread.currentThread().getContextClassLoader().loadClass(connectConfig.getAllocTaskStrategy()).newInstance();
@@ -81,8 +86,8 @@ public static DefaultMQProducer initDefaultMQProducer(ConnectConfig connectConfi
8186
}
8287
DefaultMQProducer producer = new DefaultMQProducer(rpcHook);
8388
producer.setNamesrvAddr(connectConfig.getNamesrvAddr());
84-
producer.setInstanceName(createInstance(connectConfig.getNamesrvAddr()));
85-
producer.setProducerGroup(createGroupNameV2(connectConfig.getRmqProducerGroup()));
89+
producer.setInstanceName(createUniqInstance(connectConfig.getNamesrvAddr()));
90+
producer.setProducerGroup(connectConfig.getRmqProducerGroup());
8691
producer.setSendMsgTimeout(connectConfig.getOperationTimeout());
8792
producer.setMaxMessageSize(RuntimeConfigDefine.MAX_MESSAGE_SIZE);
8893
producer.setLanguage(LanguageCode.JAVA);
@@ -96,8 +101,8 @@ public static DefaultMQPullConsumer initDefaultMQPullConsumer(ConnectConfig conn
96101
}
97102
DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(rpcHook);
98103
consumer.setNamesrvAddr(connectConfig.getNamesrvAddr());
99-
consumer.setInstanceName(createInstance(connectConfig.getNamesrvAddr()));
100-
consumer.setConsumerGroup(createGroupNameV2(connectConfig.getRmqConsumerGroup()));
104+
consumer.setInstanceName(createUniqInstance(connectConfig.getNamesrvAddr()));
105+
consumer.setConsumerGroup(connectConfig.getRmqConsumerGroup());
101106
consumer.setMaxReconsumeTimes(connectConfig.getRmqMaxRedeliveryTimes());
102107
consumer.setBrokerSuspendMaxTimeMillis(connectConfig.getBrokerSuspendMaxTimeMillis());
103108
consumer.setConsumerPullTimeoutMillis((long) connectConfig.getRmqMessageConsumeTimeout());
@@ -112,13 +117,47 @@ public static DefaultMQPushConsumer initDefaultMQPushConsumer(ConnectConfig conn
112117
}
113118
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(rpcHook);
114119
consumer.setNamesrvAddr(connectConfig.getNamesrvAddr());
115-
consumer.setInstanceName(createInstance(connectConfig.getNamesrvAddr()));
116-
consumer.setConsumerGroup(createGroupNameV2(connectConfig.getRmqConsumerGroup()));
120+
consumer.setInstanceName(createUniqInstance(connectConfig.getNamesrvAddr()));
121+
consumer.setConsumerGroup(createGroupName(connectConfig.getRmqConsumerGroup()));
117122
consumer.setMaxReconsumeTimes(connectConfig.getRmqMaxRedeliveryTimes());
118123
consumer.setConsumeTimeout((long) connectConfig.getRmqMessageConsumeTimeout());
119124
consumer.setConsumeThreadMin(connectConfig.getRmqMinConsumeThreadNums());
120125
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
121126
consumer.setLanguage(LanguageCode.JAVA);
122127
return consumer;
123128
}
129+
130+
public static DefaultMQAdminExt startMQAdminTool(ConnectConfig connectConfig) throws MQClientException {
131+
RPCHook rpcHook = null;
132+
if (connectConfig.getAclEnable()) {
133+
rpcHook = new AclClientRPCHook(new SessionCredentials(connectConfig.getAccessKey(), connectConfig.getSecretKey()));
134+
}
135+
DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
136+
defaultMQAdminExt.setNamesrvAddr(connectConfig.getNamesrvAddr());
137+
defaultMQAdminExt.setAdminExtGroup(connectConfig.getAdminExtGroup());
138+
defaultMQAdminExt.setInstanceName(ConnectUtil.createUniqInstance(connectConfig.getNamesrvAddr()));
139+
defaultMQAdminExt.start();
140+
return defaultMQAdminExt;
141+
}
142+
143+
public static String createSubGroup(ConnectConfig connectConfig, String subGroup) {
144+
DefaultMQAdminExt defaultMQAdminExt = null;
145+
try {
146+
defaultMQAdminExt = startMQAdminTool(connectConfig);
147+
SubscriptionGroupConfig initConfig = new SubscriptionGroupConfig();
148+
initConfig.setGroupName(subGroup);
149+
150+
Set<String> masterSet = CommandUtil.fetchMasterAddrByClusterName(defaultMQAdminExt, connectConfig.getClusterName());
151+
for (String addr : masterSet) {
152+
defaultMQAdminExt.createAndUpdateSubscriptionGroupConfig(addr, initConfig);
153+
}
154+
} catch (Exception e) {
155+
throw new IllegalArgumentException("create subGroup: " + subGroup + " failed", e);
156+
} finally {
157+
if (defaultMQAdminExt != null) {
158+
defaultMQAdminExt.shutdown();
159+
}
160+
}
161+
return subGroup;
162+
}
124163
}

‎rocketmq-connect/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/utils/datasync/BrokerBasedLog.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,18 @@ public BrokerBasedLog(ConnectConfig connectConfig,
9595
this.consumer.setConsumerGroup(workId);
9696
this.keyConverter = keyConverter;
9797
this.valueConverter = valueConverter;
98+
this.prepare(connectConfig);
99+
}
100+
101+
/**
102+
* Preparation before startup
103+
*
104+
* @param connectConfig
105+
*/
106+
private void prepare(ConnectConfig connectConfig) {
107+
if (connectConfig.isAutoCreateGroupEnable()) {
108+
ConnectUtil.createSubGroup(connectConfig, consumer.getConsumerGroup());
109+
}
98110
}
99111

100112
@Override

‎rocketmq-connect/rocketmq-connect-runtime/src/main/resources/connect.conf‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,9 @@ aclEnable=false
2727
accessKey=rocketmq
2828
secretKey=12345678
2929

30+
autoCreateGroupEnable=false
31+
clusterName="DefaultCluster"
32+
3033
# Source or sink connector jar file dir,The default value is rocketmq-connect-sample
3134
pluginPaths=/xxx/connector-plugins
3235

0 commit comments

Comments
 (0)