diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/common/ContextVariable.java b/proxy/src/main/java/org/apache/rocketmq/proxy/common/ContextVariable.java index 27cc04cc2a..df102b9648 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/common/ContextVariable.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/common/ContextVariable.java @@ -22,4 +22,5 @@ public class ContextVariable { public final static String LOCAL_ADDRESS = "local-address"; public static final String CLIENT_ID = "client-id"; public static final String LANGUAGE = "language"; + public final static String CHANNEL_KEY = "channel-key"; } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/common/MessageReceiptHandle.java b/proxy/src/main/java/org/apache/rocketmq/proxy/common/MessageReceiptHandle.java new file mode 100644 index 0000000000..4c396e71d0 --- /dev/null +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/common/MessageReceiptHandle.java @@ -0,0 +1,86 @@ +/* + * 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.common; + +import org.apache.rocketmq.common.message.MessageQueue; + +public class MessageReceiptHandle { + private final String group; + private final MessageQueue messageQueue; + private final String messageId; + private final long queueOffset; + private final String originalReceiptHandle; + private final long timestamp; + private final int reconsumeTimes; + private final long expectInvisibleTime; + + private String receiptHandle; + + public MessageReceiptHandle(String group, MessageQueue messageQueue, String receiptHandle, String messageId, + long queueOffset, int reconsumeTimes, long expectInvisibleTime) { + this.group = group; + this.messageQueue = messageQueue; + this.receiptHandle = receiptHandle; + this.originalReceiptHandle = receiptHandle; + this.messageId = messageId; + this.queueOffset = queueOffset; + this.reconsumeTimes = reconsumeTimes; + this.expectInvisibleTime = expectInvisibleTime; + this.timestamp = System.currentTimeMillis(); + } + + public String getGroup() { + return group; + } + + public MessageQueue getMessageQueue() { + return messageQueue; + } + + public String getReceiptHandle() { + return receiptHandle; + } + + public String getOriginalReceiptHandle() { + return originalReceiptHandle; + } + + public String getMessageId() { + return messageId; + } + + public long getQueueOffset() { + return queueOffset; + } + + public int getReconsumeTimes() { + return reconsumeTimes; + } + + public long getTimestamp() { + return timestamp; + } + + public long getExpectInvisibleTime() { + return expectInvisibleTime; + } + + public void update(String receiptHandle) { + this.receiptHandle = receiptHandle; + } +} diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/common/ReceiptHandleGroup.java b/proxy/src/main/java/org/apache/rocketmq/proxy/common/ReceiptHandleGroup.java new file mode 100644 index 0000000000..71e23932b3 --- /dev/null +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/common/ReceiptHandleGroup.java @@ -0,0 +1,38 @@ +/* + * 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.common; + +import com.google.common.collect.ImmutableMap; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +public class ReceiptHandleGroup { + private final Map receiptHandleMap = new ConcurrentHashMap<>(); + + public void put(String key, MessageReceiptHandle value) { + receiptHandleMap.put(key, value); + } + + public void remove(String key) { + receiptHandleMap.remove(key); + } + + public Map all() { + return ImmutableMap.copyOf(receiptHandleMap); + } +} diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/common/utils/ChannelUtils.java b/proxy/src/main/java/org/apache/rocketmq/proxy/common/utils/ChannelUtils.java new file mode 100644 index 0000000000..632506ee77 --- /dev/null +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/common/utils/ChannelUtils.java @@ -0,0 +1,26 @@ +/* + * 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.common.utils; + +import io.netty.channel.Channel; + +public class ChannelUtils { + public static String buildChannelKey(Channel channel, String groupName) { + return channel.id().asLongText() + "%" + groupName; + } +} diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java b/proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java index 49e7b4ae80..7b66e037c2 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java @@ -18,6 +18,7 @@ package org.apache.rocketmq.proxy.config; import java.time.Duration; +import java.util.concurrent.TimeUnit; import org.apache.rocketmq.proxy.ProxyMode; public class ProxyConfig { @@ -97,6 +98,12 @@ public class ProxyConfig { private long longPollingReserveTimeInMillis = 100; + private long invisibleTimeMillisWhenClear = 1000L; + private boolean enableProxyAutoRenew = true; + private long renewAheadTimeMillis = TimeUnit.SECONDS.toMillis(10); + private long renewSliceTimeMillis = TimeUnit.SECONDS.toMillis(60); + private long renewSchedulePeriodMillis = TimeUnit.SECONDS.toMillis(5); + private boolean enableACL = false; private boolean enableTopicMessageTypeCheck = true; @@ -532,4 +539,44 @@ public class ProxyConfig { public void setEnableTopicMessageTypeCheck(boolean enableTopicMessageTypeCheck) { this.enableTopicMessageTypeCheck = enableTopicMessageTypeCheck; } + + public long getInvisibleTimeMillisWhenClear() { + return invisibleTimeMillisWhenClear; + } + + public void setInvisibleTimeMillisWhenClear(long invisibleTimeMillisWhenClear) { + this.invisibleTimeMillisWhenClear = invisibleTimeMillisWhenClear; + } + + public boolean isEnableProxyAutoRenew() { + return enableProxyAutoRenew; + } + + public void setEnableProxyAutoRenew(boolean enableProxyAutoRenew) { + this.enableProxyAutoRenew = enableProxyAutoRenew; + } + + public long getRenewAheadTimeMillis() { + return renewAheadTimeMillis; + } + + public void setRenewAheadTimeMillis(long renewAheadTimeMillis) { + this.renewAheadTimeMillis = renewAheadTimeMillis; + } + + public long getRenewSliceTimeMillis() { + return renewSliceTimeMillis; + } + + public void setRenewSliceTimeMillis(long renewSliceTimeMillis) { + this.renewSliceTimeMillis = renewSliceTimeMillis; + } + + public long getRenewSchedulePeriodMillis() { + return renewSchedulePeriodMillis; + } + + public void setRenewSchedulePeriodMillis(long renewSchedulePeriodMillis) { + this.renewSchedulePeriodMillis = renewSchedulePeriodMillis; + } } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/AbstractMessingActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/AbstractMessingActivity.java index 7ca356df12..6e94e68723 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/AbstractMessingActivity.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/AbstractMessingActivity.java @@ -17,9 +17,12 @@ package org.apache.rocketmq.proxy.grpc.v2; import io.grpc.Context; +import io.netty.channel.Channel; import org.apache.rocketmq.proxy.common.ContextVariable; import org.apache.rocketmq.proxy.common.ProxyContext; +import org.apache.rocketmq.proxy.common.utils.ChannelUtils; import org.apache.rocketmq.proxy.grpc.interceptor.InterceptorConstants; +import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager; import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager; import org.apache.rocketmq.proxy.processor.MessagingProcessor; import org.apache.rocketmq.remoting.protocol.LanguageCode; @@ -28,16 +31,26 @@ public abstract class AbstractMessingActivity { protected final MessagingProcessor messagingProcessor; protected final GrpcClientSettingsManager grpcClientSettingsManager; + protected final GrpcChannelManager grpcChannelManager; public AbstractMessingActivity(MessagingProcessor messagingProcessor, - GrpcClientSettingsManager grpcClientSettingsManager) { + GrpcClientSettingsManager grpcClientSettingsManager, GrpcChannelManager grpcChannelManager) { this.messagingProcessor = messagingProcessor; this.grpcClientSettingsManager = grpcClientSettingsManager; + this.grpcChannelManager = grpcChannelManager; } protected ProxyContext createContext(Context ctx) { return ProxyContext.create() + .withVal(ContextVariable.LOCAL_ADDRESS, InterceptorConstants.METADATA.get(ctx).get(InterceptorConstants.LOCAL_ADDRESS)) + .withVal(ContextVariable.REMOTE_ADDRESS, InterceptorConstants.METADATA.get(ctx).get(InterceptorConstants.REMOTE_ADDRESS)) .withVal(ContextVariable.CLIENT_ID, InterceptorConstants.METADATA.get(ctx).get(InterceptorConstants.CLIENT_ID)) .withVal(ContextVariable.LANGUAGE, LanguageCode.valueOf(InterceptorConstants.METADATA.get(ctx).get(InterceptorConstants.LANGUAGE))); } + + protected void attachChannelId(Context ctx, ProxyContext context, String groupName) { + String clientId = context.getVal(ContextVariable.CLIENT_ID); + Channel channel = grpcChannelManager.createChannel(ctx, groupName, clientId); + context.withVal(ContextVariable.CHANNEL_KEY, ChannelUtils.buildChannelKey(channel, groupName)); + } } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/DefaultGrpcMessingActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/DefaultGrpcMessingActivity.java index 1422c01e78..9ac1001e6d 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/DefaultGrpcMessingActivity.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/DefaultGrpcMessingActivity.java @@ -44,6 +44,7 @@ import org.apache.rocketmq.common.constant.LoggerName; import org.apache.rocketmq.logging.InternalLogger; import org.apache.rocketmq.logging.InternalLoggerFactory; import org.apache.rocketmq.proxy.common.AbstractStartAndShutdown; +import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager; import org.apache.rocketmq.proxy.grpc.v2.client.ClientActivity; import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager; import org.apache.rocketmq.proxy.grpc.v2.consumer.AckMessageActivity; @@ -54,12 +55,11 @@ import org.apache.rocketmq.proxy.grpc.v2.producer.SendMessageActivity; import org.apache.rocketmq.proxy.grpc.v2.route.RouteActivity; import org.apache.rocketmq.proxy.grpc.v2.transaction.EndTransactionActivity; import org.apache.rocketmq.proxy.processor.MessagingProcessor; +import org.apache.rocketmq.proxy.processor.ReceiptHandleProcessor; public class DefaultGrpcMessingActivity extends AbstractStartAndShutdown implements GrpcMessingActivity { private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME); - private final GrpcClientSettingsManager grpcClientSettingsManager; - private final ReceiveMessageActivity receiveMessageActivity; private final AckMessageActivity ackMessageActivity; private final ChangeInvisibleDurationActivity changeInvisibleDurationActivity; @@ -70,16 +70,18 @@ public class DefaultGrpcMessingActivity extends AbstractStartAndShutdown impleme private final ClientActivity clientActivity; protected DefaultGrpcMessingActivity(MessagingProcessor messagingProcessor) { - this.grpcClientSettingsManager = new GrpcClientSettingsManager(messagingProcessor); + GrpcClientSettingsManager grpcClientSettingsManager = new GrpcClientSettingsManager(messagingProcessor); + GrpcChannelManager grpcChannelManager = new GrpcChannelManager(messagingProcessor.getProxyOutService()); + ReceiptHandleProcessor receiptHandleProcessor = new ReceiptHandleProcessor(messagingProcessor); - this.receiveMessageActivity = new ReceiveMessageActivity(messagingProcessor, this.grpcClientSettingsManager); - this.ackMessageActivity = new AckMessageActivity(messagingProcessor, this.grpcClientSettingsManager); - this.changeInvisibleDurationActivity = new ChangeInvisibleDurationActivity(messagingProcessor, this.grpcClientSettingsManager); - this.sendMessageActivity = new SendMessageActivity(messagingProcessor, this.grpcClientSettingsManager); - this.forwardMessageToDLQActivity = new ForwardMessageToDLQActivity(messagingProcessor, this.grpcClientSettingsManager); - this.endTransactionActivity = new EndTransactionActivity(messagingProcessor, this.grpcClientSettingsManager); - this.routeActivity = new RouteActivity(messagingProcessor, this.grpcClientSettingsManager); - this.clientActivity = new ClientActivity(messagingProcessor, this.grpcClientSettingsManager); + this.receiveMessageActivity = new ReceiveMessageActivity(messagingProcessor, receiptHandleProcessor, grpcClientSettingsManager, grpcChannelManager); + this.ackMessageActivity = new AckMessageActivity(messagingProcessor, receiptHandleProcessor, grpcClientSettingsManager, grpcChannelManager); + this.changeInvisibleDurationActivity = new ChangeInvisibleDurationActivity(messagingProcessor, receiptHandleProcessor, grpcClientSettingsManager, grpcChannelManager); + this.sendMessageActivity = new SendMessageActivity(messagingProcessor, grpcClientSettingsManager, grpcChannelManager); + this.forwardMessageToDLQActivity = new ForwardMessageToDLQActivity(messagingProcessor, grpcClientSettingsManager, grpcChannelManager); + this.endTransactionActivity = new EndTransactionActivity(messagingProcessor, grpcClientSettingsManager, grpcChannelManager); + this.routeActivity = new RouteActivity(messagingProcessor, grpcClientSettingsManager, grpcChannelManager); + this.clientActivity = new ClientActivity(messagingProcessor, grpcClientSettingsManager, grpcChannelManager); } @Override diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java index 86b9652c66..9131da633d 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java @@ -71,22 +71,10 @@ public class ClientActivity extends AbstractMessingActivity { private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME); - private final GrpcChannelManager grpcChannelManager; - public ClientActivity(MessagingProcessor messagingProcessor, - GrpcClientSettingsManager grpcClientSettingsManager) { - super(messagingProcessor, grpcClientSettingsManager); - - this.grpcChannelManager = new GrpcChannelManager(messagingProcessor.getProxyOutService()); - - this.init(); - } - - ClientActivity(MessagingProcessor messagingProcessor, GrpcClientSettingsManager grpcClientSettingsManager, GrpcChannelManager grpcChannelManager) { - super(messagingProcessor, grpcClientSettingsManager); - this.grpcChannelManager = grpcChannelManager; + super(messagingProcessor, grpcClientSettingsManager, grpcChannelManager); this.init(); } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/AckMessageActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/AckMessageActivity.java index 42cc3a8645..e76dc42780 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/AckMessageActivity.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/AckMessageActivity.java @@ -30,22 +30,29 @@ import java.util.concurrent.CompletableFuture; import org.apache.rocketmq.client.consumer.AckResult; import org.apache.rocketmq.client.consumer.AckStatus; import org.apache.rocketmq.common.consumer.ReceiptHandle; +import org.apache.rocketmq.proxy.common.ContextVariable; import org.apache.rocketmq.proxy.common.ProxyContext; import org.apache.rocketmq.proxy.grpc.v2.AbstractMessingActivity; +import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager; import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager; import org.apache.rocketmq.proxy.grpc.v2.common.GrpcConverter; import org.apache.rocketmq.proxy.grpc.v2.common.ResponseBuilder; import org.apache.rocketmq.proxy.processor.MessagingProcessor; +import org.apache.rocketmq.proxy.processor.ReceiptHandleProcessor; public class AckMessageActivity extends AbstractMessingActivity { - - public AckMessageActivity(MessagingProcessor messagingProcessor, - GrpcClientSettingsManager grpcClientSettingsManager) { - super(messagingProcessor, grpcClientSettingsManager); + protected ReceiptHandleProcessor receiptHandleProcessor; + public AckMessageActivity(MessagingProcessor messagingProcessor, ReceiptHandleProcessor receiptHandleProcessor, + GrpcClientSettingsManager grpcClientSettingsManager, + GrpcChannelManager grpcChannelManager) { + super(messagingProcessor, grpcClientSettingsManager, grpcChannelManager); + this.receiptHandleProcessor = receiptHandleProcessor; } public CompletableFuture ackMessage(Context ctx, AckMessageRequest request) { ProxyContext proxyContext = createContext(ctx); + String groupName = GrpcConverter.wrapResourceWithNamespace(request.getGroup()); + attachChannelId(ctx, proxyContext, groupName); CompletableFuture future = new CompletableFuture<>(); try { @@ -101,12 +108,16 @@ public class AckMessageActivity extends AbstractMessingActivity { ackMessageEntry.getMessageId(), GrpcConverter.wrapResourceWithNamespace(request.getGroup()), GrpcConverter.wrapResourceWithNamespace(request.getTopic())); - ackResultFuture - .thenAccept(result -> future.complete(convertToAckMessageResultEntry(ctx, ackMessageEntry, result))) - .exceptionally(throwable -> { - future.complete(failResult.setStatus(ResponseBuilder.buildStatus(throwable)).build()); - return null; - }); + ackResultFuture.thenAccept(result -> { + if (AckStatus.OK.equals(result.getStatus())) { + String channelId = ctx.getVal(ContextVariable.CHANNEL_KEY); + receiptHandleProcessor.removeReceiptHandle(channelId, ackMessageEntry.getReceiptHandle()); + } + future.complete(convertToAckMessageResultEntry(ctx, ackMessageEntry, result)); + }).exceptionally(throwable -> { + future.complete(failResult.setStatus(ResponseBuilder.buildStatus(throwable)).build()); + return null; + }); } catch (Throwable t) { future.complete(failResult.setStatus(ResponseBuilder.buildStatus(t)).build()); } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ChangeInvisibleDurationActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ChangeInvisibleDurationActivity.java index fc8fb405f7..f6f2bbec18 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ChangeInvisibleDurationActivity.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ChangeInvisibleDurationActivity.java @@ -25,23 +25,30 @@ import java.util.concurrent.CompletableFuture; import org.apache.rocketmq.client.consumer.AckResult; import org.apache.rocketmq.client.consumer.AckStatus; import org.apache.rocketmq.common.consumer.ReceiptHandle; +import org.apache.rocketmq.proxy.common.ContextVariable; import org.apache.rocketmq.proxy.common.ProxyContext; import org.apache.rocketmq.proxy.grpc.v2.AbstractMessingActivity; +import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager; import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager; import org.apache.rocketmq.proxy.grpc.v2.common.GrpcConverter; import org.apache.rocketmq.proxy.grpc.v2.common.ResponseBuilder; import org.apache.rocketmq.proxy.processor.MessagingProcessor; +import org.apache.rocketmq.proxy.processor.ReceiptHandleProcessor; public class ChangeInvisibleDurationActivity extends AbstractMessingActivity { + protected ReceiptHandleProcessor receiptHandleProcessor; - public ChangeInvisibleDurationActivity(MessagingProcessor messagingProcessor, - GrpcClientSettingsManager grpcClientSettingsManager) { - super(messagingProcessor, grpcClientSettingsManager); + public ChangeInvisibleDurationActivity(MessagingProcessor messagingProcessor, ReceiptHandleProcessor receiptHandleProcessor, + GrpcClientSettingsManager grpcClientSettingsManager, GrpcChannelManager grpcChannelManager) { + super(messagingProcessor, grpcClientSettingsManager, grpcChannelManager); + this.receiptHandleProcessor = receiptHandleProcessor; } public CompletableFuture changeInvisibleDuration(Context ctx, ChangeInvisibleDurationRequest request) { ProxyContext context = createContext(ctx); + String groupName = GrpcConverter.wrapResourceWithNamespace(request.getGroup()); + attachChannelId(ctx, context, groupName); CompletableFuture future = new CompletableFuture<>(); try { @@ -54,7 +61,13 @@ public class ChangeInvisibleDurationActivity extends AbstractMessingActivity { GrpcConverter.wrapResourceWithNamespace(request.getGroup()), GrpcConverter.wrapResourceWithNamespace(request.getTopic()), Durations.toMillis(request.getInvisibleDuration()) - ).thenApply(ackResult -> convertToChangeInvisibleDurationResponse(context, request, ackResult)); + ).thenApply(ackResult -> { + if (AckStatus.OK.equals(ackResult.getStatus())) { + String channelId = context.getVal(ContextVariable.CHANNEL_KEY); + receiptHandleProcessor.removeReceiptHandle(channelId, receiptHandle.getReceiptHandle()); + } + return convertToChangeInvisibleDurationResponse(context, request, ackResult); + }); } catch (Throwable t) { future.completeExceptionally(t); } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivity.java index 3c8b045a07..42808961ce 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivity.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivity.java @@ -25,33 +25,46 @@ import apache.rocketmq.v2.Subscription; import com.google.protobuf.util.Durations; import io.grpc.Context; import io.grpc.stub.StreamObserver; +import java.util.List; import java.util.concurrent.TimeUnit; import org.apache.commons.lang3.StringUtils; import org.apache.rocketmq.common.constant.ConsumeInitMode; import org.apache.rocketmq.common.filter.FilterAPI; +import org.apache.rocketmq.common.message.MessageConst; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.common.message.MessageQueue; import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData; +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.proxy.grpc.v2.AbstractMessingActivity; +import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager; import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager; import org.apache.rocketmq.proxy.grpc.v2.common.GrpcConverter; import org.apache.rocketmq.proxy.processor.MessagingProcessor; import org.apache.rocketmq.proxy.processor.QueueSelector; +import org.apache.rocketmq.proxy.processor.ReceiptHandleProcessor; import org.apache.rocketmq.proxy.service.route.MessageQueueSelector; import org.apache.rocketmq.proxy.service.route.MessageQueueView; import org.apache.rocketmq.proxy.service.route.SelectableMessageQueue; public class ReceiveMessageActivity extends AbstractMessingActivity { + protected ReceiptHandleProcessor receiptHandleProcessor; - public ReceiveMessageActivity(MessagingProcessor messagingProcessor, - GrpcClientSettingsManager grpcClientSettingsManager) { - super(messagingProcessor, grpcClientSettingsManager); + public ReceiveMessageActivity(MessagingProcessor messagingProcessor, ReceiptHandleProcessor receiptHandleProcessor, + GrpcClientSettingsManager grpcClientSettingsManager, GrpcChannelManager grpcChannelManager) { + super(messagingProcessor, grpcClientSettingsManager, grpcChannelManager); + this.receiptHandleProcessor = receiptHandleProcessor; } public void receiveMessage(Context ctx, ReceiveMessageRequest request, StreamObserver responseObserver) { ProxyContext proxyContext = createContext(ctx); ReceiveMessageResponseStreamWriter writer = createWriter(proxyContext, responseObserver); + String groupName = GrpcConverter.wrapResourceWithNamespace(request.getGroup()); + attachChannelId(ctx, proxyContext, groupName); try { Settings settings = this.grpcClientSettingsManager.getClientSettings(proxyContext); @@ -68,10 +81,11 @@ public class ReceiveMessageActivity extends AbstractMessingActivity { return; } - long invisibleTime = Durations.toMillis(request.getInvisibleDuration()); - if (request.getAutoRenew()) { - invisibleTime = Durations.toMillis(subscription.getLongPollingTimeout() - ); + final long requestInvisibleTime = Durations.toMillis(request.getInvisibleDuration()); + long actualInvisibleTime = requestInvisibleTime; + ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig(); + if (proxyConfig.isEnableProxyAutoRenew() && request.getAutoRenew()) { + actualInvisibleTime = Math.min(actualInvisibleTime, proxyConfig.getRenewSliceTimeMillis()); } String topic = GrpcConverter.wrapResourceWithNamespace(request.getMessageQueue().getTopic()); @@ -94,14 +108,30 @@ public class ReceiveMessageActivity extends AbstractMessingActivity { group, topic, request.getBatchSize(), - invisibleTime, + actualInvisibleTime, pollTime, ConsumeInitMode.MAX, subscriptionData, fifo, new PopMessageResultFilterImpl(grpcClientSettingsManager), timeRemaining - ).thenAccept(popResult -> writer.writeAndComplete(proxyContext, request, popResult)) + ).thenAccept(popResult -> { + if (proxyConfig.isEnableProxyAutoRenew() && request.getAutoRenew()) { + List messageExtList = popResult.getMsgFoundList(); + for (MessageExt messageExt : messageExtList) { + String receiptHandle = messageExt.getProperty(MessageConst.PROPERTY_POP_CK); + if (receiptHandle != null) { + MessageQueue messageQueue = new MessageQueue(topic, messageExt.getBrokerName(), messageExt.getQueueId()); + MessageReceiptHandle messageReceiptHandle = + new MessageReceiptHandle(group, messageQueue, receiptHandle, messageExt.getMsgId(), + messageExt.getQueueOffset(), messageExt.getReconsumeTimes(), requestInvisibleTime); + String channelId = proxyContext.getVal(ContextVariable.CHANNEL_KEY); + receiptHandleProcessor.addReceiptHandle(channelId, receiptHandle, messageReceiptHandle); + } + } + } + writer.writeAndComplete(proxyContext, request, popResult); + }) .exceptionally(t -> { writer.writeAndComplete(proxyContext, request, t); return null; diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/ForwardMessageToDLQActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/ForwardMessageToDLQActivity.java index d6133ded44..ed7f64d32d 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/ForwardMessageToDLQActivity.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/ForwardMessageToDLQActivity.java @@ -23,6 +23,7 @@ import java.util.concurrent.CompletableFuture; import org.apache.rocketmq.common.consumer.ReceiptHandle; import org.apache.rocketmq.proxy.common.ProxyContext; import org.apache.rocketmq.proxy.grpc.v2.AbstractMessingActivity; +import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager; import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager; import org.apache.rocketmq.proxy.grpc.v2.common.GrpcConverter; import org.apache.rocketmq.proxy.grpc.v2.common.ResponseBuilder; @@ -32,8 +33,8 @@ import org.apache.rocketmq.remoting.protocol.RemotingCommand; public class ForwardMessageToDLQActivity extends AbstractMessingActivity { public ForwardMessageToDLQActivity(MessagingProcessor messagingProcessor, - GrpcClientSettingsManager grpcClientSettingsManager) { - super(messagingProcessor, grpcClientSettingsManager); + GrpcClientSettingsManager grpcClientSettingsManager, GrpcChannelManager grpcChannelManager) { + super(messagingProcessor, grpcClientSettingsManager, grpcChannelManager); } public CompletableFuture forwardMessageToDeadLetterQueue(Context ctx, diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivity.java index 50419381c2..5435900416 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivity.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivity.java @@ -44,6 +44,7 @@ import org.apache.rocketmq.common.message.MessageConst; import org.apache.rocketmq.common.sysflag.MessageSysFlag; import org.apache.rocketmq.proxy.common.ProxyContext; import org.apache.rocketmq.proxy.grpc.v2.AbstractMessingActivity; +import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager; import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager; import org.apache.rocketmq.proxy.grpc.v2.common.GrpcConverter; import org.apache.rocketmq.proxy.grpc.v2.common.GrpcProxyException; @@ -56,8 +57,8 @@ import org.apache.rocketmq.proxy.service.route.SelectableMessageQueue; public class SendMessageActivity extends AbstractMessingActivity { public SendMessageActivity(MessagingProcessor messagingProcessor, - GrpcClientSettingsManager grpcClientSettingsManager) { - super(messagingProcessor, grpcClientSettingsManager); + GrpcClientSettingsManager grpcClientSettingsManager, GrpcChannelManager grpcChannelManager) { + super(messagingProcessor, grpcClientSettingsManager, grpcChannelManager); } public CompletableFuture sendMessage(Context ctx, SendMessageRequest request) { diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/route/RouteActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/route/RouteActivity.java index a36608614a..f6f9982a59 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/route/RouteActivity.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/route/RouteActivity.java @@ -41,6 +41,7 @@ import org.apache.rocketmq.common.constant.PermName; import org.apache.rocketmq.common.protocol.route.QueueData; import org.apache.rocketmq.proxy.common.ProxyContext; import org.apache.rocketmq.proxy.grpc.v2.AbstractMessingActivity; +import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager; import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager; import org.apache.rocketmq.proxy.grpc.v2.common.GrpcConverter; import org.apache.rocketmq.proxy.grpc.v2.common.ResponseBuilder; @@ -51,8 +52,8 @@ import org.apache.rocketmq.proxy.service.route.TopicRouteHelper; public class RouteActivity extends AbstractMessingActivity { public RouteActivity(MessagingProcessor messagingProcessor, - GrpcClientSettingsManager grpcClientSettingsManager) { - super(messagingProcessor, grpcClientSettingsManager); + GrpcClientSettingsManager grpcClientSettingsManager, GrpcChannelManager grpcChannelManager) { + super(messagingProcessor, grpcClientSettingsManager, grpcChannelManager); } public CompletableFuture queryRoute(Context ctx, QueryRouteRequest request) { diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/transaction/EndTransactionActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/transaction/EndTransactionActivity.java index 35c6a02218..d33f1e0a11 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/transaction/EndTransactionActivity.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/transaction/EndTransactionActivity.java @@ -25,6 +25,7 @@ import io.grpc.Context; import java.util.concurrent.CompletableFuture; import org.apache.rocketmq.proxy.common.ProxyContext; import org.apache.rocketmq.proxy.grpc.v2.AbstractMessingActivity; +import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager; import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager; import org.apache.rocketmq.proxy.grpc.v2.common.GrpcConverter; import org.apache.rocketmq.proxy.grpc.v2.common.ResponseBuilder; @@ -35,8 +36,8 @@ import org.apache.rocketmq.proxy.service.transaction.TransactionId; public class EndTransactionActivity extends AbstractMessingActivity { public EndTransactionActivity(MessagingProcessor messagingProcessor, - GrpcClientSettingsManager grpcClientSettingsManager) { - super(messagingProcessor, grpcClientSettingsManager); + GrpcClientSettingsManager grpcClientSettingsManager, GrpcChannelManager grpcChannelManager) { + super(messagingProcessor, grpcClientSettingsManager, grpcChannelManager); } public CompletableFuture endTransaction(Context ctx, EndTransactionRequest request) { diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/AbstractProcessor.java b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/AbstractProcessor.java index 6e26198bd2..4815d7dbaf 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/AbstractProcessor.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/AbstractProcessor.java @@ -20,11 +20,12 @@ import org.apache.rocketmq.common.attribute.TopicMessageType; import org.apache.rocketmq.common.consumer.ReceiptHandle; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.common.message.MessageConst; +import org.apache.rocketmq.proxy.common.AbstractStartAndShutdown; import org.apache.rocketmq.proxy.common.ProxyException; import org.apache.rocketmq.proxy.common.ProxyExceptionCode; import org.apache.rocketmq.proxy.service.ServiceManager; -public abstract class AbstractProcessor { +public abstract class AbstractProcessor extends AbstractStartAndShutdown { protected MessagingProcessor messagingProcessor; protected ServiceManager serviceManager; diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ConsumerProcessor.java b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ConsumerProcessor.java index 63e51f3ea1..8dfe714200 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ConsumerProcessor.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ConsumerProcessor.java @@ -46,8 +46,8 @@ public class ConsumerProcessor extends AbstractProcessor { private final ExecutorService executor; - public ConsumerProcessor(MessagingProcessor messagingProcessor, - ServiceManager serviceManager, ExecutorService executor) { + public ConsumerProcessor(MessagingProcessor messagingProcessor, ServiceManager serviceManager, + ExecutorService executor) { super(messagingProcessor, serviceManager); this.executor = executor; } 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 0ae12c5c89..628f1af73d 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 @@ -45,6 +45,7 @@ import org.apache.rocketmq.proxy.config.ConfigurationManager; import org.apache.rocketmq.proxy.config.ProxyConfig; import org.apache.rocketmq.proxy.service.ServiceManager; import org.apache.rocketmq.proxy.service.ServiceManagerFactory; +import org.apache.rocketmq.proxy.service.metadata.MetadataService; import org.apache.rocketmq.proxy.service.relay.ProxyRelayService; import org.apache.rocketmq.proxy.service.route.ProxyTopicRouteData; import org.apache.rocketmq.proxy.service.transaction.TransactionId; @@ -226,4 +227,8 @@ public class DefaultMessagingProcessor extends AbstractStartAndShutdown implemen return this.serviceManager.getProxyRelayService(); } + @Override + public MetadataService getMetadataService() { + return this.serviceManager.getMetadataService(); + } } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/MessagingProcessor.java b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/MessagingProcessor.java index 5b2b9b2713..c01e9c882a 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/MessagingProcessor.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/MessagingProcessor.java @@ -38,6 +38,7 @@ import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig; import org.apache.rocketmq.proxy.common.Address; import org.apache.rocketmq.proxy.common.ProxyContext; import org.apache.rocketmq.proxy.common.StartAndShutdown; +import org.apache.rocketmq.proxy.service.metadata.MetadataService; import org.apache.rocketmq.proxy.service.relay.ProxyRelayService; import org.apache.rocketmq.proxy.service.route.ProxyTopicRouteData; import org.apache.rocketmq.proxy.service.transaction.TransactionId; @@ -228,4 +229,6 @@ public interface MessagingProcessor extends StartAndShutdown { ); ProxyRelayService getProxyOutService(); + + MetadataService getMetadataService(); } 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 new file mode 100644 index 0000000000..6b2aaca25e --- /dev/null +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ReceiptHandleProcessor.java @@ -0,0 +1,202 @@ +/* + * 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.Channel; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import org.apache.rocketmq.broker.client.ClientChannelInfo; +import org.apache.rocketmq.broker.client.ConsumerGroupEvent; +import org.apache.rocketmq.broker.client.ConsumerIdsChangeListener; +import org.apache.rocketmq.client.consumer.AckResult; +import org.apache.rocketmq.client.consumer.AckStatus; +import org.apache.rocketmq.common.ThreadFactoryImpl; +import org.apache.rocketmq.common.constant.LoggerName; +import org.apache.rocketmq.common.consumer.ReceiptHandle; +import org.apache.rocketmq.common.subscription.RetryPolicy; +import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig; +import org.apache.rocketmq.common.thread.ThreadPoolMonitor; +import org.apache.rocketmq.proxy.common.AbstractStartAndShutdown; +import org.apache.rocketmq.proxy.common.MessageReceiptHandle; +import org.apache.rocketmq.proxy.common.ProxyContext; +import org.apache.rocketmq.proxy.common.ReceiptHandleGroup; +import org.apache.rocketmq.proxy.common.StartAndShutdown; +import org.apache.rocketmq.proxy.common.utils.ChannelUtils; +import org.apache.rocketmq.proxy.config.ConfigurationManager; +import org.apache.rocketmq.proxy.config.ProxyConfig; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class ReceiptHandleProcessor extends AbstractStartAndShutdown { + protected final static Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME); + protected final ConcurrentMap receiptHandleGroupMap; + protected final ScheduledExecutorService scheduledExecutorService = + Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("RenewalScheduledThread_")); + protected final ExecutorService renewalWorkerService = ThreadPoolMonitor.createAndMonitor( + 2, 4, 0, TimeUnit.MILLISECONDS, + "RenewalWorkerThread_", 10000 + ); + protected final MessagingProcessor messagingProcessor; + + public ReceiptHandleProcessor(MessagingProcessor messagingProcessor) { + this.messagingProcessor = messagingProcessor; + this.messagingProcessor.registerConsumerListener(new ConsumerIdsChangeListener() { + @Override + public void handle(ConsumerGroupEvent event, String group, Object... args) { + if (ConsumerGroupEvent.CLIENT_UNREGISTER.equals(event)) { + if (args == null || args.length < 1) { + return; + } + if (args[0] instanceof ClientChannelInfo) { + ClientChannelInfo clientChannelInfo = (ClientChannelInfo) args[0]; + Channel channel = clientChannelInfo.getChannel(); + clearGroup(ChannelUtils.buildChannelKey(channel, group)); + } + } + } + + @Override + public void shutdown() { + + } + }); + this.receiptHandleGroupMap = new ConcurrentHashMap<>(); + this.init(); + } + + protected void init() { + this.appendStartAndShutdown(new StartAndShutdown() { + @Override + public void start() throws Exception { + log.info("scan for renewal start."); + scheduledExecutorService.scheduleAtFixedRate(() -> scheduleRenewTask(), 0, + ConfigurationManager.getProxyConfig().getRenewSchedulePeriodMillis(), TimeUnit.MILLISECONDS); + log.info("renewal queue has started"); + } + + @Override + public void shutdown() throws Exception { + scheduledExecutorService.shutdown(); + } + }); + } + + protected void scheduleRenewTask() { + ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig(); + for (Map.Entry entry : receiptHandleGroupMap.entrySet()) { + String key = entry.getKey(); + ReceiptHandleGroup group = entry.getValue(); + group.all().forEach((k, v) -> { + ReceiptHandle handle = ReceiptHandle.decode(v.getReceiptHandle()); + long now = System.currentTimeMillis(); + if (handle.getNextVisibleTime() - now > proxyConfig.getRenewAheadTimeMillis()) { + return; + } + SubscriptionGroupConfig subscriptionGroupConfig = + messagingProcessor.getMetadataService().getSubscriptionGroupConfig(v.getGroup()); + if (subscriptionGroupConfig == null) { + log.error("Group's subscriptionGroupConfig is null, group = {}", v.getGroup()); + return; + } + RetryPolicy retryPolicy = subscriptionGroupConfig.getGroupRetryPolicy().getRetryPolicy(); + renewalWorkerService.submit(() -> renewMessage(key, v, handle, retryPolicy)); + }); + } + + log.info("scan for renewal done."); + } + + protected void renewMessage(String key, MessageReceiptHandle messageReceiptHandle, + ReceiptHandle handle, RetryPolicy retryPolicy) { + ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig(); + long current = System.currentTimeMillis(); + if (current - messageReceiptHandle.getTimestamp() < messageReceiptHandle.getExpectInvisibleTime()) { + CompletableFuture future = + messagingProcessor.changeInvisibleTime(ProxyContext.create(), handle, messageReceiptHandle.getMessageId(), + messageReceiptHandle.getGroup(), messageReceiptHandle.getMessageQueue().getTopic(), proxyConfig.getRenewSliceTimeMillis()); + future.thenAccept(ackResult -> { + if (AckStatus.OK.equals(ackResult.getStatus())) { + messageReceiptHandle.update(ackResult.getExtraInfo()); + addReceiptHandle(key, messageReceiptHandle.getOriginalReceiptHandle(), messageReceiptHandle); + } + }); + } else { + CompletableFuture future = messagingProcessor.changeInvisibleTime(ProxyContext.create(), + handle, messageReceiptHandle.getMessageId(), messageReceiptHandle.getGroup(), + messageReceiptHandle.getMessageQueue().getTopic(), + retryPolicy.nextDelayDuration(messageReceiptHandle.getReconsumeTimes(), TimeUnit.MILLISECONDS)); + future.thenAccept(ackResult -> { + if (AckStatus.OK.equals(ackResult.getStatus())) { + removeReceiptHandle(key, messageReceiptHandle.getOriginalReceiptHandle()); + } + }); + } + } + + public void addReceiptHandle(String key, String receiptHandle, + MessageReceiptHandle messageReceiptHandle) { + if (key == null) { + return; + } + receiptHandleGroupMap.computeIfAbsent(key, + k -> new ReceiptHandleGroup()).put(receiptHandle, messageReceiptHandle); + } + + public void removeReceiptHandle(String key, String receiptHandle) { + if (key == null) { + return; + } + receiptHandleGroupMap.computeIfPresent(key, (k, v) -> { + v.remove(receiptHandle); + return v; + }); + } + + public void clearGroup(String key) { + if (key == null) { + return; + } + ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig(); + receiptHandleGroupMap.computeIfPresent(key, (k, v) -> { + Map all = v.all(); + all.forEach((key0, value0) -> { + ReceiptHandle receiptHandle = ReceiptHandle.decode(value0.getReceiptHandle()); + messagingProcessor.changeInvisibleTime( + ProxyContext.create(), + receiptHandle, + value0.getMessageId(), + value0.getGroup(), + value0.getMessageQueue().getTopic(), + proxyConfig.getInvisibleTimeMillisWhenClear() + ); + }); + return null; + } + ); + } + + protected String keyWithChannelId(String channelId, String groupName) { + return channelId + "%" + groupName; + } +} diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/SimpleChannel.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/SimpleChannel.java index 4b700c5ed7..35e817b5a2 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/SimpleChannel.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/SimpleChannel.java @@ -23,6 +23,7 @@ import io.netty.channel.Channel; import io.netty.channel.ChannelConfig; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelId; import io.netty.channel.ChannelMetadata; import io.netty.channel.ChannelOutboundBuffer; import io.netty.channel.DefaultChannelPromise; @@ -57,7 +58,11 @@ public class SimpleChannel extends AbstractChannel { * @param localAddress Local address */ public SimpleChannel(Channel parent, String remoteAddress, String localAddress) { - super(parent); + this(parent, null, remoteAddress, localAddress); + } + + public SimpleChannel(Channel parent, ChannelId id, String remoteAddress, String localAddress) { + super(parent, id); lastAccessTime = System.currentTimeMillis(); this.remoteAddress = remoteAddress; this.localAddress = localAddress; diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyChannel.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyChannel.java index 12a6ae6541..fb2d0b091c 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyChannel.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyChannel.java @@ -17,7 +17,6 @@ package org.apache.rocketmq.proxy.service.relay; -import io.netty.channel.AbstractChannel; import io.netty.channel.Channel; import io.netty.channel.ChannelConfig; import io.netty.channel.ChannelFuture; @@ -43,36 +42,29 @@ import org.apache.rocketmq.logging.InternalLogger; import org.apache.rocketmq.logging.InternalLoggerFactory; import org.apache.rocketmq.proxy.common.ContextVariable; import org.apache.rocketmq.proxy.common.ProxyContext; +import org.apache.rocketmq.proxy.service.channel.SimpleChannel; import org.apache.rocketmq.proxy.service.transaction.TransactionId; import org.apache.rocketmq.remoting.common.RemotingUtil; import org.apache.rocketmq.remoting.protocol.RemotingCommand; -public abstract class ProxyChannel extends AbstractChannel { +public abstract class ProxyChannel extends SimpleChannel { private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME); - protected final String remoteAddress; protected final SocketAddress remoteSocketAddress; - protected final String localAddress; protected final SocketAddress localSocketAddress; protected final ProxyRelayService proxyRelayService; - protected ProxyChannel(ProxyRelayService proxyRelayService, Channel parent, String remoteAddress, - String localAddress) { - super(parent); + protected ProxyChannel(ProxyRelayService proxyRelayService, Channel parent, String remoteAddress, String localAddress) { + super(parent, remoteAddress, localAddress); this.proxyRelayService = proxyRelayService; - this.remoteAddress = remoteAddress; this.remoteSocketAddress = RemotingUtil.string2SocketAddress(remoteAddress); - this.localAddress = localAddress; this.localSocketAddress = RemotingUtil.string2SocketAddress(localAddress); } - protected ProxyChannel(ProxyRelayService proxyRelayService, Channel parent, ChannelId id, String remoteAddress, - String localAddress) { - super(parent, id); + protected ProxyChannel(ProxyRelayService proxyRelayService, Channel parent, ChannelId id, String remoteAddress, String localAddress) { + super(parent, id, remoteAddress, localAddress); this.proxyRelayService = proxyRelayService; - this.remoteAddress = remoteAddress; this.remoteSocketAddress = RemotingUtil.string2SocketAddress(remoteAddress); - this.localAddress = localAddress; this.localSocketAddress = RemotingUtil.string2SocketAddress(localAddress); } diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/BaseActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/BaseActivityTest.java index bb28e43aea..86ecf5da57 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/BaseActivityTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/BaseActivityTest.java @@ -26,8 +26,10 @@ import java.util.concurrent.TimeUnit; import org.apache.rocketmq.common.protocol.header.ExtraInfoUtil; import org.apache.rocketmq.proxy.config.InitConfigAndLoggerTest; import org.apache.rocketmq.proxy.grpc.interceptor.InterceptorConstants; +import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager; import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager; import org.apache.rocketmq.proxy.processor.MessagingProcessor; +import org.apache.rocketmq.proxy.processor.ReceiptHandleProcessor; import org.apache.rocketmq.proxy.service.relay.ProxyRelayService; import org.junit.Ignore; import org.junit.runner.RunWith; @@ -42,7 +44,9 @@ public class BaseActivityTest extends InitConfigAndLoggerTest { protected static final Random RANDOM = new Random(); protected MessagingProcessor messagingProcessor; protected GrpcClientSettingsManager grpcClientSettingsManager; + protected GrpcChannelManager grpcChannelManager; protected ProxyRelayService proxyRelayService; + protected ReceiptHandleProcessor receiptHandleProcessor; protected static final String REMOTE_ADDR = "192.168.0.1:8080"; protected static final String LOCAL_ADDR = "127.0.0.1:8080"; @@ -55,12 +59,14 @@ public class BaseActivityTest extends InitConfigAndLoggerTest { messagingProcessor = mock(MessagingProcessor.class); grpcClientSettingsManager = mock(GrpcClientSettingsManager.class); proxyRelayService = mock(ProxyRelayService.class); + receiptHandleProcessor = mock(ReceiptHandleProcessor.class); metadata.put(InterceptorConstants.CLIENT_ID, CLIENT_ID); metadata.put(InterceptorConstants.LANGUAGE, "JAVA"); metadata.put(InterceptorConstants.REMOTE_ADDRESS, REMOTE_ADDR); metadata.put(InterceptorConstants.LOCAL_ADDRESS, LOCAL_ADDR); when(messagingProcessor.getProxyOutService()).thenReturn(proxyRelayService); + grpcChannelManager = new GrpcChannelManager(messagingProcessor.getProxyOutService()); } protected Context createContext() { diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java index 242d62ee5a..15bde9918d 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java @@ -89,7 +89,7 @@ public class ClientActivityTest extends BaseActivityTest { @Before public void before() throws Throwable { super.before(); - this.clientActivity = new ClientActivity(this.messagingProcessor, this.grpcClientSettingsManager); + this.clientActivity = new ClientActivity(this.messagingProcessor, this.grpcClientSettingsManager, grpcChannelManager); } protected TelemetryCommand sendProducerTelemetry(Context context) throws Throwable { diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/AckMessageActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/AckMessageActivityTest.java index d4a34cbb0c..523c30bfcf 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/AckMessageActivityTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/AckMessageActivityTest.java @@ -45,7 +45,7 @@ public class AckMessageActivityTest extends BaseActivityTest { @Before public void before() throws Throwable { super.before(); - this.ackMessageActivity = new AckMessageActivity(this.messagingProcessor, this.grpcClientSettingsManager); + this.ackMessageActivity = new AckMessageActivity(messagingProcessor, receiptHandleProcessor, grpcClientSettingsManager, grpcChannelManager); } @Test diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ChangeInvisibleDurationActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ChangeInvisibleDurationActivityTest.java index 4d6655530f..79a8648408 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ChangeInvisibleDurationActivityTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ChangeInvisibleDurationActivityTest.java @@ -45,7 +45,8 @@ public class ChangeInvisibleDurationActivityTest extends BaseActivityTest { @Before public void before() throws Throwable { super.before(); - this.changeInvisibleDurationActivity = new ChangeInvisibleDurationActivity(this.messagingProcessor, this.grpcClientSettingsManager); + this.changeInvisibleDurationActivity = new ChangeInvisibleDurationActivity(messagingProcessor, receiptHandleProcessor, + grpcClientSettingsManager, grpcChannelManager); } @Test diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivityTest.java index 40a5ed4243..ae8ac2b684 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivityTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivityTest.java @@ -69,7 +69,8 @@ public class ReceiveMessageActivityTest extends BaseActivityTest { @Before public void before() throws Throwable { super.before(); - this.receiveMessageActivity = new ReceiveMessageActivity(this.messagingProcessor, this.grpcClientSettingsManager); + this.receiveMessageActivity = new ReceiveMessageActivity(messagingProcessor, receiptHandleProcessor, + grpcClientSettingsManager, grpcChannelManager); } @Test @@ -117,8 +118,7 @@ public class ReceiveMessageActivityTest extends BaseActivityTest { any(), anyBoolean(), any(), - anyLong() - )).thenReturn(CompletableFuture.completedFuture(popResult)); + anyLong())).thenReturn(CompletableFuture.completedFuture(popResult)); this.receiveMessageActivity.receiveMessage( createContext(), diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/ForwardMessageToDLQActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/ForwardMessageToDLQActivityTest.java index 1bae776f9c..3e2a0b4200 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/ForwardMessageToDLQActivityTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/ForwardMessageToDLQActivityTest.java @@ -41,7 +41,7 @@ public class ForwardMessageToDLQActivityTest extends BaseActivityTest { @Before public void before() throws Throwable { super.before(); - this.forwardMessageToDLQActivity = new ForwardMessageToDLQActivity(this.messagingProcessor, this.grpcClientSettingsManager); + this.forwardMessageToDLQActivity = new ForwardMessageToDLQActivity(messagingProcessor, grpcClientSettingsManager, grpcChannelManager); } @Test diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivityTest.java index e4e87a221b..8ee4433763 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivityTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivityTest.java @@ -71,7 +71,7 @@ public class SendMessageActivityTest extends BaseActivityTest { @Before public void before() throws Throwable { super.before(); - this.sendMessageActivity = new SendMessageActivity(this.messagingProcessor, this.grpcClientSettingsManager); + this.sendMessageActivity = new SendMessageActivity(messagingProcessor, grpcClientSettingsManager, grpcChannelManager); } @Test diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/route/RouteActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/route/RouteActivityTest.java index 583dbc995e..8d10efb135 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/route/RouteActivityTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/route/RouteActivityTest.java @@ -77,7 +77,7 @@ public class RouteActivityTest extends BaseActivityTest { @Before public void before() throws Throwable { super.before(); - this.routeActivity = new RouteActivity(this.messagingProcessor, this.grpcClientSettingsManager); + this.routeActivity = new RouteActivity(messagingProcessor, grpcClientSettingsManager, grpcChannelManager); } @Test diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/transaction/EndTransactionActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/transaction/EndTransactionActivityTest.java index a2444a6e88..aa009ef7ae 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/transaction/EndTransactionActivityTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/transaction/EndTransactionActivityTest.java @@ -60,7 +60,7 @@ public class EndTransactionActivityTest extends BaseActivityTest { @Before public void before() throws Throwable { super.before(); - this.endTransactionActivity = new EndTransactionActivity(this.messagingProcessor, this.grpcClientSettingsManager); + this.endTransactionActivity = new EndTransactionActivity(messagingProcessor, grpcClientSettingsManager, grpcChannelManager); } @Test diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/processor/BaseProcessorTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/processor/BaseProcessorTest.java index f67d861158..58d7dce874 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/processor/BaseProcessorTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/processor/BaseProcessorTest.java @@ -83,6 +83,7 @@ public class BaseProcessorTest extends InitConfigAndLoggerTest { when(serviceManager.getTransactionService()).thenReturn(transactionService); when(serviceManager.getProxyRelayService()).thenReturn(proxyRelayService); when(serviceManager.getMetadataService()).thenReturn(metadataService); + when(messagingProcessor.getMetadataService()).thenReturn(metadataService); } protected static ProxyContext createContext() { diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ConsumerProcessorTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ConsumerProcessorTest.java index 35ab32a9e9..99ebba4223 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ConsumerProcessorTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ConsumerProcessorTest.java @@ -64,7 +64,8 @@ public class ConsumerProcessorTest extends BaseProcessorTest { @Before public void before() throws Throwable { super.before(); - this.consumerProcessor = new ConsumerProcessor(this.messagingProcessor, this.serviceManager, Executors.newCachedThreadPool()); + ReceiptHandleProcessor receiptHandleProcessor = new ReceiptHandleProcessor(messagingProcessor); + this.consumerProcessor = new ConsumerProcessor(messagingProcessor, serviceManager, Executors.newCachedThreadPool()); } @Test 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 0000000000..73c05a6784 --- /dev/null +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ReceiptHandleProcessorTest.java @@ -0,0 +1,192 @@ +/* + * 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 java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import org.apache.rocketmq.broker.client.ConsumerIdsChangeListener; +import org.apache.rocketmq.client.consumer.AckResult; +import org.apache.rocketmq.client.consumer.AckStatus; +import org.apache.rocketmq.common.consumer.ReceiptHandle; +import org.apache.rocketmq.common.message.MessageQueue; +import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig; +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.junit.Before; +import org.junit.Test; +import org.mockito.Mockito; + +public class ReceiptHandleProcessorTest extends BaseProcessorTest { + ReceiptHandleProcessor receiptHandleProcessor; + + ProxyContext context = ProxyContext.create(); + String group = "group"; + MessageQueue messageQueue = new MessageQueue("topic", "broker", 1); + String messageId = "messageId"; + long offset = 123L; + long invisibleTime = 100000L; + int reconsumeTimes = 1; + MessageReceiptHandle messageReceiptHandle; + + String receiptHandle = ReceiptHandle.builder() + .startOffset(0L) + .retrieveTime(0) + .invisibleTime(invisibleTime) + .reviveQueueId(1) + .topicType(ReceiptHandle.NORMAL_TOPIC) + .brokerName(messageQueue.getBrokerName()) + .queueId(messageQueue.getQueueId()) + .offset(offset) + .commitLogOffset(0L) + .build().encode(); + + @Before + public void setup() { + context.withVal(ContextVariable.CHANNEL_KEY, "channel-id"); + receiptHandleProcessor = new ReceiptHandleProcessor(messagingProcessor); + Mockito.doNothing().when(messagingProcessor).registerConsumerListener(Mockito.any(ConsumerIdsChangeListener.class)); + messageReceiptHandle = new MessageReceiptHandle(group, messageQueue, receiptHandle, messageId, offset, + reconsumeTimes, invisibleTime); + } + + @Test + public void testAddReceiptHandle() { + String channelId = context.getVal(ContextVariable.CHANNEL_KEY); + receiptHandleProcessor.addReceiptHandle(channelId, receiptHandle, messageReceiptHandle); + Mockito.when(metadataService.getSubscriptionGroupConfig(Mockito.eq(group))).thenReturn(new SubscriptionGroupConfig()); + receiptHandleProcessor.scheduleRenewTask(); + Mockito.verify(messagingProcessor, Mockito.timeout(1000).times(1)) + .changeInvisibleTime(Mockito.any(ProxyContext.class), Mockito.any(ReceiptHandle.class), Mockito.eq(messageId), + Mockito.eq(group), Mockito.eq(messageQueue.getTopic()), Mockito.eq(ConfigurationManager.getProxyConfig().getRenewSliceTimeMillis())); + } + + @Test + public void testRenewReceiptHandle() { + String channelId = context.getVal(ContextVariable.CHANNEL_KEY); + receiptHandleProcessor.addReceiptHandle(channelId, receiptHandle, messageReceiptHandle); + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + Mockito.when(metadataService.getSubscriptionGroupConfig(Mockito.eq(group))).thenReturn(groupConfig); + long newInvisibleTime = 2000L; + ReceiptHandle newReceiptHandleClass = ReceiptHandle.builder() + .startOffset(0L) + .retrieveTime(0) + .invisibleTime(newInvisibleTime) + .reviveQueueId(1) + .topicType(ReceiptHandle.NORMAL_TOPIC) + .brokerName(messageQueue.getBrokerName()) + .queueId(messageQueue.getQueueId()) + .offset(offset) + .commitLogOffset(0L) + .build(); + String newReceiptHandle = newReceiptHandleClass.encode(); + AckResult ackResult = new AckResult(); + ackResult.setStatus(AckStatus.OK); + ackResult.setExtraInfo(newReceiptHandle); + Mockito.when(messagingProcessor.changeInvisibleTime(Mockito.any(ProxyContext.class), Mockito.any(ReceiptHandle.class), Mockito.eq(messageId), + Mockito.eq(group), Mockito.eq(messageQueue.getTopic()), Mockito.eq(ConfigurationManager.getProxyConfig().getRenewSliceTimeMillis()))) + .thenReturn(CompletableFuture.completedFuture(ackResult)); + receiptHandleProcessor.scheduleRenewTask(); + Mockito.verify(messagingProcessor, Mockito.timeout(1000).times(1)) + .changeInvisibleTime(Mockito.any(ProxyContext.class), Mockito.argThat((r) -> r.getInvisibleTime() == invisibleTime), Mockito.eq(messageId), + Mockito.eq(group), Mockito.eq(messageQueue.getTopic()), Mockito.eq(ConfigurationManager.getProxyConfig().getRenewSliceTimeMillis())); + receiptHandleProcessor.scheduleRenewTask(); + Mockito.verify(messagingProcessor, Mockito.timeout(1000).times(1)) + .changeInvisibleTime(Mockito.any(ProxyContext.class), Mockito.argThat((r) -> r.getInvisibleTime() == newInvisibleTime), Mockito.eq(messageId), + Mockito.eq(group), Mockito.eq(messageQueue.getTopic()), Mockito.eq(ConfigurationManager.getProxyConfig().getRenewSliceTimeMillis())); + } + + @Test + public void testRenewReceiptHandleWhenTimeout() { + long newInvisibleTime = 0L; + String newReceiptHandle = ReceiptHandle.builder() + .startOffset(0L) + .retrieveTime(0) + .invisibleTime(newInvisibleTime) + .reviveQueueId(1) + .topicType(ReceiptHandle.NORMAL_TOPIC) + .brokerName(messageQueue.getBrokerName()) + .queueId(messageQueue.getQueueId()) + .offset(offset) + .commitLogOffset(0L) + .build().encode(); + messageReceiptHandle = new MessageReceiptHandle(group, messageQueue, receiptHandle, messageId, offset, + reconsumeTimes, newInvisibleTime); + String channelId = context.getVal(ContextVariable.CHANNEL_KEY); + receiptHandleProcessor.addReceiptHandle(channelId, newReceiptHandle, messageReceiptHandle); + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + Mockito.when(metadataService.getSubscriptionGroupConfig(Mockito.eq(group))).thenReturn(groupConfig); + receiptHandleProcessor.scheduleRenewTask(); + Mockito.verify(messagingProcessor, Mockito.timeout(1000).times(1)) + .changeInvisibleTime(Mockito.any(ProxyContext.class), Mockito.any(ReceiptHandle.class), Mockito.eq(messageId), + Mockito.eq(group), Mockito.eq(messageQueue.getTopic()), Mockito.eq(groupConfig.getGroupRetryPolicy().getRetryPolicy().nextDelayDuration(reconsumeTimes, TimeUnit.MILLISECONDS))); + } + + + @Test + public void testRenewReceiptHandleWhenNotArrivingTime() { + String newReceiptHandle = ReceiptHandle.builder() + .startOffset(0L) + .retrieveTime(System.currentTimeMillis()) + .invisibleTime(invisibleTime) + .reviveQueueId(1) + .topicType(ReceiptHandle.NORMAL_TOPIC) + .brokerName(messageQueue.getBrokerName()) + .queueId(messageQueue.getQueueId()) + .offset(offset) + .commitLogOffset(0L) + .build().encode(); + messageReceiptHandle = new MessageReceiptHandle(group, messageQueue, newReceiptHandle, messageId, offset, + reconsumeTimes, invisibleTime); + String channelId = context.getVal(ContextVariable.CHANNEL_KEY); + receiptHandleProcessor.addReceiptHandle(channelId, newReceiptHandle, messageReceiptHandle); + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + Mockito.when(metadataService.getSubscriptionGroupConfig(Mockito.eq(group))).thenReturn(groupConfig); + receiptHandleProcessor.scheduleRenewTask(); + Mockito.verify(messagingProcessor, Mockito.timeout(1000).times(0)) + .changeInvisibleTime(Mockito.any(ProxyContext.class), Mockito.any(ReceiptHandle.class), Mockito.anyString(), + Mockito.anyString(), Mockito.anyString(), Mockito.anyLong()); + } + + @Test + public void testRemoveReceiptHandle() { + String channelId = context.getVal(ContextVariable.CHANNEL_KEY); + receiptHandleProcessor.addReceiptHandle(channelId, receiptHandle, messageReceiptHandle); + receiptHandleProcessor.removeReceiptHandle(channelId, receiptHandle); + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + Mockito.when(metadataService.getSubscriptionGroupConfig(Mockito.eq(group))).thenReturn(groupConfig); + receiptHandleProcessor.scheduleRenewTask(); + Mockito.verify(messagingProcessor, Mockito.timeout(1000).times(0)) + .changeInvisibleTime(Mockito.any(ProxyContext.class), Mockito.any(ReceiptHandle.class), Mockito.anyString(), + Mockito.anyString(), Mockito.anyString(), Mockito.anyLong()); + } + + @Test + public void testClearGroup() { + String channelId = context.getVal(ContextVariable.CHANNEL_KEY); + receiptHandleProcessor.addReceiptHandle(channelId, receiptHandle, messageReceiptHandle); + receiptHandleProcessor.clearGroup(channelId); + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + Mockito.when(metadataService.getSubscriptionGroupConfig(Mockito.eq(group))).thenReturn(groupConfig); + receiptHandleProcessor.scheduleRenewTask(); + Mockito.verify(messagingProcessor, Mockito.timeout(1000).times(1)) + .changeInvisibleTime(Mockito.any(ProxyContext.class), Mockito.any(ReceiptHandle.class), Mockito.eq(messageId), + Mockito.eq(group), Mockito.eq(messageQueue.getTopic()), Mockito.eq(ConfigurationManager.getProxyConfig().getInvisibleTimeMillisWhenClear())); + } +} \ No newline at end of file