This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
The following commit(s) were added to refs/heads/master by this push:
new 315f98eb [Java] Record consumegroup when consumeMessage (#1290)
315f98eb is described below
commit 315f98eb894ba95bdad19ea312bd9088fe2e2f15
Author: ymwneu <[email protected]>
AuthorDate: Mon Jul 6 10:23:37 2026 +0800
[Java] Record consumegroup when consumeMessage (#1290)
---
.../client/java/impl/consumer/ConsumeService.java | 8 +++--
.../client/java/impl/consumer/ConsumeTask.java | 9 +++--
.../java/impl/consumer/FifoConsumeService.java | 4 +--
.../java/impl/consumer/LiteFifoConsumeService.java | 4 +--
.../java/impl/consumer/LitePushConsumerImpl.java | 7 ++--
.../impl/consumer/LiteStandardConsumeService.java | 4 +--
.../java/impl/consumer/PushConsumerImpl.java | 7 ++--
.../java/impl/consumer/StandardConsumeService.java | 4 +--
.../java/impl/consumer/ConsumeServiceTest.java | 16 ++++-----
.../client/java/impl/consumer/ConsumeTaskTest.java | 39 ++++++++++++++++++++--
10 files changed, 74 insertions(+), 28 deletions(-)
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeService.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeService.java
index 40adb9fb..11075c1e 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeService.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeService.java
@@ -41,14 +41,17 @@ public abstract class ConsumeService {
private static final Logger log =
LoggerFactory.getLogger(ConsumeService.class);
protected final ClientId clientId;
+ protected final String consumerGroup;
private final MessageListener messageListener;
private final ThreadPoolExecutor consumptionExecutor;
private final MessageInterceptor messageInterceptor;
private final ScheduledExecutorService scheduler;
- public ConsumeService(ClientId clientId, MessageListener messageListener,
ThreadPoolExecutor consumptionExecutor,
+ public ConsumeService(ClientId clientId, String consumerGroup,
+ MessageListener messageListener, ThreadPoolExecutor
consumptionExecutor,
MessageInterceptor messageInterceptor, ScheduledExecutorService
scheduler) {
this.clientId = clientId;
+ this.consumerGroup = consumerGroup;
this.messageListener = messageListener;
this.consumptionExecutor = consumptionExecutor;
this.messageInterceptor = messageInterceptor;
@@ -63,7 +66,8 @@ public abstract class ConsumeService {
public ListenableFuture<ConsumeResult> consume(MessageViewImpl
messageView, Duration delay) {
final ListeningExecutorService executorService =
MoreExecutors.listeningDecorator(consumptionExecutor);
- final ConsumeTask task = new ConsumeTask(clientId, messageListener,
messageView, messageInterceptor);
+ final ConsumeTask task = new ConsumeTask(clientId, consumerGroup,
+ messageListener, messageView, messageInterceptor);
// Consume message with no delay.
if (Duration.ZERO.compareTo(delay) >= 0) {
return executorService.submit(task);
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeTask.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeTask.java
index 19cea3e0..83a77ee4 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeTask.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeTask.java
@@ -40,17 +40,20 @@ public class ConsumeTask implements Callable<ConsumeResult>
{
static final AttributeKey<MessageViewImpl> MESSAGE_VIEW_CONTEXT_KEY =
AttributeKey.create("messageView");
static final AttributeKey<String> REMOTE_ADDR_CONTEXT_KEY =
AttributeKey.create("remoteAddr");
static final AttributeKey<Throwable> CONSUME_ERROR_CONTEXT_KEY =
AttributeKey.create("consumeError");
+ static final AttributeKey<String> CONSUMER_GROUP_CONTEXT_KEY =
AttributeKey.create("consumerGroup");
private static final Logger log =
LoggerFactory.getLogger(ConsumeTask.class);
private final ClientId clientId;
+ private final String consumerGroup;
private final MessageListener messageListener;
private final MessageViewImpl messageView;
private final MessageInterceptor messageInterceptor;
- public ConsumeTask(ClientId clientId, MessageListener messageListener,
MessageViewImpl messageView,
- MessageInterceptor messageInterceptor) {
+ public ConsumeTask(ClientId clientId, String consumerGroup,
MessageListener messageListener,
+ MessageViewImpl messageView, MessageInterceptor
messageInterceptor) {
this.clientId = clientId;
+ this.consumerGroup = consumerGroup;
this.messageListener = messageListener;
this.messageView = messageView;
this.messageInterceptor = messageInterceptor;
@@ -72,6 +75,8 @@ public class ConsumeTask implements Callable<ConsumeResult> {
context.putAttribute(REMOTE_ADDR_CONTEXT_KEY,
Attribute.create(remoteAddr));
// Add message view to context.
context.putAttribute(MESSAGE_VIEW_CONTEXT_KEY,
Attribute.create(messageView));
+ // Add consumer group to context.
+ context.putAttribute(CONSUMER_GROUP_CONTEXT_KEY,
Attribute.create(consumerGroup));
messageInterceptor.doBefore(context, generalMessages);
Throwable throwable = null;
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/FifoConsumeService.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/FifoConsumeService.java
index e122dc43..81bea5a6 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/FifoConsumeService.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/FifoConsumeService.java
@@ -39,10 +39,10 @@ class FifoConsumeService extends ConsumeService {
private static final Logger log =
LoggerFactory.getLogger(FifoConsumeService.class);
private final boolean enableFifoConsumeAccelerator;
- public FifoConsumeService(ClientId clientId, MessageListener
messageListener,
+ public FifoConsumeService(ClientId clientId, String consumerGroup,
MessageListener messageListener,
ThreadPoolExecutor consumptionExecutor, MessageInterceptor
messageInterceptor,
ScheduledExecutorService scheduler, boolean
enableFifoConsumeAccelerator) {
- super(clientId, messageListener, consumptionExecutor,
messageInterceptor, scheduler);
+ super(clientId, consumerGroup, messageListener, consumptionExecutor,
messageInterceptor, scheduler);
this.enableFifoConsumeAccelerator = enableFifoConsumeAccelerator;
}
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/LiteFifoConsumeService.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/LiteFifoConsumeService.java
index 74076aeb..5e79850a 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/LiteFifoConsumeService.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/LiteFifoConsumeService.java
@@ -35,10 +35,10 @@ import org.apache.rocketmq.client.java.misc.ClientId;
public class LiteFifoConsumeService extends FifoConsumeService {
- public LiteFifoConsumeService(ClientId clientId, MessageListener
messageListener,
+ public LiteFifoConsumeService(ClientId clientId, String consumerGroup,
MessageListener messageListener,
ThreadPoolExecutor consumptionExecutor, MessageInterceptor
messageInterceptor,
ScheduledExecutorService scheduler, boolean
enableFifoConsumeAccelerator) {
- super(clientId, messageListener, consumptionExecutor,
+ super(clientId, consumerGroup, messageListener, consumptionExecutor,
messageInterceptor, scheduler, enableFifoConsumeAccelerator);
}
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/LitePushConsumerImpl.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/LitePushConsumerImpl.java
index 24fc45d9..e206faa6 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/LitePushConsumerImpl.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/LitePushConsumerImpl.java
@@ -93,11 +93,12 @@ public class LitePushConsumerImpl extends PushConsumerImpl
implements LitePushCo
if (getSettings().isFifo()) {
log.info("Create Lite FIFO consume service, consumerGroup={},
clientId={}, enableFifoConsumeAccelerator={}",
getConsumerGroup(), clientId, enableFifoConsumeAccelerator);
- return new LiteFifoConsumeService(clientId, messageListener,
consumptionExecutor, this,
- scheduler, enableFifoConsumeAccelerator);
+ return new LiteFifoConsumeService(clientId, getConsumerGroup(),
messageListener,
+ consumptionExecutor, this, scheduler,
enableFifoConsumeAccelerator);
}
log.info("Create Lite standard consume service, consumerGroup={},
clientId={}", getConsumerGroup(), clientId);
- return new LiteStandardConsumeService(clientId, messageListener,
consumptionExecutor, this, scheduler);
+ return new LiteStandardConsumeService(clientId, getConsumerGroup(),
+ messageListener, consumptionExecutor, this, scheduler);
}
}
\ No newline at end of file
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/LiteStandardConsumeService.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/LiteStandardConsumeService.java
index efbe1ab6..6044aac3 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/LiteStandardConsumeService.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/LiteStandardConsumeService.java
@@ -35,10 +35,10 @@ import org.slf4j.LoggerFactory;
public class LiteStandardConsumeService extends ConsumeService {
private static final Logger log =
LoggerFactory.getLogger(LiteStandardConsumeService.class);
- public LiteStandardConsumeService(ClientId clientId, MessageListener
messageListener,
+ public LiteStandardConsumeService(ClientId clientId, String consumerGroup,
MessageListener messageListener,
ThreadPoolExecutor consumptionExecutor, MessageInterceptor
messageInterceptor,
ScheduledExecutorService scheduler) {
- super(clientId, messageListener, consumptionExecutor,
messageInterceptor, scheduler);
+ super(clientId, consumerGroup, messageListener, consumptionExecutor,
messageInterceptor, scheduler);
}
@Override
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PushConsumerImpl.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PushConsumerImpl.java
index 002a14f4..52f7fe67 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PushConsumerImpl.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PushConsumerImpl.java
@@ -250,11 +250,12 @@ class PushConsumerImpl extends ConsumerImpl implements
PushConsumer {
if (getSettings().isFifo()) {
log.info("Create FIFO consume service, consumerGroup={},
clientId={}, enableFifoConsumeAccelerator={}",
getConsumerGroup(), clientId, enableFifoConsumeAccelerator);
- return new FifoConsumeService(clientId, messageListener,
consumptionExecutor, this,
- scheduler, enableFifoConsumeAccelerator);
+ return new FifoConsumeService(clientId, getConsumerGroup(),
messageListener,
+ consumptionExecutor, this, scheduler,
enableFifoConsumeAccelerator);
}
log.info("Create standard consume service, consumerGroup={},
clientId={}", getConsumerGroup(), clientId);
- return new StandardConsumeService(clientId, messageListener,
consumptionExecutor, this, scheduler);
+ return new StandardConsumeService(clientId, getConsumerGroup(),
+ messageListener, consumptionExecutor, this, scheduler);
}
/**
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/StandardConsumeService.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/StandardConsumeService.java
index 2775077a..828fb8e0 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/StandardConsumeService.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/StandardConsumeService.java
@@ -36,10 +36,10 @@ import org.slf4j.LoggerFactory;
public class StandardConsumeService extends ConsumeService {
private static final Logger log =
LoggerFactory.getLogger(StandardConsumeService.class);
- public StandardConsumeService(ClientId clientId, MessageListener
messageListener,
+ public StandardConsumeService(ClientId clientId, String consumerGroup,
MessageListener messageListener,
ThreadPoolExecutor consumptionExecutor, MessageInterceptor
messageInterceptor,
ScheduledExecutorService scheduler) {
- super(clientId, messageListener, consumptionExecutor,
messageInterceptor, scheduler);
+ super(clientId, consumerGroup, messageListener, consumptionExecutor,
messageInterceptor, scheduler);
}
@Override
diff --git
a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeServiceTest.java
b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeServiceTest.java
index dbf8c89b..687ff759 100644
---
a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeServiceTest.java
+++
b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeServiceTest.java
@@ -51,8 +51,8 @@ public class ConsumeServiceTest extends TestBase {
@Test
public void testConsumeSuccess() throws ExecutionException,
InterruptedException, TimeoutException {
final MessageListener messageListener = messageView ->
ConsumeResult.SUCCESS;
- final ConsumeService consumeService = new ConsumeService(clientId,
messageListener,
- consumptionExecutor, interceptor, scheduler) {
+ final ConsumeService consumeService = new ConsumeService(clientId,
"testConsumerGroup",
+ messageListener, consumptionExecutor, interceptor, scheduler) {
@Override
public void consume(ProcessQueue pq, List<MessageViewImpl>
messageViews) {
}
@@ -66,8 +66,8 @@ public class ConsumeServiceTest extends TestBase {
@Test
public void testConsumeFailure() throws ExecutionException,
InterruptedException, TimeoutException {
final MessageListener messageListener = messageView ->
ConsumeResult.FAILURE;
- final ConsumeService consumeService = new ConsumeService(clientId,
messageListener,
- consumptionExecutor, interceptor, scheduler) {
+ final ConsumeService consumeService = new ConsumeService(clientId,
"testConsumerGroup",
+ messageListener, consumptionExecutor, interceptor, scheduler) {
@Override
public void consume(ProcessQueue pq, List<MessageViewImpl>
messageViews) {
}
@@ -83,8 +83,8 @@ public class ConsumeServiceTest extends TestBase {
final MessageListener messageListener = messageView -> {
throw new RuntimeException();
};
- final ConsumeService consumeService = new ConsumeService(clientId,
messageListener,
- consumptionExecutor, interceptor, scheduler) {
+ final ConsumeService consumeService = new ConsumeService(clientId,
"testConsumerGroup",
+ messageListener, consumptionExecutor, interceptor, scheduler) {
@Override
public void consume(ProcessQueue pq, List<MessageViewImpl>
messageViews) {
@@ -99,8 +99,8 @@ public class ConsumeServiceTest extends TestBase {
@Test
public void testConsumeWithDelay() throws ExecutionException,
InterruptedException {
final MessageListener messageListener = messageView ->
ConsumeResult.SUCCESS;
- final ConsumeService consumeService = new ConsumeService(clientId,
messageListener,
- consumptionExecutor, interceptor, scheduler) {
+ final ConsumeService consumeService = new ConsumeService(clientId,
"testConsumerGroup",
+ messageListener, consumptionExecutor, interceptor, scheduler) {
@Override
public void consume(ProcessQueue pq, List<MessageViewImpl>
messageViews) {
diff --git
a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeTaskTest.java
b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeTaskTest.java
index 23312142..9be98a00 100644
---
a/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeTaskTest.java
+++
b/java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeTaskTest.java
@@ -19,9 +19,14 @@ package org.apache.rocketmq.client.java.impl.consumer;
import static org.junit.Assert.assertEquals;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.MessageListener;
+import org.apache.rocketmq.client.java.hook.Attribute;
import org.apache.rocketmq.client.java.hook.MessageInterceptor;
+import org.apache.rocketmq.client.java.hook.MessageInterceptorContext;
+import org.apache.rocketmq.client.java.message.GeneralMessage;
import org.apache.rocketmq.client.java.message.MessageViewImpl;
import org.apache.rocketmq.client.java.misc.ClientId;
import org.apache.rocketmq.client.java.tool.TestBase;
@@ -37,7 +42,8 @@ public class ConsumeTaskTest extends TestBase {
final MessageListener messageListener =
Mockito.mock(MessageListener.class);
Mockito.when(messageListener.consume(messageView)).thenReturn(ConsumeResult.SUCCESS);
final MessageInterceptor messageInterceptor =
Mockito.mock(MessageInterceptor.class);
- final ConsumeTask consumeTask = new ConsumeTask(clientId,
messageListener, messageView, messageInterceptor);
+ final ConsumeTask consumeTask = new ConsumeTask(clientId,
"testConsumerGroup",
+ messageListener, messageView, messageInterceptor);
final ConsumeResult consumeResult = consumeTask.call();
assertEquals(ConsumeResult.SUCCESS, consumeResult);
}
@@ -49,8 +55,37 @@ public class ConsumeTaskTest extends TestBase {
final MessageListener messageListener =
Mockito.mock(MessageListener.class);
Mockito.when(messageListener.consume(messageView)).thenThrow(new
RuntimeException());
final MessageInterceptor messageInterceptor =
Mockito.mock(MessageInterceptor.class);
- final ConsumeTask consumeTask = new ConsumeTask(clientId,
messageListener, messageView, messageInterceptor);
+ final ConsumeTask consumeTask = new ConsumeTask(clientId,
"testConsumerGroup",
+ messageListener, messageView, messageInterceptor);
final ConsumeResult consumeResult = consumeTask.call();
assertEquals(ConsumeResult.FAILURE, consumeResult);
}
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void testConsumerGroupInContext() {
+ final String expectedGroup = "myConsumerGroup";
+ ClientId clientId = new ClientId();
+ final MessageViewImpl messageView = fakeMessageViewImpl();
+ final MessageListener messageListener =
Mockito.mock(MessageListener.class);
+
Mockito.when(messageListener.consume(messageView)).thenReturn(ConsumeResult.SUCCESS);
+ final AtomicReference<String> capturedGroup = new AtomicReference<>();
+ final MessageInterceptor messageInterceptor = new MessageInterceptor()
{
+ @Override
+ public void doBefore(MessageInterceptorContext context,
List<GeneralMessage> messages) {
+ Attribute<String> attr =
context.getAttribute(ConsumeTask.CONSUMER_GROUP_CONTEXT_KEY);
+ if (attr != null) {
+ capturedGroup.set(attr.get());
+ }
+ }
+
+ @Override
+ public void doAfter(MessageInterceptorContext context,
List<GeneralMessage> messages) {
+ }
+ };
+ final ConsumeTask consumeTask = new ConsumeTask(clientId,
expectedGroup, messageListener,
+ messageView, messageInterceptor);
+ consumeTask.call();
+ assertEquals(expectedGroup, capturedGroup.get());
+ }
}
\ No newline at end of file
