opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-01 15:47:01 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #5862] remove offline producer group (#5865)

Browse Source
This commit is contained in:
lk
2023-01-12 10:28:45 +08:00
committed by GitHub
parent 7694a69d59
commit 07ce0be339
3 changed files with 9 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/src/main/java/org/apache/rocketmq/proxy/service/ClusterServiceManager.java
+1 -1
View File
@@ -99,7 +99,7 @@ public class ClusterServiceManager extends AbstractStartAndShutdown implements S
new ProxyClientRemotingProcessor(producerManager),
rpcHook,
scheduledExecutorService);
this.clusterTransactionService = new ClusterTransactionService(this.topicRouteService, this.producerManager, rpcHook,
this.clusterTransactionService = new ClusterTransactionService(this.topicRouteService, this.producerManager,
this.transactionClientAPIFactory);
this.proxyRelayService = new ClusterProxyRelayService(this.clusterTransactionService);
proxy/src/main/java/org/apache/rocketmq/proxy/service/transaction/ClusterTransactionService.java
+5 -2
View File
@@ -41,7 +41,6 @@ import org.apache.rocketmq.proxy.config.ProxyConfig;
import org.apache.rocketmq.proxy.service.mqclient.MQClientAPIFactory;
import org.apache.rocketmq.proxy.service.route.MessageQueueView;
import org.apache.rocketmq.proxy.service.route.TopicRouteService;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.protocol.heartbeat.HeartbeatData;
import org.apache.rocketmq.remoting.protocol.heartbeat.ProducerData;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
@@ -53,6 +52,7 @@ public class ClusterTransactionService extends AbstractTransactionService {
private final MQClientAPIFactory mqClientAPIFactory;
private final TopicRouteService topicRouteService;
private final ProducerManager producerManager;
private ThreadPoolExecutor heartbeatExecutors;
private final Map<String /* group */, Set<ClusterData>/* cluster list */> groupClusterData = new ConcurrentHashMap<>();
@@ -60,9 +60,9 @@ public class ClusterTransactionService extends AbstractTransactionService {
private TxHeartbeatServiceThread txHeartbeatServiceThread;
public ClusterTransactionService(TopicRouteService topicRouteService, ProducerManager producerManager,
RPCHook rpcHook,
MQClientAPIFactory mqClientAPIFactory) {
this.topicRouteService = topicRouteService;
this.producerManager = producerManager;
this.mqClientAPIFactory = mqClientAPIFactory;
}
@@ -130,6 +130,9 @@ public class ClusterTransactionService extends AbstractTransactionService {
if (clusterDataSet.isEmpty()) {
return null;
}
if (!this.producerManager.groupOnline(groupName)) {
return null;
}
ProducerData producerData = new ProducerData();
producerData.setGroupName(groupName);
proxy/src/test/java/org/apache/rocketmq/proxy/service/transaction/ClusterTransactionServiceTest.java
+3 -1
View File
@@ -59,7 +59,7 @@ public class ClusterTransactionServiceTest extends BaseServiceTest {
@Before
public void before() throws Throwable {
super.before();
this.clusterTransactionService = new ClusterTransactionService(this.topicRouteService, this.producerManager, null,
this.clusterTransactionService = new ClusterTransactionService(this.topicRouteService, this.producerManager,
this.mqClientAPIFactory);
MessageQueueView messageQueueView = new MessageQueueView(TOPIC, topicRouteData);
@@ -108,6 +108,8 @@ public class ClusterTransactionServiceTest extends BaseServiceTest {
@Test
public void testScanProducerHeartBeat() throws Exception {
when(this.producerManager.groupOnline(anyString())).thenReturn(true);
Mockito.reset(this.topicRouteService);
String brokerName2 = "broker-2-01";
String clusterName2 = "broker-2";