This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new 5fd8b42be fix(cluster): preserve interrupts from pooled client actions
(#6467)
5fd8b42be is described below
commit 5fd8b42be3f406bb6184e3b2684b372eedbb67dd
Author: Loyal-Young <[email protected]>
AuthorDate: Thu Oct 15 18:12:10 2026 +0800
fix(cluster): preserve interrupts from pooled client actions (#6467)
---
.../studio/cluster/broker/MqAdminExtFactory.java | 3 ++
.../studio/cluster/broker/MqClientPool.java | 3 ++
.../cluster/broker/MqAdminExtFactoryTest.java | 36 +++++++++++++
.../studio/cluster/broker/MqClientPoolTest.java | 60 ++++++++++++++++++++++
4 files changed, 102 insertions(+)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/MqAdminExtFactory.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/MqAdminExtFactory.java
index 5acdcc3e0..72b81eafa 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/MqAdminExtFactory.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/MqAdminExtFactory.java
@@ -131,6 +131,9 @@ public class MqAdminExtFactory {
} catch (BusinessException ex) {
throw ex;
} catch (Exception ex) {
+ if (ex instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
log.warn("Admin action failed against namesrv {}: {}",
namesrvAddr, ex.getMessage());
throw new BusinessException(502, "RocketMQ admin call failed: " +
rootMessage(ex));
} finally {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/MqClientPool.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/MqClientPool.java
index e95bcda33..0ee8e3e96 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/MqClientPool.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/MqClientPool.java
@@ -122,6 +122,9 @@ public class MqClientPool {
} catch (BusinessException ex) {
throw ex;
} catch (Exception ex) {
+ if (ex instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
log.warn("RocketMQ client action failed against namesrv {}: {}",
cacheKey.namesrvAddr(), ex.getMessage());
throw new BusinessException(502, "RocketMQ client call failed: " +
rootMessage(ex));
} finally {
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/MqAdminExtFactoryTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/MqAdminExtFactoryTest.java
index a1538e629..44b1a2784 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/MqAdminExtFactoryTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/MqAdminExtFactoryTest.java
@@ -54,6 +54,42 @@ class MqAdminExtFactoryTest {
}
}
+ @Test
+ void interruptedAdminActionShouldRestoreInterruptAndReleaseLeaseTest() {
+ DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
+ RecordingFactory factory = new RecordingFactory(admin);
+ try {
+ assertThatThrownBy(() -> factory.execute("namesrv:9876", null,
ignored -> {
+ Thread.currentThread().interrupt();
+ Thread.sleep(1);
+ return null;
+ })).isInstanceOf(BusinessException.class)
+ .hasMessageContaining("RocketMQ admin call failed");
+ assertThat(Thread.currentThread().isInterrupted()).isTrue();
+ } finally {
+ Thread.interrupted();
+ factory.shutdown();
+ }
+ verify(admin).shutdown();
+ }
+
+ @Test
+ void ordinaryAdminFailureShouldNotSetInterruptTest() {
+ DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
+ RecordingFactory factory = new RecordingFactory(admin);
+ try {
+ assertThatThrownBy(() -> factory.execute("namesrv:9876", null,
ignored -> {
+ throw new IllegalStateException("admin failure");
+ })).isInstanceOf(BusinessException.class)
+ .hasMessage("RocketMQ admin call failed: admin failure");
+ assertThat(Thread.currentThread().isInterrupted()).isFalse();
+ } finally {
+ Thread.interrupted();
+ factory.shutdown();
+ }
+ verify(admin).shutdown();
+ }
+
@Test
void executeShouldRejectBlankNamesrvAddr() {
MqAdminExtFactory factory = new MqAdminExtFactory();
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/MqClientPoolTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/MqClientPoolTest.java
index ab6ac0f42..f8bbcc7e8 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/MqClientPoolTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/MqClientPoolTest.java
@@ -18,6 +18,7 @@ package org.apache.rocketmq.studio.cluster.broker;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.junit.jupiter.api.Test;
import org.mockito.MockedConstruction;
@@ -26,12 +27,71 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.mockConstruction;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
class MqClientPoolTest {
+ @Test
+ void interruptedProducerActionShouldRestoreInterruptAndReleaseLeaseTest() {
+ try (MockedConstruction<DefaultMQProducer> construction =
mockConstruction(DefaultMQProducer.class)) {
+ MqClientPool pool = new MqClientPool();
+ try {
+ assertThatThrownBy(() -> pool.withProducer("namesrv:9876",
null, "credential-a", ignored -> {
+ Thread.currentThread().interrupt();
+ Thread.sleep(1);
+ return null;
+ })).isInstanceOf(BusinessException.class)
+ .hasMessageContaining("RocketMQ client call failed");
+ assertThat(Thread.currentThread().isInterrupted()).isTrue();
+ } finally {
+ Thread.interrupted();
+ pool.shutdown();
+ }
+ verify(construction.constructed().getFirst()).shutdown();
+ }
+ }
+
+ @Test
+ void interruptedPullActionShouldRestoreInterruptAndReleaseLeaseTest() {
+ try (MockedConstruction<DefaultMQPullConsumer> construction =
mockConstruction(DefaultMQPullConsumer.class)) {
+ MqClientPool pool = new MqClientPool();
+ try {
+ assertThatThrownBy(() -> pool.withPullConsumer("namesrv:9876",
null, "credential-a", ignored -> {
+ Thread.currentThread().interrupt();
+ Thread.sleep(1);
+ return null;
+ })).isInstanceOf(BusinessException.class)
+ .hasMessageContaining("RocketMQ client call failed");
+ assertThat(Thread.currentThread().isInterrupted()).isTrue();
+ } finally {
+ Thread.interrupted();
+ pool.shutdown();
+ }
+ verify(construction.constructed().getFirst()).shutdown();
+ }
+ }
+
+ @Test
+ void ordinaryProducerFailureShouldNotSetInterruptTest() {
+ try (MockedConstruction<DefaultMQProducer> construction =
mockConstruction(DefaultMQProducer.class)) {
+ MqClientPool pool = new MqClientPool();
+ try {
+ assertThatThrownBy(() -> pool.withProducer("namesrv:9876",
null, "credential-a", ignored -> {
+ throw new IllegalStateException("producer failure");
+ })).isInstanceOf(BusinessException.class)
+ .hasMessage("RocketMQ client call failed: producer
failure");
+ assertThat(Thread.currentThread().isInterrupted()).isFalse();
+ } finally {
+ Thread.interrupted();
+ pool.shutdown();
+ }
+ verify(construction.constructed().getFirst()).shutdown();
+ }
+ }
+
@Test
void releaseShouldWaitForInFlightProducerBeforeShutdownTest() throws
Exception {
try (MockedConstruction<DefaultMQProducer> construction =
