Skip to content

[ISSUE #10906] Prevent duplicate MQClientInstance creation #10907

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lizhimins merged 4 commits into from
Aug 12, 2026
Merged

[ISSUE #10906] Prevent duplicate MQClientInstance creation #10907

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
571f74a
[ISSUE #10906] Prevent duplicate MQClientInstance creation
qianye1001 Aug 11, 2026
5b9e638
[ISSUE #10906] Document client factory lifecycle assumptions
qianye1001 Aug 11, 2026
52184f8
Stabilize asynchronous integration tests
qianye1001 Aug 11, 2026
1385ac7
Stabilize ServiceThread wakeup stress test
qianye1001 Aug 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 17 additions & 10 deletions client/src/main/java/org/apache/rocketmq/client/impl/MQClientManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,16 +51,15 @@ 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 {
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;
Expand All @@ -82,10 +81,18 @@ 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);
}

public void removeClientFactory(final String clientId, final MQClientInstance instance) {
this.factoryTable.remove(clientId, instance);
}

public ConcurrentMap<String, MQClientInstance> getFactoryTable() {
return factoryTable;
}
Expand Down
204 changes: 126 additions & 78 deletions client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,86 +148,114 @@ 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<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) {
Expand Down Expand Up @@ -313,22 +341,31 @@ 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) {
// 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);
Expand All @@ -338,6 +375,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(() -> {
Expand Down Expand Up @@ -1077,7 +1125,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:
Expand Down
Loading
Loading