This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new 97a797543b [ISSUE #10906] Prevent duplicate MQClientInstance creation
(#10907)
97a797543b is described below
commit 97a797543bad7697a79ea227671f3047266b6620
Author: qianye <[email protected]>
AuthorDate: Wed Aug 12 14:02:31 2026 +0800
[ISSUE #10906] Prevent duplicate MQClientInstance creation (#10907)
---
.../rocketmq/client/impl/MQClientManager.java | 27 +-
.../client/impl/factory/MQClientInstance.java | 204 ++++++----
.../rocketmq/client/impl/MQClientManagerTest.java | 440 +++++++++++++++++++++
.../apache/rocketmq/common/ServiceThreadTest.java | 58 ++-
.../test/client/consumer/pop/BatchAckIT.java | 24 +-
.../client/producer/querymsg/QueryMsgByKeyIT.java | 13 +-
6 files changed, 638 insertions(+), 128 deletions(-)
diff --git
a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientManager.java
b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientManager.java
index ca6f461745..e6f17727ff 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientManager.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientManager.java
@@ -51,16 +51,15 @@ public class MQClientManager {
String clientId = clientConfig.buildMQClientId();
MQClientInstance instance = this.factoryTable.get(clientId);
if (null == instance) {
- instance =
- new MQClientInstance(clientConfig.cloneClientConfig(),
- this.factoryIndexGenerator.getAndIncrement(), clientId,
rpcHook);
- MQClientInstance prev = this.factoryTable.putIfAbsent(clientId,
instance);
- if (prev != null) {
- instance = prev;
- log.warn("Returned Previous MQClientInstance for
clientId:[{}]", clientId);
- } else {
- log.info("Created new MQClientInstance for clientId:[{}]",
clientId);
- }
+ ClientConfig clonedClientConfig = clientConfig.cloneClientConfig();
+ // MQClientInstance construction must not call back into
factoryTable. ConcurrentHashMap rejects
+ // recursive updates from a mapping function with
IllegalStateException.
+ instance = this.factoryTable.computeIfAbsent(clientId, key -> {
+ MQClientInstance newInstance = new
MQClientInstance(clonedClientConfig,
+ this.factoryIndexGenerator.getAndIncrement(), key,
rpcHook);
+ log.info("Created new MQClientInstance for clientId:[{}]",
key);
+ return newInstance;
+ });
}
return instance;
@@ -82,10 +81,18 @@ public class MQClientManager {
return accumulator;
}
+ /**
+ * Removes the mapped factory without checking its identity. Lifecycle
cleanup should prefer
+ * {@link #removeClientFactory(String, MQClientInstance)} to avoid
removing a replacement instance.
+ */
public void removeClientFactory(final String clientId) {
this.factoryTable.remove(clientId);
}
+ public void removeClientFactory(final String clientId, final
MQClientInstance instance) {
+ this.factoryTable.remove(clientId, instance);
+ }
+
public ConcurrentMap<String, MQClientInstance> getFactoryTable() {
return factoryTable;
}
diff --git
a/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java
b/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java
index cd45fed2a3..5208e49fe7 100644
---
a/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java
+++
b/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java
@@ -148,86 +148,114 @@ public class MQClientInstance {
}
public MQClientInstance(ClientConfig clientConfig, int instanceIndex,
String clientId, RPCHook rpcHook) {
- this.clientConfig = clientConfig;
- this.nettyClientConfig = new NettyClientConfig();
-
this.nettyClientConfig.setClientCallbackExecutorThreads(clientConfig.getClientCallbackExecutorThreads());
- this.nettyClientConfig.setUseTLS(clientConfig.isUseTLS());
-
this.nettyClientConfig.setSocksProxyConfig(clientConfig.getSocksProxyConfig());
- this.nettyClientConfig.setScanAvailableNameSrv(false);
- ClientRemotingProcessor clientRemotingProcessor = new
ClientRemotingProcessor(this);
- ChannelEventListener channelEventListener;
- if (clientConfig.isEnableHeartbeatChannelEventListener()) {
- channelEventListener = new ChannelEventListener() {
-
- private final ConcurrentMap<String, HashMap<Long, String>>
brokerAddrTable = MQClientInstance.this.brokerAddrTable;
-
- @Override
- public void onChannelConnect(String remoteAddr, Channel
channel) {
- }
+ MQClientAPIImpl clientAPI = null;
+ try {
+ this.clientConfig = clientConfig;
+ this.nettyClientConfig = new NettyClientConfig();
+
this.nettyClientConfig.setClientCallbackExecutorThreads(clientConfig.getClientCallbackExecutorThreads());
+ this.nettyClientConfig.setUseTLS(clientConfig.isUseTLS());
+
this.nettyClientConfig.setSocksProxyConfig(clientConfig.getSocksProxyConfig());
+ this.nettyClientConfig.setScanAvailableNameSrv(false);
+ ClientRemotingProcessor clientRemotingProcessor = new
ClientRemotingProcessor(this);
+ ChannelEventListener channelEventListener;
+ if (clientConfig.isEnableHeartbeatChannelEventListener()) {
+ channelEventListener = new ChannelEventListener() {
+
+ private final ConcurrentMap<String, HashMap<Long, String>>
brokerAddrTable = MQClientInstance.this.brokerAddrTable;
+
+ @Override
+ public void onChannelConnect(String remoteAddr, Channel
channel) {
+ }
- @Override
- public void onChannelClose(String remoteAddr, Channel channel)
{
- }
+ @Override
+ public void onChannelClose(String remoteAddr, Channel
channel) {
+ }
- @Override
- public void onChannelException(String remoteAddr, Channel
channel) {
- }
+ @Override
+ public void onChannelException(String remoteAddr, Channel
channel) {
+ }
- @Override
- public void onChannelIdle(String remoteAddr, Channel channel) {
- }
+ @Override
+ public void onChannelIdle(String remoteAddr, Channel
channel) {
+ }
- @Override
- public void onChannelActive(String remoteAddr, Channel
channel) {
- for (Map.Entry<String, HashMap<Long, String>> addressEntry
: brokerAddrTable.entrySet()) {
- for (Map.Entry<Long, String> entry :
addressEntry.getValue().entrySet()) {
- String addr = entry.getValue();
- if (addr.equals(remoteAddr)) {
- long id = entry.getKey();
- String brokerName = addressEntry.getKey();
- if (sendHeartbeatToBroker(id, brokerName,
addr, false)) {
- rebalanceImmediately();
+ @Override
+ public void onChannelActive(String remoteAddr, Channel
channel) {
+ for (Map.Entry<String, HashMap<Long, String>>
addressEntry : brokerAddrTable.entrySet()) {
+ for (Map.Entry<Long, String> entry :
addressEntry.getValue().entrySet()) {
+ String addr = entry.getValue();
+ if (addr.equals(remoteAddr)) {
+ long id = entry.getKey();
+ String brokerName = addressEntry.getKey();
+ if (sendHeartbeatToBroker(id, brokerName,
addr, false)) {
+ rebalanceImmediately();
+ }
+ break;
}
- break;
}
}
}
- }
- };
- } else {
- channelEventListener = null;
- }
- this.mQClientAPIImpl = new MQClientAPIImpl(this.nettyClientConfig,
clientRemotingProcessor, rpcHook, clientConfig, channelEventListener);
+ };
+ } else {
+ channelEventListener = null;
+ }
+ this.mQClientAPIImpl = new MQClientAPIImpl(this.nettyClientConfig,
clientRemotingProcessor, rpcHook, clientConfig, channelEventListener);
+ clientAPI = this.mQClientAPIImpl;
- if (this.clientConfig.getNamesrvAddr() != null) {
-
this.mQClientAPIImpl.updateNameServerAddressList(this.clientConfig.getNamesrvAddr());
- log.info("user specified name server address: {}",
this.clientConfig.getNamesrvAddr());
- }
+ if (this.clientConfig.getNamesrvAddr() != null) {
+
this.mQClientAPIImpl.updateNameServerAddressList(this.clientConfig.getNamesrvAddr());
+ log.info("user specified name server address: {}",
this.clientConfig.getNamesrvAddr());
+ }
+
+ this.clientId = clientId;
- this.clientId = clientId;
+ this.mQAdminImpl = new MQAdminImpl(this);
- this.mQAdminImpl = new MQAdminImpl(this);
+ this.pullMessageService = new PullMessageService(this);
- this.pullMessageService = new PullMessageService(this);
+ this.rebalanceService = new RebalanceService(this);
- this.rebalanceService = new RebalanceService(this);
+ this.defaultMQProducer = new
DefaultMQProducer(MixAll.CLIENT_INNER_PRODUCER_GROUP);
+ this.defaultMQProducer.resetClientConfig(clientConfig);
- this.defaultMQProducer = new
DefaultMQProducer(MixAll.CLIENT_INNER_PRODUCER_GROUP);
- this.defaultMQProducer.resetClientConfig(clientConfig);
+ this.consumerStatsManager = new
ConsumerStatsManager(this.scheduledExecutorService);
+
+ if (this.clientConfig.isEnableConcurrentHeartbeat()) {
+ this.concurrentHeartbeatExecutor =
Executors.newFixedThreadPool(
+ clientConfig.getConcurrentHeartbeatThreadPoolSize(),
+ new
ThreadFactoryImpl("MQClientConcurrentHeartbeatThread_", true));
+ }
- this.consumerStatsManager = new
ConsumerStatsManager(this.scheduledExecutorService);
+ log.info("Created a new client Instance, InstanceIndex:{},
ClientID:{}, ClientConfig:{}, ClientVersion:{}, SerializerType:{}",
+ instanceIndex,
+ this.clientId,
+ this.clientConfig,
+ MQVersion.getVersionDesc(MQVersion.CURRENT_VERSION),
RemotingCommand.getSerializeTypeConfigInThisServer());
+ } catch (RuntimeException | Error e) {
+ cleanupAfterConstructionFailure(clientAPI, e);
+ throw e;
+ }
+ }
- if (this.clientConfig.isEnableConcurrentHeartbeat()) {
- this.concurrentHeartbeatExecutor = Executors.newFixedThreadPool(
- clientConfig.getConcurrentHeartbeatThreadPoolSize(),
- new ThreadFactoryImpl("MQClientConcurrentHeartbeatThread_",
true));
+ private void cleanupAfterConstructionFailure(MQClientAPIImpl clientAPI,
Throwable cause) {
+ runCleanup(this.scheduledExecutorService::shutdownNow, cause);
+ if (this.concurrentHeartbeatExecutor != null) {
+ runCleanup(this.concurrentHeartbeatExecutor::shutdownNow, cause);
}
+ if (clientAPI != null) {
+ runCleanup(clientAPI::shutdown, cause);
+ }
+ }
- log.info("Created a new client Instance, InstanceIndex:{},
ClientID:{}, ClientConfig:{}, ClientVersion:{}, SerializerType:{}",
- instanceIndex,
- this.clientId,
- this.clientConfig,
- MQVersion.getVersionDesc(MQVersion.CURRENT_VERSION),
RemotingCommand.getSerializeTypeConfigInThisServer());
+ private static void runCleanup(Runnable cleanup, Throwable cause) {
+ try {
+ cleanup.run();
+ } catch (Throwable t) {
+ // Cleanup on Error paths is best effort; always preserve the
original failure.
+ if (t != cause) {
+ cause.addSuppressed(t);
+ }
+ }
}
public static TopicPublishInfo topicRouteData2TopicPublishInfo(final
String topic, final TopicRouteData route) {
@@ -313,22 +341,31 @@ public class MQClientInstance {
switch (this.serviceState) {
case CREATE_JUST:
this.serviceState = ServiceState.START_FAILED;
- // If not specified,looking address from name server
- if (null == this.clientConfig.getNamesrvAddr()) {
- this.mQClientAPIImpl.fetchNameServerAddr();
+ try {
+ // If not specified,looking address from name server
+ if (null == this.clientConfig.getNamesrvAddr()) {
+ this.mQClientAPIImpl.fetchNameServerAddr();
+ }
+ // Start request-response channel
+ this.mQClientAPIImpl.start();
+ // Start various schedule tasks
+ this.startScheduledTask();
+ // Start pull service
+ this.pullMessageService.start();
+ // Start rebalance service
+ this.rebalanceService.start();
+ // Start push service
+
this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
+ log.info("the client factory [{}] start OK",
this.clientId);
+ this.serviceState = ServiceState.RUNNING;
+ } catch (MQClientException | RuntimeException | Error e) {
+ // Do not apply the normal shutdown registration
guards here: a factory that never reached
+ // RUNNING cannot serve any registered client, and its
partially started resources must stop.
+ // Existing holders still observe START_FAILED; a
later manager lookup may create a replacement.
+ cleanupAfterStartFailure(e);
+
MQClientManager.getInstance().removeClientFactory(this.clientId, this);
+ throw e;
}
- // Start request-response channel
- this.mQClientAPIImpl.start();
- // Start various schedule tasks
- this.startScheduledTask();
- // Start pull service
- this.pullMessageService.start();
- // Start rebalance service
- this.rebalanceService.start();
- // Start push service
-
this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
- log.info("the client factory [{}] start OK",
this.clientId);
- this.serviceState = ServiceState.RUNNING;
break;
case START_FAILED:
throw new MQClientException("The Factory object[" +
this.getClientId() + "] has been created before, and failed.", null);
@@ -338,6 +375,17 @@ public class MQClientInstance {
}
}
+ private void cleanupAfterStartFailure(Throwable cause) {
+ runCleanup(this.scheduledExecutorService::shutdownNow, cause);
+ if (this.concurrentHeartbeatExecutor != null) {
+ runCleanup(this.concurrentHeartbeatExecutor::shutdownNow, cause);
+ }
+ runCleanup(() ->
this.defaultMQProducer.getDefaultMQProducerImpl().shutdown(false), cause);
+ runCleanup(() -> this.pullMessageService.shutdown(true), cause);
+ runCleanup(this.rebalanceService::shutdown, cause);
+ runCleanup(this.mQClientAPIImpl::shutdown, cause);
+ }
+
private void startScheduledTask() {
if (null == this.clientConfig.getNamesrvAddr()) {
this.scheduledExecutorService.scheduleAtFixedRate(() -> {
@@ -1077,7 +1125,7 @@ public class MQClientInstance {
this.concurrentHeartbeatExecutor.shutdown();
}
-
MQClientManager.getInstance().removeClientFactory(this.clientId);
+
MQClientManager.getInstance().removeClientFactory(this.clientId, this);
log.info("the client factory [{}] shutdown OK",
this.clientId);
break;
case CREATE_JUST:
diff --git
a/client/src/test/java/org/apache/rocketmq/client/impl/MQClientManagerTest.java
b/client/src/test/java/org/apache/rocketmq/client/impl/MQClientManagerTest.java
new file mode 100644
index 0000000000..82aea4c421
--- /dev/null
+++
b/client/src/test/java/org/apache/rocketmq/client/impl/MQClientManagerTest.java
@@ -0,0 +1,440 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.client.impl;
+
+import java.lang.reflect.Constructor;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.rocketmq.client.ClientConfig;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.impl.consumer.PullMessageService;
+import org.apache.rocketmq.client.impl.factory.MQClientInstance;
+import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
+import org.apache.rocketmq.common.ServiceState;
+import org.junit.After;
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+public class MQClientManagerTest {
+ private static final String SCHEDULER_THREAD_NAME =
"MQClientFactoryScheduledThread";
+ private static final long TIMEOUT_SECONDS = 30;
+
+ private final Set<MQClientInstance> instancesToDispose =
+ Collections.newSetFromMap(new IdentityHashMap<MQClientInstance,
Boolean>());
+
+ @After
+ public void tearDown() throws Exception {
+ for (MQClientInstance instance : instancesToDispose) {
+ dispose(instance);
+ }
+ instancesToDispose.clear();
+ }
+
+ @Test
+ public void concurrentSameClientIdCreatesOneInstance() throws Exception {
+ int callers = 16;
+ MQClientManager manager = newManager();
+ FieldUtils.writeDeclaredField(manager, "factoryTable", new
BarrierGetMap<>(callers), true);
+ ThreadGroup threadGroup = new ThreadGroup("same-client-id-" +
System.nanoTime());
+ ExecutorService executor = newExecutor(threadGroup, callers);
+ ClientConfig config = newConfig("same-client-id");
+
+ try {
+ List<Future<MQClientInstance>> futures = new ArrayList<>();
+ for (int i = 0; i < callers; i++) {
+ futures.add(executor.submit(() ->
manager.getOrCreateMQClientInstance(config)));
+ }
+
+ Set<MQClientInstance> returned = Collections.newSetFromMap(
+ new IdentityHashMap<MQClientInstance, Boolean>());
+ for (Future<MQClientInstance> future : futures) {
+ returned.add(future.get(TIMEOUT_SECONDS, TimeUnit.SECONDS));
+ }
+
+ assertThat(returned).hasSize(1);
+ MQClientInstance instance = returned.iterator().next();
+ track(instance);
+ assertSame(instance,
manager.getFactoryTable().get(config.buildMQClientId()));
+ assertThat(factoryIndex(manager).get()).isEqualTo(1);
+ awaitSchedulerThreadCount(threadGroup, 1);
+ } finally {
+ shutdown(executor);
+ }
+ }
+
+ @Test
+ public void differentClientIdsAreConstructedConcurrently() throws
Exception {
+ MQClientManager manager = newManager();
+ CyclicBarrier constructorBarrier = new CyclicBarrier(2);
+ ThreadGroup threadGroup = new ThreadGroup("different-client-id-" +
System.nanoTime());
+ ExecutorService executor = newExecutor(threadGroup, 2);
+ ClientConfig firstConfig = concurrentConstructorConfig(
+ "different-a-" + System.nanoTime(), constructorBarrier);
+ ClientConfig secondConfig = concurrentConstructorConfig(
+ differentHashBinInstanceName(firstConfig.buildMQClientId(),
"different-b"), constructorBarrier);
+
+ try {
+ Future<MQClientInstance> firstFuture = executor.submit(
+ () -> manager.getOrCreateMQClientInstance(firstConfig));
+ Future<MQClientInstance> secondFuture = executor.submit(
+ () -> manager.getOrCreateMQClientInstance(secondConfig));
+
+ MQClientInstance first = firstFuture.get(TIMEOUT_SECONDS,
TimeUnit.SECONDS);
+ MQClientInstance second = secondFuture.get(TIMEOUT_SECONDS,
TimeUnit.SECONDS);
+ track(first);
+ track(second);
+ assertThat(first).isNotSameAs(second);
+ assertThat(manager.getFactoryTable()).hasSize(2);
+ assertThat(factoryIndex(manager).get()).isEqualTo(2);
+ awaitSchedulerThreadCount(threadGroup, 2);
+ } finally {
+ shutdown(executor);
+ }
+ }
+
+ @Test
+ public void constructorFailureRollsBackResourcesAndAllowsRetry() throws
Exception {
+ MQClientManager manager = newManager();
+ ThreadGroup threadGroup = new ThreadGroup("constructor-failure-" +
System.nanoTime());
+ ExecutorService executor = newExecutor(threadGroup, 1);
+ ClientConfig config = newConfig("constructor-failure");
+ // The heartbeat executor is currently created after
ConsumerStatsManager registers its tasks. A zero-sized
+ // pool therefore injects a late constructor failure and verifies
rollback of already-started resources.
+ config.setEnableConcurrentHeartbeat(true);
+ config.setConcurrentHeartbeatThreadPoolSize(0);
+
+ try {
+ Future<MQClientInstance> failed = executor.submit(
+ () -> manager.getOrCreateMQClientInstance(config));
+ try {
+ failed.get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ fail("Expected constructor failure");
+ } catch (ExecutionException e) {
+
assertThat(e.getCause()).isInstanceOf(IllegalArgumentException.class);
+ }
+
+
assertThat(manager.getFactoryTable()).doesNotContainKey(config.buildMQClientId());
+ awaitSchedulerThreadCount(threadGroup, 0);
+
+ config.setConcurrentHeartbeatThreadPoolSize(1);
+ MQClientInstance retried = executor.submit(
+ () ->
manager.getOrCreateMQClientInstance(config)).get(TIMEOUT_SECONDS,
TimeUnit.SECONDS);
+ track(retried);
+ assertSame(retried,
manager.getFactoryTable().get(config.buildMQClientId()));
+ awaitSchedulerThreadCount(threadGroup, 1);
+ } finally {
+ shutdown(executor);
+ }
+ }
+
+ @Test
+ public void removeAllowsInstanceToBeRecreated() throws Exception {
+ MQClientManager manager = newManager();
+ ClientConfig config = newConfig("remove-recreate");
+ MQClientInstance first = manager.getOrCreateMQClientInstance(config);
+ track(first);
+
+ manager.removeClientFactory(config.buildMQClientId());
+ MQClientInstance second = manager.getOrCreateMQClientInstance(config);
+ track(second);
+
+ assertThat(second).isNotSameAs(first);
+ assertSame(second,
manager.getFactoryTable().get(config.buildMQClientId()));
+ manager.removeClientFactory(config.buildMQClientId(), first);
+ assertSame(second,
manager.getFactoryTable().get(config.buildMQClientId()));
+ }
+
+ @Test
+ public void startAndShutdownRemainIdempotentAndAllowRecreation() throws
Exception {
+ MQClientManager manager = MQClientManager.getInstance();
+ ClientConfig config = newConfig("start-shutdown");
+ ThreadGroup threadGroup = new ThreadGroup("start-shutdown-" +
System.nanoTime());
+ ExecutorService executor = newExecutor(threadGroup, 1);
+
+ try {
+ MQClientInstance first = executor.submit(
+ () ->
manager.getOrCreateMQClientInstance(config)).get(TIMEOUT_SECONDS,
TimeUnit.SECONDS);
+ track(first);
+ first.start();
+ first.start();
+ assertThat(serviceState(first)).isEqualTo(ServiceState.RUNNING);
+
+ first.shutdown();
+ first.shutdown();
+
assertThat(serviceState(first)).isEqualTo(ServiceState.SHUTDOWN_ALREADY);
+
assertThat(manager.getFactoryTable()).doesNotContainKey(config.buildMQClientId());
+ assertExecutorTerminated(scheduler(first));
+
+ MQClientInstance second = executor.submit(
+ () ->
manager.getOrCreateMQClientInstance(config)).get(TIMEOUT_SECONDS,
TimeUnit.SECONDS);
+ track(second);
+ assertThat(second).isNotSameAs(first);
+ assertSame(second,
manager.getFactoryTable().get(config.buildMQClientId()));
+ } finally {
+ shutdown(executor);
+ }
+ }
+
+ @Test
+ public void startFailureRollsBackResourcesAndAllowsReplacement() throws
Exception {
+ MQClientManager manager = MQClientManager.getInstance();
+ ClientConfig config = newConfig("start-failure");
+ ThreadGroup threadGroup = new ThreadGroup("start-failure-" +
System.nanoTime());
+ ExecutorService executor = newExecutor(threadGroup, 1);
+
+ try {
+ MQClientInstance failedInstance = executor.submit(
+ () ->
manager.getOrCreateMQClientInstance(config)).get(TIMEOUT_SECONDS,
TimeUnit.SECONDS);
+ track(failedInstance);
+ MQClientAPIImpl originalClientAPI = clientAPI(failedInstance);
+ originalClientAPI.shutdown();
+ PullMessageService originalPullMessageService =
(PullMessageService) FieldUtils.readDeclaredField(
+ failedInstance, "pullMessageService", true);
+ originalPullMessageService.shutdown(true);
+ MQClientAPIImpl clientAPI = mock(MQClientAPIImpl.class);
+ PullMessageService failingPullMessageService =
mock(PullMessageService.class);
+ doThrow(new IllegalStateException("injected start
failure")).when(failingPullMessageService).start();
+ // Test-only fault injection deliberately replaces final
collaborators by field name. This avoids adding
+ // production injection hooks, but these assignments must be
updated if the fields are renamed.
+ FieldUtils.writeDeclaredField(failedInstance, "mQClientAPIImpl",
clientAPI, true);
+ FieldUtils.writeDeclaredField(failedInstance,
"pullMessageService", failingPullMessageService, true);
+
+ assertThrows(IllegalStateException.class, failedInstance::start);
+ verify(clientAPI).start();
+ verify(clientAPI).shutdown();
+ verify(failingPullMessageService).start();
+ verify(failingPullMessageService).shutdown(true);
+
assertThat(serviceState(failedInstance)).isEqualTo(ServiceState.START_FAILED);
+
assertThat(manager.getFactoryTable()).doesNotContainKey(config.buildMQClientId());
+ assertExecutorTerminated(scheduler(failedInstance));
+ awaitSchedulerThreadCount(threadGroup, 0);
+ assertThrows(MQClientException.class, failedInstance::start);
+
+ MQClientInstance replacement = executor.submit(
+ () ->
manager.getOrCreateMQClientInstance(config)).get(TIMEOUT_SECONDS,
TimeUnit.SECONDS);
+ track(replacement);
+ assertThat(replacement).isNotSameAs(failedInstance);
+ assertSame(replacement,
manager.getFactoryTable().get(config.buildMQClientId()));
+ } finally {
+ shutdown(executor);
+ }
+ }
+
+ @Test
+ public void staleShutdownDoesNotRemoveReplacement() throws Exception {
+ MQClientManager manager = MQClientManager.getInstance();
+ ClientConfig config = newConfig("stale-shutdown");
+ MQClientInstance first = manager.getOrCreateMQClientInstance(config);
+ track(first);
+ first.start();
+
+ manager.removeClientFactory(config.buildMQClientId());
+ MQClientInstance replacement =
manager.getOrCreateMQClientInstance(config);
+ track(replacement);
+ first.shutdown();
+
+ assertSame(replacement,
manager.getFactoryTable().get(config.buildMQClientId()));
+ }
+
+ private MQClientInstance track(MQClientInstance instance) {
+ instancesToDispose.add(instance);
+ return instance;
+ }
+
+ private static MQClientManager newManager() throws Exception {
+ Constructor<MQClientManager> constructor =
MQClientManager.class.getDeclaredConstructor();
+ constructor.setAccessible(true);
+ return constructor.newInstance();
+ }
+
+ private static ClientConfig newConfig(String suffix) {
+ ClientConfig config = new ClientConfig();
+ config.setInstanceName(suffix + "-" + System.nanoTime());
+ config.setNamesrvAddr("127.0.0.1:9876");
+ return config;
+ }
+
+ private static ClientConfig concurrentConstructorConfig(String
instanceName, CyclicBarrier barrier) {
+ ClientConfig config = new ClientConfig() {
+ @Override
+ public ClientConfig cloneClientConfig() {
+ ClientConfig cloned = new ClientConfig() {
+ @Override
+ public int getClientCallbackExecutorThreads() {
+ await(barrier);
+ return super.getClientCallbackExecutorThreads();
+ }
+ };
+ cloned.resetClientConfig(this);
+ return cloned;
+ }
+ };
+ config.setInstanceName(instanceName);
+ config.setNamesrvAddr("127.0.0.1:9876");
+ return config;
+ }
+
+ private static String differentHashBinInstanceName(String firstClientId,
String prefix) {
+ // ConcurrentHashMap currently spreads h as h ^ (h >>> 16) and starts
with 16 bins. Selecting a different
+ // initial bin makes the constructor barrier prove that creation is
not globally serialized. Update this
+ // helper if the JDK's ConcurrentHashMap hashing or initial table size
changes.
+ int firstBin = spread(firstClientId.hashCode()) & 15;
+ for (int i = 0; ; i++) {
+ String candidate = prefix + "-" + i;
+ ClientConfig config = new ClientConfig();
+ config.setInstanceName(candidate);
+ if ((spread(config.buildMQClientId().hashCode()) & 15) !=
firstBin) {
+ return candidate;
+ }
+ }
+ }
+
+ private static int spread(int hashCode) {
+ return hashCode ^ (hashCode >>> 16);
+ }
+
+ private static AtomicInteger factoryIndex(MQClientManager manager) throws
IllegalAccessException {
+ return (AtomicInteger) FieldUtils.readDeclaredField(manager,
"factoryIndexGenerator", true);
+ }
+
+ private static ServiceState serviceState(MQClientInstance instance) throws
IllegalAccessException {
+ return (ServiceState) FieldUtils.readDeclaredField(instance,
"serviceState", true);
+ }
+
+ private static ScheduledExecutorService scheduler(MQClientInstance
instance) throws IllegalAccessException {
+ return (ScheduledExecutorService) FieldUtils.readDeclaredField(
+ instance, "scheduledExecutorService", true);
+ }
+
+ private static MQClientAPIImpl clientAPI(MQClientInstance instance) throws
IllegalAccessException {
+ return (MQClientAPIImpl) FieldUtils.readDeclaredField(instance,
"mQClientAPIImpl", true);
+ }
+
+ private static ExecutorService newExecutor(ThreadGroup threadGroup, int
threads) {
+ AtomicInteger index = new AtomicInteger();
+ return Executors.newFixedThreadPool(threads,
+ task -> new Thread(threadGroup, task, "MQClientManagerTestCaller-"
+ index.getAndIncrement()));
+ }
+
+ private static void shutdown(ExecutorService executor) throws
InterruptedException {
+ executor.shutdownNow();
+ assertThat(executor.awaitTermination(TIMEOUT_SECONDS,
TimeUnit.SECONDS)).isTrue();
+ }
+
+ private static void awaitSchedulerThreadCount(ThreadGroup threadGroup, int
expected) {
+ org.awaitility.Awaitility.await().atMost(TIMEOUT_SECONDS,
TimeUnit.SECONDS).untilAsserted(() ->
+ assertThat(countThreads(threadGroup,
SCHEDULER_THREAD_NAME)).isEqualTo(expected));
+ }
+
+ private static int countThreads(ThreadGroup threadGroup, String
threadName) {
+ int capacity = Math.max(16, threadGroup.activeCount() * 2);
+ while (true) {
+ Thread[] threads = new Thread[capacity];
+ int count = threadGroup.enumerate(threads, true);
+ if (count < capacity) {
+ int matches = 0;
+ for (int i = 0; i < count; i++) {
+ if (threadName.equals(threads[i].getName()) &&
threads[i].isAlive()) {
+ matches++;
+ }
+ }
+ return matches;
+ }
+ capacity *= 2;
+ }
+ }
+
+ private static void assertExecutorTerminated(ScheduledExecutorService
executor) throws InterruptedException {
+ assertThat(executor.isShutdown()).isTrue();
+ assertThat(executor.awaitTermination(TIMEOUT_SECONDS,
TimeUnit.SECONDS)).isTrue();
+ }
+
+ private static void dispose(MQClientInstance instance) throws Exception {
+ try {
+ instance.shutdown();
+ } finally {
+ ScheduledExecutorService scheduler = scheduler(instance);
+ scheduler.shutdownNow();
+ scheduler.awaitTermination(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+
+ ExecutorService heartbeatExecutor = (ExecutorService)
FieldUtils.readDeclaredField(
+ instance, "concurrentHeartbeatExecutor", true);
+ if (heartbeatExecutor != null) {
+ heartbeatExecutor.shutdownNow();
+ heartbeatExecutor.awaitTermination(TIMEOUT_SECONDS,
TimeUnit.SECONDS);
+ }
+
+ DefaultMQProducerImpl producer = (DefaultMQProducerImpl)
FieldUtils.readDeclaredField(
+ FieldUtils.readDeclaredField(instance, "defaultMQProducer",
true),
+ "defaultMQProducerImpl", true);
+ producer.shutdown(false);
+ ((PullMessageService) FieldUtils.readDeclaredField(instance,
"pullMessageService", true)).shutdown(true);
+ clientAPI(instance).shutdown();
+
MQClientManager.getInstance().removeClientFactory(instance.getClientId(),
instance);
+ }
+ }
+
+ private static void await(CyclicBarrier barrier) {
+ try {
+ barrier.await(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError(e);
+ } catch (BrokenBarrierException | TimeoutException e) {
+ throw new AssertionError(e);
+ }
+ }
+
+ private static class BarrierGetMap<K, V> extends ConcurrentHashMap<K, V> {
+ private final CyclicBarrier barrier;
+
+ BarrierGetMap(int parties) {
+ this.barrier = new CyclicBarrier(parties);
+ }
+
+ @Override
+ public V get(Object key) {
+ V value = super.get(key);
+ if (value == null) {
+ await(barrier);
+ }
+ return value;
+ }
+ }
+}
diff --git
a/common/src/test/java/org/apache/rocketmq/common/ServiceThreadTest.java
b/common/src/test/java/org/apache/rocketmq/common/ServiceThreadTest.java
index e27fd497bd..9cb70f5210 100644
--- a/common/src/test/java/org/apache/rocketmq/common/ServiceThreadTest.java
+++ b/common/src/test/java/org/apache/rocketmq/common/ServiceThreadTest.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.common;
+import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@@ -159,17 +160,15 @@ public class ServiceThreadTest {
*/
@Test(timeout = 30000)
public void serviceThreadShouldNotLoseWakeupUnderStress() throws Exception
{
- final int stressIterations = 10000;
+ final int stressIterations = 1000;
final int wakerThreads = 4;
- final long waitTimeoutMs = 20;
- final long lostWakeupThresholdMs = 18;
-
- StressServiceThread service = new StressServiceThread();
- AtomicInteger activeIteration = new AtomicInteger(-1);
- AtomicInteger completedIteration = new AtomicInteger(-1);
- AtomicInteger lostWakeups = new AtomicInteger(0);
- AtomicInteger maxElapsedMs = new AtomicInteger(0);
- AtomicBoolean running = new AtomicBoolean(true);
+ final long waitTimeoutMs = TimeUnit.MINUTES.toMillis(2);
+ final long coordinationTimeoutSeconds = 20;
+
+ CyclicBarrier iterationStart = new CyclicBarrier(wakerThreads + 1);
+ CyclicBarrier iterationComplete = new CyclicBarrier(wakerThreads + 1);
+ AtomicReference<StressServiceThread> activeService = new
AtomicReference<>();
+ AtomicInteger completedIterations = new AtomicInteger(0);
AtomicReference<Throwable> failure = new AtomicReference<>();
ExecutorService executor = Executors.newFixedThreadPool(wakerThreads +
1);
@@ -177,32 +176,28 @@ public class ServiceThreadTest {
executor.submit(() -> {
try {
for (int i = 0; i < stressIterations; i++) {
- activeIteration.set(i);
- long elapsed = service.awaitOnce(waitTimeoutMs);
- maxElapsedMs.accumulateAndGet((int) elapsed,
Math::max);
- if (elapsed >= lostWakeupThresholdMs) {
- lostWakeups.incrementAndGet();
- running.set(false);
- break;
- }
- completedIteration.set(i);
- Thread.yield();
+ StressServiceThread service = new
StressServiceThread();
+ activeService.set(service);
+ iterationStart.await(coordinationTimeoutSeconds,
TimeUnit.SECONDS);
+ service.awaitOnce(waitTimeoutMs);
+ completedIterations.incrementAndGet();
+ iterationComplete.await(coordinationTimeoutSeconds,
TimeUnit.SECONDS);
}
} catch (Throwable t) {
failure.compareAndSet(null, t);
- } finally {
- running.set(false);
}
});
for (int w = 0; w < wakerThreads; w++) {
executor.submit(() -> {
- while (running.get()) {
- int iteration = activeIteration.get();
- if (iteration >= 0 && completedIteration.get() <
iteration) {
- service.wakeup();
+ try {
+ for (int i = 0; i < stressIterations; i++) {
+ iterationStart.await(coordinationTimeoutSeconds,
TimeUnit.SECONDS);
+ activeService.get().wakeup();
+
iterationComplete.await(coordinationTimeoutSeconds, TimeUnit.SECONDS);
}
- Thread.yield();
+ } catch (Throwable t) {
+ failure.compareAndSet(null, t);
}
});
}
@@ -214,10 +209,9 @@ public class ServiceThreadTest {
if (error != null) {
throw new AssertionError("stress test failed", error);
}
- assertEquals("ServiceThread lost wakeups under stress
(maxElapsedMs=" + maxElapsedMs.get() + ")",
- 0, lostWakeups.get());
+ assertEquals("ServiceThread must complete every notified wait",
stressIterations,
+ completedIterations.get());
} finally {
- running.set(false);
executor.shutdownNow();
}
}
@@ -283,10 +277,8 @@ public class ServiceThreadTest {
public void run() {
}
- long awaitOnce(long intervalMillis) {
- long begin = System.nanoTime();
+ void awaitOnce(long intervalMillis) {
waitForRunning(intervalMillis);
- return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - begin);
}
}
}
diff --git
a/test/src/test/java/org/apache/rocketmq/test/client/consumer/pop/BatchAckIT.java
b/test/src/test/java/org/apache/rocketmq/test/client/consumer/pop/BatchAckIT.java
index ec9153ccc9..fdc8118d64 100644
---
a/test/src/test/java/org/apache/rocketmq/test/client/consumer/pop/BatchAckIT.java
+++
b/test/src/test/java/org/apache/rocketmq/test/client/consumer/pop/BatchAckIT.java
@@ -47,6 +47,8 @@ import static org.awaitility.Awaitility.await;
import static org.junit.Assert.assertEquals;
public class BatchAckIT extends BasePop {
+ private static final int QUEUE_COUNT = 8;
+ private static final Duration POP_ASSERT_TIMEOUT = Duration.ofSeconds(30);
protected String topic;
protected String group;
@@ -60,7 +62,8 @@ public class BatchAckIT extends BasePop {
brokerAddr = brokerController1.getBrokerAddr();
topic = MQRandomUtils.getRandomTopic();
group = initConsumerGroup();
- IntegrationTestBase.initTopic(topic, NAMESRV_ADDR, BROKER1_NAME, 8,
CQType.SimpleCQ, TopicMessageType.NORMAL);
+ IntegrationTestBase.initTopic(topic, NAMESRV_ADDR, BROKER1_NAME,
QUEUE_COUNT, CQType.SimpleCQ,
+ TopicMessageType.NORMAL);
producer = getProducer(NAMESRV_ADDR, topic);
client = getRMQPopClient();
messageQueue = new MessageQueue(topic, BROKER1_NAME, -1);
@@ -113,8 +116,10 @@ public class BatchAckIT extends BasePop {
public void testBatchAck(Supplier<PopResult> popResultSupplier) throws
Throwable {
// Send 10 messages but do not ack, let them enter the retry topic
producer.send(10);
+ awaitStoredMessageCount(10);
AtomicInteger firstMsgRcvNum = new AtomicInteger();
- await().atMost(Duration.ofSeconds(3)).untilAsserted(() -> {
+ // A single POP long poll can take up to three seconds, so leave
enough time for retries on a busy CI runner.
+ await().atMost(POP_ASSERT_TIMEOUT).untilAsserted(() -> {
PopResult popResult = popResultSupplier.get();
if (popResult.getPopStatus().equals(PopStatus.FOUND)) {
firstMsgRcvNum.addAndGet(popResult.getMsgFoundList().size());
@@ -125,8 +130,9 @@ public class BatchAckIT extends BasePop {
TimeUnit.SECONDS.sleep(6);
producer.send(20);
+ awaitStoredMessageCount(30);
List<String> extraInfoList = new ArrayList<>();
- await().atMost(Duration.ofSeconds(3)).untilAsserted(() -> {
+ await().atMost(POP_ASSERT_TIMEOUT).untilAsserted(() -> {
PopResult popResult = popResultSupplier.get();
if (popResult.getPopStatus().equals(PopStatus.FOUND)) {
for (MessageExt messageExt : popResult.getMsgFoundList()) {
@@ -145,6 +151,18 @@ public class BatchAckIT extends BasePop {
assertEquals(PopStatus.POLLING_NOT_FOUND, popResult.getPopStatus());
}
+ private void awaitStoredMessageCount(int expectedCount) {
+ // Sending completes before consume-queue dispatch necessarily catches
up. Starting an orderly POP too early can
+ // lock a partially dispatched queue and prevent the remainder from
being returned by a subsequent POP.
+ await().atMost(POP_ASSERT_TIMEOUT).untilAsserted(() -> {
+ long storedMessageCount = 0;
+ for (int queueId = 0; queueId < QUEUE_COUNT; queueId++) {
+ storedMessageCount +=
brokerController1.getMessageStore().getMaxOffsetInQueue(topic, queueId);
+ }
+ assertEquals(expectedCount, storedMessageCount);
+ });
+ }
+
private CompletableFuture<PopResult> popMessageAsync() {
return client.popMessageAsync(
brokerAddr, messageQueue, Duration.ofSeconds(3).toMillis(), 30,
group, 3000, false,
diff --git
a/test/src/test/java/org/apache/rocketmq/test/client/producer/querymsg/QueryMsgByKeyIT.java
b/test/src/test/java/org/apache/rocketmq/test/client/producer/querymsg/QueryMsgByKeyIT.java
index 69dd26cf84..87ae5285e8 100644
---
a/test/src/test/java/org/apache/rocketmq/test/client/producer/querymsg/QueryMsgByKeyIT.java
+++
b/test/src/test/java/org/apache/rocketmq/test/client/producer/querymsg/QueryMsgByKeyIT.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.test.client.producer.querymsg;
+import java.time.Duration;
import java.util.List;
import org.apache.rocketmq.client.exception.MQClientException;
@@ -33,6 +34,7 @@ import org.junit.Before;
import org.junit.Test;
import static com.google.common.truth.Truth.assertThat;
+import static org.awaitility.Awaitility.await;
public class QueryMsgByKeyIT extends BaseConf {
private static Logger logger =
LoggerFactory.getLogger(QueryMsgByKeyIT.class);
@@ -154,9 +156,12 @@ public class QueryMsgByKeyIT extends BaseConf {
long begin = System.currentTimeMillis() - 500000;
long end = System.currentTimeMillis() + 500000;
- List<MessageExt> list = producerA.getProducer().queryMessage(topicA,
keyA, msgSize * 10, begin, end).getMessageList();
-
- assertThat(list).isNotNull();
- assertThat(list.size()).isEqualTo(1);
+ // Message indexes are built asynchronously, so querying immediately
after send can temporarily return no message.
+
await().ignoreException(MQClientException.class).atMost(Duration.ofSeconds(30)).untilAsserted(()
-> {
+ List<MessageExt> list = producerA.getProducer()
+ .queryMessage(topicA, keyA, msgSize * 10, begin,
end).getMessageList();
+ assertThat(list).isNotNull();
+ assertThat(list.size()).isEqualTo(1);
+ });
}
}
