From 66856e698736a5bfda19ecd7d197f7b88c8c4fbb Mon Sep 17 00:00:00 2001 From: redlsz Date: Tue, 20 Feb 2024 13:40:17 +0800 Subject: [PATCH 1/2] Fix start and shutdown process of DefaultMessagingProcessor --- .../processor/DefaultMessagingProcessor.java | 6 ++ .../processor/ReceiptHandleProcessor.java | 1 + .../processor/ReceiptHandleProcessorTest.java | 83 +++++++++++++++++++ 3 files changed, 90 insertions(+) create mode 100644 proxy/src/test/java/org/apache/rocketmq/proxy/processor/ReceiptHandleProcessorTest.java diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/DefaultMessagingProcessor.java b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/DefaultMessagingProcessor.java index ba150051bc0..d3343d8db38 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/DefaultMessagingProcessor.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/DefaultMessagingProcessor.java @@ -124,6 +124,12 @@ public static DefaultMessagingProcessor createForClusterMode(RPCHook rpcHook) { protected void init() { this.appendStartAndShutdown(this.serviceManager); + this.appendStartAndShutdown(producerProcessor); + this.appendStartAndShutdown(consumerProcessor); + this.appendStartAndShutdown(transactionProcessor); + this.appendStartAndShutdown(clientProcessor); + this.appendStartAndShutdown(requestBrokerProcessor); + this.appendStartAndShutdown(receiptHandleProcessor); this.appendShutdown(this.producerProcessorExecutor::shutdown); this.appendShutdown(this.consumerProcessorExecutor::shutdown); } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ReceiptHandleProcessor.java b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ReceiptHandleProcessor.java index 5e1be932183..9b010fad598 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ReceiptHandleProcessor.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ReceiptHandleProcessor.java @@ -51,6 +51,7 @@ public ReceiptHandleProcessor(MessagingProcessor messagingProcessor, ServiceMana }); }; this.receiptHandleManager = new DefaultReceiptHandleManager(serviceManager.getMetadataService(), serviceManager.getConsumerManager(), eventListener); + this.appendStartAndShutdown(receiptHandleManager); } protected ProxyContext createContext(String actionName) { diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ReceiptHandleProcessorTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ReceiptHandleProcessorTest.java new file mode 100644 index 00000000000..dc324e13184 --- /dev/null +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ReceiptHandleProcessorTest.java @@ -0,0 +1,83 @@ +/* + * 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.processor; + +import io.netty.channel.local.LocalChannel; +import org.apache.rocketmq.broker.client.ClientChannelInfo; +import org.apache.rocketmq.broker.client.ConsumerIdsChangeListener; +import org.apache.rocketmq.common.consumer.ReceiptHandle; +import org.apache.rocketmq.common.message.MessageClientIDSetter; +import org.apache.rocketmq.proxy.common.ContextVariable; +import org.apache.rocketmq.proxy.common.MessageReceiptHandle; +import org.apache.rocketmq.proxy.common.ProxyContext; +import org.apache.rocketmq.proxy.config.ConfigurationManager; +import org.apache.rocketmq.proxy.config.ProxyConfig; +import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig; +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mockito; + +public class ReceiptHandleProcessorTest extends BaseProcessorTest { + private static final ProxyContext PROXY_CONTEXT = ProxyContext.create(); + private static final String CONSUMER_GROUP = "consumerGroup"; + private static final String TOPIC = "topic"; + private static final String BROKER_NAME = "broker"; + private static final int QUEUE_ID = 1; + private static final String MESSAGE_ID = "messageId"; + private static final long OFFSET = 123L; + private static final long INVISIBLE_TIME = 60000L; + private static final int RECONSUME_TIMES = 1; + private static final String MSG_ID = MessageClientIDSetter.createUniqID(); + private MessageReceiptHandle messageReceiptHandle; + + private ReceiptHandleProcessor receiptHandleProcessor; + + @Before + public void before() throws Throwable { + super.before(); + this.receiptHandleProcessor = new ReceiptHandleProcessor(this.messagingProcessor, this.serviceManager); + ProxyConfig config = ConfigurationManager.getProxyConfig(); + String receiptHandle = ReceiptHandle.builder() + .startOffset(0L) + .retrieveTime(System.currentTimeMillis() - INVISIBLE_TIME + config.getRenewAheadTimeMillis() - 5) + .invisibleTime(INVISIBLE_TIME) + .reviveQueueId(1) + .topicType(ReceiptHandle.NORMAL_TOPIC) + .brokerName(BROKER_NAME) + .queueId(QUEUE_ID) + .offset(OFFSET) + .commitLogOffset(0L) + .build().encode(); + PROXY_CONTEXT.withVal(ContextVariable.CLIENT_ID, "channel-id"); + PROXY_CONTEXT.withVal(ContextVariable.CHANNEL, new LocalChannel()); + Mockito.doNothing().when(consumerManager).appendConsumerIdsChangeListener(Mockito.any(ConsumerIdsChangeListener.class)); + messageReceiptHandle = new MessageReceiptHandle(CONSUMER_GROUP, TOPIC, QUEUE_ID, receiptHandle, MESSAGE_ID, OFFSET, + RECONSUME_TIMES); + } + + @Test + public void testStartAutoRenewTask() throws Exception { + receiptHandleProcessor.start(); + receiptHandleProcessor.addReceiptHandle(PROXY_CONTEXT, PROXY_CONTEXT.getChannel(), CONSUMER_GROUP, MSG_ID, messageReceiptHandle); + Mockito.when(metadataService.getSubscriptionGroupConfig(Mockito.any(), Mockito.eq(CONSUMER_GROUP))).thenReturn(new SubscriptionGroupConfig()); + Mockito.when(consumerManager.findChannel(Mockito.eq(CONSUMER_GROUP), Mockito.eq(PROXY_CONTEXT.getChannel()))).thenReturn(Mockito.mock(ClientChannelInfo.class)); + Mockito.verify(messagingProcessor, Mockito.timeout(10000).times(1)) + .changeInvisibleTime(Mockito.any(ProxyContext.class), Mockito.any(ReceiptHandle.class), Mockito.eq(MESSAGE_ID), + Mockito.eq(CONSUMER_GROUP), Mockito.eq(TOPIC), Mockito.eq(ConfigurationManager.getProxyConfig().getDefaultInvisibleTimeMills())); + } + +} From c8638f60f7735f43304df86d77ee80d2b71de7bc Mon Sep 17 00:00:00 2001 From: redlsz Date: Mon, 26 Feb 2024 11:20:18 +0800 Subject: [PATCH 2/2] minimal changes --- .../proxy/processor/DefaultMessagingProcessor.java | 7 +------ .../proxy/processor/ReceiptHandleProcessorTest.java | 2 +- 2 files changed, 2 insertions(+), 7 deletions(-) diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/DefaultMessagingProcessor.java b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/DefaultMessagingProcessor.java index d3343d8db38..fc45afd2657 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/DefaultMessagingProcessor.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/DefaultMessagingProcessor.java @@ -124,12 +124,7 @@ public static DefaultMessagingProcessor createForClusterMode(RPCHook rpcHook) { protected void init() { this.appendStartAndShutdown(this.serviceManager); - this.appendStartAndShutdown(producerProcessor); - this.appendStartAndShutdown(consumerProcessor); - this.appendStartAndShutdown(transactionProcessor); - this.appendStartAndShutdown(clientProcessor); - this.appendStartAndShutdown(requestBrokerProcessor); - this.appendStartAndShutdown(receiptHandleProcessor); + this.appendStartAndShutdown(this.receiptHandleProcessor); this.appendShutdown(this.producerProcessorExecutor::shutdown); this.appendShutdown(this.consumerProcessorExecutor::shutdown); } diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ReceiptHandleProcessorTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ReceiptHandleProcessorTest.java index dc324e13184..ceb0836557b 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ReceiptHandleProcessorTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ReceiptHandleProcessorTest.java @@ -70,7 +70,7 @@ public void before() throws Throwable { } @Test - public void testStartAutoRenewTask() throws Exception { + public void testStart() throws Exception { receiptHandleProcessor.start(); receiptHandleProcessor.addReceiptHandle(PROXY_CONTEXT, PROXY_CONTEXT.getChannel(), CONSUMER_GROUP, MSG_ID, messageReceiptHandle); Mockito.when(metadataService.getSubscriptionGroupConfig(Mockito.any(), Mockito.eq(CONSUMER_GROUP))).thenReturn(new SubscriptionGroupConfig());