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 eedd7607de [ISSUE #11083] Release owned gRPC event loops and stabilize
POP priority tests (#11084)
eedd7607de is described below
commit eedd7607de97368c0dcb45a4c28867ce3226df24
Author: qianye <[email protected]>
AuthorDate: Thu Sep 10 13:45:34 2026 +0800
[ISSUE #11083] Release owned gRPC event loops and stabilize POP priority
tests (#11084)
---
.github/workflows/bazel.yml | 10 +-
proxy/BUILD.bazel | 1 +
.../org/apache/rocketmq/proxy/grpc/GrpcServer.java | 41 +++-
.../rocketmq/proxy/grpc/GrpcServerBuilder.java | 95 +++++++--
.../apache/rocketmq/proxy/grpc/GrpcServerTest.java | 231 +++++++++++++++++++++
.../test/client/consumer/pop/PopPriorityIT.java | 99 ++++++---
6 files changed, 428 insertions(+), 49 deletions(-)
diff --git a/.github/workflows/bazel.yml b/.github/workflows/bazel.yml
index 666512017e..a19bcfc7d5 100644
--- a/.github/workflows/bazel.yml
+++ b/.github/workflows/bazel.yml
@@ -32,4 +32,12 @@ jobs:
- name: Build
run: bazel build --config=remote //...
- name: Run Tests
- run: bazel test --config=remote //...
\ No newline at end of file
+ run: bazel test --config=remote //...
+ - name: Upload Bazel test results
+ if: failure()
+ uses: actions/upload-artifact@v4
+ with:
+ name: bazel-test-results
+ path: bazel-testlogs/**/test.xml
+ if-no-files-found: ignore
+ retention-days: 7
diff --git a/proxy/BUILD.bazel b/proxy/BUILD.bazel
index 0711d48a66..8ddaa52382 100644
--- a/proxy/BUILD.bazel
+++ b/proxy/BUILD.bazel
@@ -95,6 +95,7 @@ java_library(
"@maven//:io_grpc_grpc_api",
"@maven//:io_grpc_grpc_context",
"@maven//:io_grpc_grpc_netty_shaded",
+ "@maven//:io_grpc_grpc_protobuf",
"@maven//:io_grpc_grpc_stub",
"@maven//:io_netty_netty_all",
"@maven//:io_github_aliyunmq_rocketmq_grpc_netty_codec_haproxy",
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServer.java
b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServer.java
index af3d6b4c6c..74b71b14ee 100644
--- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServer.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServer.java
@@ -19,6 +19,7 @@ package org.apache.rocketmq.proxy.grpc;
import com.google.common.annotations.VisibleForTesting;
import io.grpc.Server;
+import io.grpc.netty.shaded.io.netty.channel.EventLoopGroup;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.utils.StartAndShutdown;
import org.apache.rocketmq.logging.org.slf4j.Logger;
@@ -33,6 +34,8 @@ public class GrpcServer implements StartAndShutdown {
private static final Logger log =
LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
private final Server server;
+ private final EventLoopGroup ownedBossGroup;
+ private final EventLoopGroup ownedWorkerGroup;
private final long timeout;
@@ -43,7 +46,14 @@ public class GrpcServer implements StartAndShutdown {
protected GrpcServer(Server server, long timeout, TimeUnit unit,
TlsCertificateManager tlsCertificateManager) throws Exception {
+ this(server, timeout, unit, tlsCertificateManager, null, null);
+ }
+
+ GrpcServer(Server server, long timeout, TimeUnit unit,
TlsCertificateManager tlsCertificateManager,
+ EventLoopGroup ownedBossGroup, EventLoopGroup ownedWorkerGroup) throws
Exception {
this.server = server;
+ this.ownedBossGroup = ownedBossGroup;
+ this.ownedWorkerGroup = ownedWorkerGroup;
this.timeout = timeout;
this.unit = unit;
this.tlsCertificateManager = tlsCertificateManager;
@@ -51,11 +61,15 @@ public class GrpcServer implements StartAndShutdown {
}
public void start() throws Exception {
- // Register the TLS context reload handler
- tlsCertificateManager.registerReloadListener(this.tlsReloadHandler);
-
- this.server.start();
- log.info("grpc server start successfully.");
+ try {
+ // Register the TLS context reload handler
+
tlsCertificateManager.registerReloadListener(this.tlsReloadHandler);
+ this.server.start();
+ log.info("grpc server start successfully.");
+ } catch (Exception | Error e) {
+ shutdown();
+ throw e;
+ }
}
public void shutdown() {
@@ -63,12 +77,25 @@ public class GrpcServer implements StartAndShutdown {
// Unregister the TLS context reload handler
tlsCertificateManager.unregisterReloadListener(this.tlsReloadHandler);
- this.server.shutdown().awaitTermination(timeout, unit);
+ if (!this.server.shutdown().awaitTermination(timeout, unit)) {
+ this.server.shutdownNow().awaitTermination(timeout, unit);
+ }
log.info("grpc server shutdown successfully.");
+ } catch (InterruptedException e) {
+ this.server.shutdownNow();
+ Thread.currentThread().interrupt();
+ log.error("Interrupted while shutting down grpc server", e);
} catch (Exception e) {
- e.printStackTrace();
+ this.server.shutdownNow();
log.error("Failed to shutdown grpc server", e);
+ } finally {
+ if (ownedBossGroup != null) {
+ ownedBossGroup.shutdownGracefully();
+ }
+ if (ownedWorkerGroup != null) {
+ ownedWorkerGroup.shutdownGracefully();
+ }
}
}
diff --git
a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServerBuilder.java
b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServerBuilder.java
index f59c4982c4..35d035b993 100644
--- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServerBuilder.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServerBuilder.java
@@ -20,10 +20,12 @@ import io.grpc.BindableService;
import io.grpc.ServerInterceptor;
import io.grpc.ServerServiceDefinition;
import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
+import io.grpc.netty.shaded.io.netty.channel.EventLoopGroup;
import io.grpc.netty.shaded.io.netty.channel.epoll.EpollEventLoopGroup;
import io.grpc.netty.shaded.io.netty.channel.epoll.EpollServerSocketChannel;
import io.grpc.netty.shaded.io.netty.channel.nio.NioEventLoopGroup;
import io.grpc.netty.shaded.io.netty.channel.socket.nio.NioServerSocketChannel;
+import java.util.Objects;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.common.constant.LoggerName;
@@ -39,6 +41,11 @@ import
org.apache.rocketmq.proxy.service.cert.TlsCertificateManager;
public class GrpcServerBuilder {
private static final Logger log =
LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
protected NettyServerBuilder serverBuilder;
+ private final int bossLoopNum;
+ private final int workerLoopNum;
+ private final boolean enableEpoll;
+ private final EventLoopGroup bossGroup;
+ private final EventLoopGroup workerGroup;
protected long time = 30;
@@ -51,8 +58,50 @@ public class GrpcServerBuilder {
return new GrpcServerBuilder(executor, port, tlsCertificateManager);
}
+ /**
+ * Creates a server with internally owned event loops. Zero selects
Netty's default thread count.
+ */
+ public static GrpcServerBuilder newBuilder(ThreadPoolExecutor executor,
int port,
+ TlsCertificateManager tlsCertificateManager, int bossLoopNum, int
workerLoopNum) {
+ return new GrpcServerBuilder(executor, port, tlsCertificateManager,
bossLoopNum, workerLoopNum);
+ }
+
+ /**
+ * Creates a server using caller-owned event loops. Both groups must match
enableGrpcEpoll.
+ * The caller must close them after all servers using them have terminated.
+ */
+ public static GrpcServerBuilder newBuilder(ThreadPoolExecutor executor,
int port,
+ TlsCertificateManager tlsCertificateManager, EventLoopGroup bossGroup,
EventLoopGroup workerGroup) {
+ return new GrpcServerBuilder(executor, port, tlsCertificateManager,
bossGroup, workerGroup);
+ }
+
protected GrpcServerBuilder(ThreadPoolExecutor executor, int port,
TlsCertificateManager tlsCertificateManager) {
+ this(executor, port, tlsCertificateManager,
ConfigurationManager.getProxyConfig().getGrpcBossLoopNum(),
+ ConfigurationManager.getProxyConfig().getGrpcWorkerLoopNum());
+ }
+
+ protected GrpcServerBuilder(ThreadPoolExecutor executor, int port,
TlsCertificateManager tlsCertificateManager,
+ int bossLoopNum, int workerLoopNum) {
+ this(executor, port, tlsCertificateManager, bossLoopNum,
workerLoopNum, null, null);
+ }
+
+ protected GrpcServerBuilder(ThreadPoolExecutor executor, int port,
TlsCertificateManager tlsCertificateManager,
+ EventLoopGroup bossGroup, EventLoopGroup workerGroup) {
+ this(executor, port, tlsCertificateManager, 0, 0,
+ Objects.requireNonNull(bossGroup, "bossGroup"),
Objects.requireNonNull(workerGroup, "workerGroup"));
+ }
+
+ private GrpcServerBuilder(ThreadPoolExecutor executor, int port,
TlsCertificateManager tlsCertificateManager,
+ int bossLoopNum, int workerLoopNum, EventLoopGroup bossGroup,
EventLoopGroup workerGroup) {
+ if (bossLoopNum < 0 || workerLoopNum < 0) {
+ throw new IllegalArgumentException("Event loop thread counts must
not be negative");
+ }
ProxyConfig config = ConfigurationManager.getProxyConfig();
+ this.bossLoopNum = bossLoopNum;
+ this.workerLoopNum = workerLoopNum;
+ this.enableEpoll = config.isEnableGrpcEpoll();
+ this.bossGroup = bossGroup;
+ this.workerGroup = workerGroup;
this.tlsCertificateManager = tlsCertificateManager;
serverBuilder = NettyServerBuilder.forPort(port)
.maxConcurrentCallsPerConnection(config.getGrpcMaxConcurrentCallsPerConnection());
@@ -60,30 +109,20 @@ public class GrpcServerBuilder {
serverBuilder.protocolNegotiator(new ProxyAndTlsProtocolNegotiator());
// build server
- int bossLoopNum = config.getGrpcBossLoopNum();
- int workerLoopNum = config.getGrpcWorkerLoopNum();
int maxInboundMessageSize = config.getGrpcMaxInboundMessageSize();
long idleTimeMills = config.getGrpcClientIdleTimeMills();
- if (config.isEnableGrpcEpoll()) {
- serverBuilder.bossEventLoopGroup(new
EpollEventLoopGroup(bossLoopNum))
- .workerEventLoopGroup(new EpollEventLoopGroup(workerLoopNum))
- .channelType(EpollServerSocketChannel.class)
- .executor(executor);
- } else {
- serverBuilder.bossEventLoopGroup(new
NioEventLoopGroup(bossLoopNum))
- .workerEventLoopGroup(new NioEventLoopGroup(workerLoopNum))
- .channelType(NioServerSocketChannel.class)
- .executor(executor);
- }
+ serverBuilder.channelType(enableEpoll ? EpollServerSocketChannel.class
: NioServerSocketChannel.class)
+ .executor(executor);
serverBuilder.maxInboundMessageSize(maxInboundMessageSize)
.maxConnectionIdle(idleTimeMills, TimeUnit.MILLISECONDS)
.permitKeepAliveTime(config.getGrpcServerPermitKeepAliveTimeMillis(),
TimeUnit.MILLISECONDS)
.permitKeepAliveWithoutCalls(config.isGrpcServerPermitKeepAliveWithoutCalls());
- log.info("grpc server has built. port: {}, bossLoopNum: {},
workerLoopNum: {}, maxInboundMessageSize: {}",
- port, bossLoopNum, workerLoopNum, maxInboundMessageSize);
+ log.info("grpc server builder initialized. port: {}, bossLoopNum: {},
workerLoopNum: {}, "
+ + "callerOwnedEventLoops: {}, maxInboundMessageSize: {}",
+ port, bossLoopNum, workerLoopNum, bossGroup != null,
maxInboundMessageSize);
}
public GrpcServerBuilder shutdownTime(long time, TimeUnit unit) {
@@ -108,7 +147,31 @@ public class GrpcServerBuilder {
}
public GrpcServer build() throws Exception {
- return new GrpcServer(this.serverBuilder.build(), time, unit,
tlsCertificateManager);
+ EventLoopGroup boss = bossGroup;
+ EventLoopGroup worker = workerGroup;
+ boolean ownsEventLoops = boss == null;
+ try {
+ if (ownsEventLoops) {
+ boss = newEventLoopGroup(bossLoopNum);
+ worker = newEventLoopGroup(workerLoopNum);
+ }
+ return new
GrpcServer(serverBuilder.bossEventLoopGroup(boss).workerEventLoopGroup(worker).build(),
+ time, unit, tlsCertificateManager, ownsEventLoops ? boss :
null, ownsEventLoops ? worker : null);
+ } catch (Exception | Error e) {
+ if (ownsEventLoops) {
+ if (boss != null) {
+ boss.shutdownGracefully();
+ }
+ if (worker != null) {
+ worker.shutdownGracefully();
+ }
+ }
+ throw e;
+ }
+ }
+
+ private EventLoopGroup newEventLoopGroup(int threads) {
+ return enableEpoll ? new EpollEventLoopGroup(threads) : new
NioEventLoopGroup(threads);
}
public GrpcServerBuilder configInterceptor() {
diff --git
a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/GrpcServerTest.java
b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/GrpcServerTest.java
new file mode 100644
index 0000000000..4e9624fc49
--- /dev/null
+++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/GrpcServerTest.java
@@ -0,0 +1,231 @@
+/*
+ * 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.proxy.grpc;
+
+import com.google.protobuf.StringValue;
+import io.grpc.CallOptions;
+import io.grpc.ManagedChannel;
+import io.grpc.MethodDescriptor;
+import io.grpc.Server;
+import io.grpc.ServerServiceDefinition;
+import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
+import io.grpc.netty.shaded.io.netty.channel.EventLoopGroup;
+import io.grpc.netty.shaded.io.netty.channel.MultithreadEventLoopGroup;
+import io.grpc.netty.shaded.io.netty.channel.epoll.Epoll;
+import io.grpc.netty.shaded.io.netty.channel.nio.NioEventLoopGroup;
+import io.grpc.protobuf.ProtoUtils;
+import io.grpc.stub.ClientCalls;
+import io.grpc.stub.ServerCalls;
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import org.apache.rocketmq.proxy.config.ConfigurationManager;
+import org.apache.rocketmq.proxy.config.InitConfigTest;
+import org.apache.rocketmq.proxy.config.ProxyConfig;
+import org.apache.rocketmq.proxy.service.cert.TlsCertificateManager;
+import org.junit.After;
+import org.junit.Assume;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class GrpcServerTest extends InitConfigTest {
+ private final ThreadPoolExecutor executor = (ThreadPoolExecutor)
Executors.newFixedThreadPool(2);
+ private final TlsCertificateManager certificates =
mock(TlsCertificateManager.class);
+ private static final MethodDescriptor<StringValue, StringValue> METHOD =
+ MethodDescriptor.<StringValue, StringValue>newBuilder()
+ .setType(MethodDescriptor.MethodType.UNARY)
+ .setFullMethodName("test.Echo/Ping")
+
.setRequestMarshaller(ProtoUtils.marshaller(StringValue.getDefaultInstance()))
+
.setResponseMarshaller(ProtoUtils.marshaller(StringValue.getDefaultInstance()))
+ .build();
+
+ @Before
+ public void configure() {
+ ProxyConfig config = ConfigurationManager.getProxyConfig();
+ config.setEnableGrpcEpoll(false);
+ config.setTlsTestModeEnable(true);
+ config.setGrpcBossLoopNum(2);
+ config.setGrpcWorkerLoopNum(3);
+ }
+
+ @After
+ public void cleanup() {
+ executor.shutdownNow();
+ }
+
+ @Test
+ public void testDefaultOwnedGroupsTerminate() throws Exception {
+ checkOwnedGroups(GrpcServerBuilder.newBuilder(executor, 0,
certificates), 2, 3);
+ }
+
+ @Test
+ public void testExplicitOwnedGroupCounts() throws Exception {
+ checkOwnedGroups(GrpcServerBuilder.newBuilder(executor, 0,
certificates, 1, 1), 1, 1);
+ }
+
+ @Test
+ public void testEpollOwnedGroupsTerminate() throws Exception {
+ Assume.assumeTrue(Epoll.isAvailable());
+ ConfigurationManager.getProxyConfig().setEnableGrpcEpoll(true);
+ checkOwnedGroups(GrpcServerBuilder.newBuilder(executor, 0,
certificates, 1, 1), 1, 1);
+ }
+
+ private void checkOwnedGroups(GrpcServerBuilder builder, int bossThreads,
int workerThreads) throws Exception {
+ GrpcServer server = builder.addService(service()).build();
+ MultithreadEventLoopGroup boss = (MultithreadEventLoopGroup)
field(server, "ownedBossGroup");
+ MultithreadEventLoopGroup worker = (MultithreadEventLoopGroup)
field(server, "ownedWorkerGroup");
+ try {
+ assertThat(boss.executorCount()).isEqualTo(bossThreads);
+ assertThat(worker.executorCount()).isEqualTo(workerThreads);
+ server.start();
+ ping(server);
+ } finally {
+ server.shutdown();
+ }
+ assertTerminated(boss, worker);
+ verify(certificates).unregisterReloadListener(server.tlsReloadHandler);
+ }
+
+ @Test
+ public void testBorrowedGroupsSurviveShutdownAndBindFailure() throws
Exception {
+ EventLoopGroup boss = new NioEventLoopGroup(1);
+ EventLoopGroup worker = new NioEventLoopGroup(1);
+ GrpcServer first = GrpcServerBuilder.newBuilder(executor, 0,
certificates, boss, worker)
+ .addService(service()).build();
+ GrpcServer second = GrpcServerBuilder.newBuilder(executor, 0,
certificates, boss, worker)
+ .addService(service()).build();
+ GrpcServer failed = null;
+ try {
+ first.start();
+ second.start();
+ ping(first);
+ ping(second);
+ first.shutdown();
+ ping(second);
+ assertThat(boss.isShuttingDown()).isFalse();
+ assertThat(worker.isShuttingDown()).isFalse();
+
+ failed = GrpcServerBuilder.newBuilder(executor, port(second),
certificates, boss, worker).build();
+ assertThatThrownBy(failed::start).isInstanceOf(IOException.class);
+ ping(second);
+ assertThat(boss.isShuttingDown()).isFalse();
+ assertThat(worker.isShuttingDown()).isFalse();
+ } finally {
+ first.shutdown();
+ second.shutdown();
+ if (failed != null) {
+ failed.shutdown();
+ }
+ boss.shutdownGracefully();
+ worker.shutdownGracefully();
+ }
+ assertTerminated(boss, worker);
+ }
+
+ @Test
+ public void testOwnedGroupsTerminateAfterBindFailure() throws Exception {
+ GrpcServer running = GrpcServerBuilder.newBuilder(executor, 0,
certificates, 1, 1).build();
+ GrpcServer failed = null;
+ try {
+ running.start();
+ failed = GrpcServerBuilder.newBuilder(executor, port(running),
certificates, 1, 1).build();
+ assertThatThrownBy(failed::start).isInstanceOf(IOException.class);
+ assertTerminated((EventLoopGroup) field(failed, "ownedBossGroup"),
+ (EventLoopGroup) field(failed, "ownedWorkerGroup"));
+ } finally {
+ running.shutdown();
+ if (failed != null) {
+ failed.shutdown();
+ }
+ }
+ }
+
+ @Test
+ public void testShutdownTimeoutReleasesOwnedGroups() throws Exception {
+ Server delegate = mock(Server.class);
+ EventLoopGroup boss = mock(EventLoopGroup.class);
+ EventLoopGroup worker = mock(EventLoopGroup.class);
+ when(delegate.shutdown()).thenReturn(delegate);
+ when(delegate.shutdownNow()).thenReturn(delegate);
+ when(delegate.awaitTermination(1, TimeUnit.SECONDS)).thenReturn(false,
true);
+ new GrpcServer(delegate, 1, TimeUnit.SECONDS, certificates, boss,
worker).shutdown();
+ verify(delegate).shutdownNow();
+ verify(boss).shutdownGracefully();
+ verify(worker).shutdownGracefully();
+ }
+
+ @Test
+ public void testInterruptedShutdownReleasesOwnedGroups() throws Exception {
+ Server delegate = mock(Server.class);
+ EventLoopGroup boss = mock(EventLoopGroup.class);
+ EventLoopGroup worker = mock(EventLoopGroup.class);
+ when(delegate.shutdown()).thenReturn(delegate);
+ when(delegate.awaitTermination(1, TimeUnit.SECONDS)).thenThrow(new
InterruptedException());
+ try {
+ new GrpcServer(delegate, 1, TimeUnit.SECONDS, certificates, boss,
worker).shutdown();
+ assertThat(Thread.currentThread().isInterrupted()).isTrue();
+ verify(delegate).shutdownNow();
+ verify(boss).shutdownGracefully();
+ verify(worker).shutdownGracefully();
+ } finally {
+ Thread.interrupted();
+ }
+ }
+
+ private ServerServiceDefinition service() {
+ return ServerServiceDefinition.builder("test.Echo")
+ .addMethod(METHOD, ServerCalls.asyncUnaryCall((request, response)
-> {
+ response.onNext(request);
+ response.onCompleted();
+ })).build();
+ }
+
+ private int port(GrpcServer server) throws Exception {
+ return ((Server) field(server, "server")).getPort();
+ }
+
+ private void ping(GrpcServer server) throws Exception {
+ ManagedChannel channel = NettyChannelBuilder.forAddress("127.0.0.1",
port(server)).usePlaintext().build();
+ try {
+ assertThat(ClientCalls.blockingUnaryCall(channel, METHOD,
+ CallOptions.DEFAULT.withDeadlineAfter(5, TimeUnit.SECONDS),
StringValue.of("ping")))
+ .isEqualTo(StringValue.of("ping"));
+ } finally {
+ channel.shutdownNow().awaitTermination(5, TimeUnit.SECONDS);
+ }
+ }
+
+ private void assertTerminated(EventLoopGroup boss, EventLoopGroup worker) {
+ assertThat(boss.terminationFuture().awaitUninterruptibly(10,
TimeUnit.SECONDS)).isTrue();
+ assertThat(worker.terminationFuture().awaitUninterruptibly(10,
TimeUnit.SECONDS)).isTrue();
+ }
+
+ private Object field(GrpcServer server, String name) throws Exception {
+ Field field = GrpcServer.class.getDeclaredField(name);
+ field.setAccessible(true);
+ return field.get(server);
+ }
+}
diff --git
a/test/src/test/java/org/apache/rocketmq/test/client/consumer/pop/PopPriorityIT.java
b/test/src/test/java/org/apache/rocketmq/test/client/consumer/pop/PopPriorityIT.java
index 9a9e67feac..c3e0f90a03 100644
---
a/test/src/test/java/org/apache/rocketmq/test/client/consumer/pop/PopPriorityIT.java
+++
b/test/src/test/java/org/apache/rocketmq/test/client/consumer/pop/PopPriorityIT.java
@@ -20,6 +20,8 @@ package org.apache.rocketmq.test.client.consumer.pop;
import org.apache.rocketmq.client.consumer.PopResult;
import org.apache.rocketmq.client.consumer.PopStatus;
import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.KeyBuilder;
+import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.attribute.AttributeParser;
import org.apache.rocketmq.common.attribute.CQType;
import org.apache.rocketmq.common.attribute.TopicMessageType;
@@ -38,15 +40,16 @@ import org.junit.runners.Parameterized;
import java.time.Duration;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
-import java.util.Random;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import static
org.apache.rocketmq.common.SubscriptionGroupAttributes.PRIORITY_FACTOR_ATTRIBUTE;
import static org.awaitility.Awaitility.await;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
@RunWith(Parameterized.class)
@@ -61,7 +64,7 @@ public class PopPriorityIT extends BasePopNormally {
this.priorityOrderAsc = priorityOrderAsc;
}
- @Parameterized.Parameters
+ @Parameterized.Parameters(name = "kv={0}, ascending={1}")
public static List<Object[]> params() {
List<Object[]> result = new ArrayList<>();
result.add(new Object[] {false, true});
@@ -120,10 +123,9 @@ public class PopPriorityIT extends BasePopNormally {
producer.send(message);
}
}
- Assert.assertTrue(awaitDispatchMs(2000));
+ awaitStoredMessages(topic, writeQueueNum * msgNumPerQueue);
for (int i = 0; i < msgNumPerQueue; i++) {
- PopResult popResult =
popMessageAsync(Duration.ofSeconds(600).toMillis(), 1, 30000).get();
- TestUtil.waitForMonment(20); // wait lock release
+ PopResult popResult =
popMessages(Duration.ofSeconds(600).toMillis(), 1, 30000);
assertEquals(PopStatus.FOUND, popResult.getPopStatus());
MessageExt message = popResult.getMsgFoundList().get(0);
assertEquals(maxPriority, message.getPriority()); // not a
coincidence
@@ -136,9 +138,9 @@ public class PopPriorityIT extends BasePopNormally {
Message message = mockMessage(topic, i, String.valueOf(i));
producer.send(message);
}
- Assert.assertTrue(awaitDispatchMs(2000));
+ awaitStoredMessages(topic, writeQueueNum);
for (int i = 0; i < writeQueueNum; i++) {
- PopResult popResult =
popMessageAsync(Duration.ofSeconds(30).toMillis(), 1, 30000).get();
+ PopResult popResult =
popMessages(Duration.ofSeconds(600).toMillis(), 1, 30000);
assertEquals(PopStatus.FOUND, popResult.getPopStatus());
MessageExt message = popResult.getMsgFoundList().get(0);
int expectPriority = priorityOrderAsc ? writeQueueNum - 1 - i : i;
@@ -162,12 +164,11 @@ public class PopPriorityIT extends BasePopNormally {
producer.send(message);
}
}
- Assert.assertTrue(awaitDispatchMs(2000));
+ awaitStoredMessages(topic, writeQueueNum * msgNumPerQueue);
int sampleCount = 800;
int[] queueIdCount = new int[writeQueueNum];
for (int i = 0; i < sampleCount; i++) {
- PopResult popResult =
popMessageAsync(Duration.ofSeconds(600).toMillis(), 1, 30000).get();
- TestUtil.waitForMonment(10); // wait lock release
+ PopResult popResult =
popMessages(Duration.ofSeconds(600).toMillis(), 1, 30000);
assertEquals(PopStatus.FOUND, popResult.getPopStatus());
MessageExt message = popResult.getMsgFoundList().get(0);
queueIdCount[message.getQueueId()] =
queueIdCount[message.getQueueId()] + 1;
@@ -175,7 +176,8 @@ public class PopPriorityIT extends BasePopNormally {
double expectAverage = (double) sampleCount / writeQueueNum;
for (int count : queueIdCount) {
- assertTrue(Math.abs(count - expectAverage) < expectAverage * 0.4);
+ assertTrue("Unexpected queue distribution: " +
Arrays.toString(queueIdCount),
+ Math.abs(count - expectAverage) < expectAverage * 0.4);
}
}
@@ -184,22 +186,22 @@ public class PopPriorityIT extends BasePopNormally {
// retry as lowest by default
int count = 100;
for (int i = 0; i < count; i++) {
- Message message = mockMessage(topic, new
Random().nextInt(writeQueueNum), String.valueOf(i));
+ Message message = mockMessage(topic, i % writeQueueNum,
String.valueOf(i));
producer.send(message);
}
+ awaitStoredMessages(topic, count);
int invisibleTime = 3;
- PopResult popResult =
popMessageAsync(Duration.ofSeconds(invisibleTime).toMillis(), 1, 30000).get();
+ PopResult popResult =
popMessages(Duration.ofSeconds(invisibleTime).toMillis(), 1, 30000);
assertEquals(PopStatus.FOUND, popResult.getPopStatus());
String retryId = popResult.getMsgFoundList().get(0).getMsgId();
- TestUtil.waitForSeconds(invisibleTime + 3);
- Assert.assertTrue(awaitDispatchMs(2000));
+ awaitRetryMessages(1);
List<MessageExt> collect = new ArrayList<>();
await()
.pollInterval(1, TimeUnit.SECONDS)
.atMost(35, TimeUnit.SECONDS)
.until(() -> {
- PopResult result =
popMessageAsync(Duration.ofSeconds(600).toMillis(), 32, 5000).get();
+ PopResult result =
popMessages(Duration.ofSeconds(600).toMillis(), 32, 5000);
if (PopStatus.FOUND.equals(result.getPopStatus())) {
collect.addAll(result.getMsgFoundList());
}
@@ -215,22 +217,22 @@ public class PopPriorityIT extends BasePopNormally {
brokerController1.getBrokerConfig().setPopFromRetryProbabilityForPriority(100);
int count = 100;
for (int i = 0; i < count; i++) {
- Message message = mockMessage(topic, new
Random().nextInt(writeQueueNum), String.valueOf(i));
+ Message message = mockMessage(topic, i % writeQueueNum,
String.valueOf(i));
producer.send(message);
}
+ awaitStoredMessages(topic, count);
int invisibleTime = 3;
- PopResult popResult =
popMessageAsync(Duration.ofSeconds(invisibleTime).toMillis(), 1, 30000).get();
+ PopResult popResult =
popMessages(Duration.ofSeconds(invisibleTime).toMillis(), 1, 30000);
assertEquals(PopStatus.FOUND, popResult.getPopStatus());
String retryId = popResult.getMsgFoundList().get(0).getMsgId();
- TestUtil.waitForSeconds(invisibleTime + 3);
- Assert.assertTrue(awaitDispatchMs(2000));
+ awaitRetryMessages(1);
List<MessageExt> collect = new ArrayList<>();
await()
.pollInterval(1, TimeUnit.SECONDS)
.atMost(35, TimeUnit.SECONDS)
.until(() -> {
- PopResult result =
popMessageAsync(Duration.ofSeconds(600).toMillis(), 32, 5000).get();
+ PopResult result =
popMessages(Duration.ofSeconds(600).toMillis(), 32, 5000);
if (PopStatus.FOUND.equals(result.getPopStatus())) {
collect.addAll(result.getMsgFoundList());
}
@@ -249,19 +251,19 @@ public class PopPriorityIT extends BasePopNormally {
Message message = mockMessage(topic, i, String.valueOf(i));
producer.send(message);
}
- Assert.assertTrue(awaitDispatchMs(2000));
+ awaitStoredMessages(topic, writeQueueNum);
int invisibleTime = 3;
- PopResult popResult =
popMessageAsync(Duration.ofSeconds(invisibleTime).toMillis(), writeQueueNum,
30000).get();
+ PopResult popResult =
popMessages(Duration.ofSeconds(invisibleTime).toMillis(), writeQueueNum, 30000);
assertEquals(PopStatus.FOUND, popResult.getPopStatus());
assertEquals(writeQueueNum, popResult.getMsgFoundList().size());
- TestUtil.waitForSeconds(invisibleTime + 3);
+ awaitRetryMessages(writeQueueNum);
List<MessageExt> collect = new ArrayList<>();
await()
.pollInterval(1, TimeUnit.SECONDS)
.atMost(35, TimeUnit.SECONDS)
.until(() -> {
- PopResult result =
popMessageAsync(Duration.ofSeconds(600).toMillis(), 32, 5000).get();
+ PopResult result =
popMessages(Duration.ofSeconds(600).toMillis(), 32, 5000);
if (PopStatus.FOUND.equals(result.getPopStatus())) {
collect.addAll(result.getMsgFoundList());
}
@@ -315,6 +317,53 @@ public class PopPriorityIT extends BasePopNormally {
assertEquals(0, msgList.get(msgList.size() - 1).getQueueOffset()); //
means a separate retry queue
}
+ private void awaitStoredMessages(String storedTopic, int expectedCount) {
+ await().alias("consume queues for " + storedTopic)
+ .pollInterval(10, TimeUnit.MILLISECONDS)
+ .atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(() -> {
+ TopicConfig topicConfig =
brokerController1.getTopicConfigManager().selectTopicConfig(storedTopic);
+ assertNotNull("Topic has not been created: " + storedTopic,
topicConfig);
+ long storedCount = 0;
+ for (int queueId = 0; queueId <
topicConfig.getReadQueueNums(); queueId++) {
+ storedCount +=
brokerController1.getMessageStore().getMaxOffsetInQueue(storedTopic, queueId);
+ }
+ assertEquals("Messages dispatched to " + storedTopic,
expectedCount, storedCount);
+ });
+ }
+
+ private void awaitRetryMessages(int expectedCount) {
+ // Expiration alone does not mean the retry has been revived and
dispatched to its consume queue.
+ String retryTopic = KeyBuilder.buildPopRetryTopic(topic, group,
+ brokerController1.getBrokerConfig().isEnableRetryTopicV2());
+ awaitStoredMessages(retryTopic, expectedCount);
+ }
+
+ private PopResult popMessages(long invisibleTime, int maxNums, long
timeout) throws Exception {
+ // A response can reach the client before the previous request's
queue-lock completion callback runs.
+ await().alias("previous POP locks released")
+ .pollDelay(0, TimeUnit.MILLISECONDS)
+ .pollInterval(10, TimeUnit.MILLISECONDS)
+ .atMost(10, TimeUnit.SECONDS)
+ .until(() -> {
+ if (popConsumerKVServiceEnable) {
+ if
(!brokerController1.getPopConsumerService().getConsumerLockService().tryLock(group,
topic)) {
+ return false;
+ }
+
brokerController1.getPopConsumerService().getConsumerLockService().unlock(group,
topic);
+ } else {
+ for (int queueId = 0; queueId < writeQueueNum; queueId++) {
+ if
(!brokerController1.getPopMessageProcessor().getQueueLockManager().tryLock(topic,
group, queueId)) {
+ return false;
+ }
+
brokerController1.getPopMessageProcessor().getQueueLockManager().unLock(topic,
group, queueId);
+ }
+ }
+ return true;
+ });
+ return popMessageAsync(invisibleTime, maxNums, timeout).get();
+ }
+
private static Message mockMessage(String topic, int priority, String key)
{
Message msg = new Message(topic, "HW".getBytes());
if (priority >= 0) {
