From 571f74ada81f98b2e2082781c88b245a59bb724f Mon Sep 17 00:00:00 2001 From: qianye Date: Tue, 11 Aug 2026 15:00:43 +0800 Subject: [PATCH 1/4] [ISSUE #10906] Prevent duplicate MQClientInstance creation Serialize construction per client ID and roll back resources when construction or start fails. Use identity-aware removal so stale shutdown cannot remove a replacement instance. Signed-off-by: qianye --- .../rocketmq/client/impl/MQClientManager.java | 19 +- .../client/impl/factory/MQClientInstance.java | 200 ++++---- .../client/impl/MQClientManagerTest.java | 433 ++++++++++++++++++ 3 files changed, 565 insertions(+), 87 deletions(-) create mode 100644 client/src/test/java/org/apache/rocketmq/client/impl/MQClientManagerTest.java 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 ca6f4617456..0d4580221e8 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,13 @@ public MQClientInstance getOrCreateMQClientInstance(final ClientConfig clientCon 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 { + ClientConfig clonedClientConfig = clientConfig.cloneClientConfig(); + instance = this.factoryTable.computeIfAbsent(clientId, key -> { + MQClientInstance newInstance = new MQClientInstance(clonedClientConfig, + this.factoryIndexGenerator.getAndIncrement(), key, rpcHook); log.info("Created new MQClientInstance for clientId:[{}]", clientId); - } + return newInstance; + }); } return instance; @@ -86,6 +83,10 @@ 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 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 cd45fed2a3a..7cc5bc0e705 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,113 @@ public MQClientInstance(ClientConfig clientConfig, int instanceIndex, String cli } 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> 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> 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> addressEntry : brokerAddrTable.entrySet()) { - for (Map.Entry 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> addressEntry : brokerAddrTable.entrySet()) { + for (Map.Entry 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) { + if (t != cause) { + cause.addSuppressed(t); + } + } } public static TopicPublishInfo topicRouteData2TopicPublishInfo(final String topic, final TopicRouteData route) { @@ -313,22 +340,28 @@ public void start() throws MQClientException { 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) { + 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 +371,17 @@ public void start() throws MQClientException { } } + 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 +1121,7 @@ public void shutdown() { 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 00000000000..beedf624a4b --- /dev/null +++ b/client/src/test/java/org/apache/rocketmq/client/impl/MQClientManagerTest.java @@ -0,0 +1,433 @@ +/* + * 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 instancesToDispose = + Collections.newSetFromMap(new IdentityHashMap()); + + @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> futures = new ArrayList<>(); + for (int i = 0; i < callers; i++) { + futures.add(executor.submit(() -> manager.getOrCreateMQClientInstance(config))); + } + + Set returned = Collections.newSetFromMap( + new IdentityHashMap()); + for (Future 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 firstFuture = executor.submit( + () -> manager.getOrCreateMQClientInstance(firstConfig)); + Future 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"); + config.setEnableConcurrentHeartbeat(true); + config.setConcurrentHeartbeatThreadPoolSize(0); + + try { + Future 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(); + 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 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) { + 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 extends ConcurrentHashMap { + 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; + } + } +} From 5b9e6383bfc1c44ffcf74978e5c91fc29af0723b Mon Sep 17 00:00:00 2001 From: qianye Date: Tue, 11 Aug 2026 15:48:39 +0800 Subject: [PATCH 2/4] [ISSUE #10906] Document client factory lifecycle assumptions Clarify computeIfAbsent recursion constraints, failed-start cleanup semantics, identity-aware removal, and test implementation assumptions. Signed-off-by: qianye --- .../org/apache/rocketmq/client/impl/MQClientManager.java | 8 +++++++- .../rocketmq/client/impl/factory/MQClientInstance.java | 4 ++++ .../apache/rocketmq/client/impl/MQClientManagerTest.java | 7 +++++++ 3 files changed, 18 insertions(+), 1 deletion(-) 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 0d4580221e8..e6f17727ffb 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 @@ -52,10 +52,12 @@ public MQClientInstance getOrCreateMQClientInstance(final ClientConfig clientCon MQClientInstance instance = this.factoryTable.get(clientId); if (null == instance) { 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:[{}]", clientId); + log.info("Created new MQClientInstance for clientId:[{}]", key); return newInstance; }); } @@ -79,6 +81,10 @@ public ProduceAccumulator getOrCreateProduceAccumulator(final ClientConfig clien 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); } 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 7cc5bc0e705..5208e49fe7e 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 @@ -251,6 +251,7 @@ 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); } @@ -358,6 +359,9 @@ public void start() throws MQClientException { 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; 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 index beedf624a4b..82aea4c4219 100644 --- a/client/src/test/java/org/apache/rocketmq/client/impl/MQClientManagerTest.java +++ b/client/src/test/java/org/apache/rocketmq/client/impl/MQClientManagerTest.java @@ -134,6 +134,8 @@ public void constructorFailureRollsBackResourcesAndAllowsRetry() throws Exceptio 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); @@ -228,6 +230,8 @@ public void startFailureRollsBackResourcesAndAllowsReplacement() throws Exceptio 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); @@ -307,6 +311,9 @@ public int getClientCallbackExecutorThreads() { } 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; From 52184f84f17fdf5c8ab86ffe6593d5f147666fe6 Mon Sep 17 00:00:00 2001 From: qianye Date: Tue, 11 Aug 2026 17:08:35 +0800 Subject: [PATCH 3/4] Stabilize asynchronous integration tests --- .../test/client/consumer/pop/BatchAckIT.java | 24 ++++++++++++++++--- .../producer/querymsg/QueryMsgByKeyIT.java | 13 ++++++---- 2 files changed, 30 insertions(+), 7 deletions(-) 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 ec9153ccc98..fdc8118d64a 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.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 void setUp() { 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 void testBatchAckOrderly() throws Throwable { public void testBatchAck(Supplier 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 void testBatchAck(Supplier popResultSupplier) throws Throwable TimeUnit.SECONDS.sleep(6); producer.send(20); + awaitStoredMessageCount(30); List 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 void testBatchAck(Supplier popResultSupplier) throws Throwable 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 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 69dd26cf845..87ae5285e8c 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.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 void testQueryMsgWithSameHash2() throws Exception { long begin = System.currentTimeMillis() - 500000; long end = System.currentTimeMillis() + 500000; - List 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 list = producerA.getProducer() + .queryMessage(topicA, keyA, msgSize * 10, begin, end).getMessageList(); + assertThat(list).isNotNull(); + assertThat(list.size()).isEqualTo(1); + }); } } From 1385ac7975d48b68dab9a6149f3b23df3a3ebf5f Mon Sep 17 00:00:00 2001 From: qianye Date: Tue, 11 Aug 2026 17:46:24 +0800 Subject: [PATCH 4/4] Stabilize ServiceThread wakeup stress test --- .../rocketmq/common/ServiceThreadTest.java | 58 ++++++++----------- 1 file changed, 25 insertions(+), 33 deletions(-) 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 e27fd497bdc..9cb70f52108 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 void testNoWakeupLostUnderStress() throws Exception { */ @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 activeService = new AtomicReference<>(); + AtomicInteger completedIterations = new AtomicInteger(0); AtomicReference failure = new AtomicReference<>(); ExecutorService executor = Executors.newFixedThreadPool(wakerThreads + 1); @@ -177,32 +176,28 @@ public void serviceThreadShouldNotLoseWakeupUnderStress() throws Exception { 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 void serviceThreadShouldNotLoseWakeupUnderStress() throws Exception { 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 String getServiceName() { public void run() { } - long awaitOnce(long intervalMillis) { - long begin = System.nanoTime(); + void awaitOnce(long intervalMillis) { waitForRunning(intervalMillis); - return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - begin); } } }