opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-21 13:49:50 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #3949] for code style

Browse Source
This commit is contained in:
kaiyi.lk
2022-07-13 11:30:50 +08:00
committed by zhouxiang
parent 405f5aad83
commit 73618554f2
56 changed files with 189 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java
+6 -2
View File
@@ -149,9 +149,13 @@ public class ProxyStartup {
}
public static void initThreadPoolMonitor() {
ThreadPoolMonitor.init();
ProxyConfig config = ConfigurationManager.getProxyConfig();
ThreadPoolMonitor.config(config.isEnablePrintJstack(), config.getPrintJstackInMillis());
ThreadPoolMonitor.config(
InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME),
InternalLoggerFactory.getLogger(LoggerName.PROXY_WATER_MARK_LOGGER_NAME),
config.isEnablePrintJstack(), config.getPrintJstackInMillis(),
config.getPrintThreadPoolStatusInMillis());
ThreadPoolMonitor.init();
}
public static void initLogger() throws JoranException {
proxy/src/main/java/org/apache/rocketmq/proxy/common/ParameterConverter.java
-24
View File
@@ -1,24 +0,0 @@
/*
* 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 io.grpc.Context;
@FunctionalInterface
public interface ParameterConverter<T, R> {
R convert(Context ctx, T parameter) throws Throwable;
}
proxy/src/main/java/org/apache/rocketmq/proxy/common/utils/FilterUtils.java
+1 -1
View File
@@ -24,7 +24,7 @@ public class FilterUtils {
* Whether the message's tag matches consumerGroup's SubscriptionData
*
* @param tagsSet, tagSet in {@link SubscriptionData}, tagSet empty means SubscriptionData.SUB_ALL(*)
* @param tags, message's tags, null means not tag attached to the message.
* @param tags, message's tags, null means not tag attached to the message.
*/
public static boolean isTagMatched(Set<String> tagsSet, String tags) {
if (tagsSet.isEmpty()) {
proxy/src/main/java/org/apache/rocketmq/proxy/common/utils/FutureUtils.java
+2 -1
View File
@@ -22,7 +22,8 @@ import java.util.concurrent.ExecutorService;
public class FutureUtils {
public static <T> CompletableFuture<T> appendNextFuture(CompletableFuture<T> future, CompletableFuture<T> nextFuture, ExecutorService executor) {
public static <T> CompletableFuture<T> appendNextFuture(CompletableFuture<T> future,
CompletableFuture<T> nextFuture, ExecutorService executor) {
future.whenCompleteAsync((t, throwable) -> {
if (throwable != null) {
nextFuture.completeExceptionally(throwable);
proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java
+9
View File
@@ -31,6 +31,7 @@ public class ProxyConfig {
*/
private boolean enablePrintJstack = true;
private long printJstackInMillis = Duration.ofSeconds(60).toMillis();
private long printThreadPoolStatusInMillis = Duration.ofSeconds(3).toMillis();
private String nameSrvAddr = "";
private String nameSrvDomain = "";
@@ -124,6 +125,14 @@ public class ProxyConfig {
this.printJstackInMillis = printJstackInMillis;
}
public long getPrintThreadPoolStatusInMillis() {
return printThreadPoolStatusInMillis;
}
public void setPrintThreadPoolStatusInMillis(long printThreadPoolStatusInMillis) {
this.printThreadPoolStatusInMillis = printThreadPoolStatusInMillis;
}
public String getNameSrvAddr() {
return nameSrvAddr;
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/interceptor/InterceptorConstants.java
-1
View File
@@ -35,7 +35,6 @@ public class InterceptorConstants {
public static final Metadata.Key<String> LOCAL_ADDRESS
= Metadata.Key.of("rpc-local-address", Metadata.ASCII_STRING_MARSHALLER);
public static final Metadata.Key<String> AUTHORIZATION
= Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER);
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/DefaultGrpcMessingActivity.java
+2 -1
View File
@@ -103,7 +103,8 @@ public class DefaultGrpcMessingActivity extends AbstractStartAndShutdown impleme
}
@Override
public void receiveMessage(Context ctx, ReceiveMessageRequest request, StreamObserver<ReceiveMessageResponse> responseObserver) {
public void receiveMessage(Context ctx, ReceiveMessageRequest request,
StreamObserver<ReceiveMessageResponse> responseObserver) {
this.receiveMessageActivity.receiveMessage(ctx, request, responseObserver);
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/GrpcMessingActivity.java
+8 -4
View File
@@ -53,17 +53,21 @@ public interface GrpcMessingActivity extends StartAndShutdown {
CompletableFuture<QueryAssignmentResponse> queryAssignment(Context ctx, QueryAssignmentRequest request);
void receiveMessage(Context ctx, ReceiveMessageRequest request, StreamObserver<ReceiveMessageResponse> responseObserver);
void receiveMessage(Context ctx, ReceiveMessageRequest request,
StreamObserver<ReceiveMessageResponse> responseObserver);
CompletableFuture<AckMessageResponse> ackMessage(Context ctx, AckMessageRequest request);
CompletableFuture<ForwardMessageToDeadLetterQueueResponse> forwardMessageToDeadLetterQueue(Context ctx, ForwardMessageToDeadLetterQueueRequest request);
CompletableFuture<ForwardMessageToDeadLetterQueueResponse> forwardMessageToDeadLetterQueue(Context ctx,
ForwardMessageToDeadLetterQueueRequest request);
CompletableFuture<EndTransactionResponse> endTransaction(Context ctx, EndTransactionRequest request);
CompletableFuture<NotifyClientTerminationResponse> notifyClientTermination(Context ctx, NotifyClientTerminationRequest request);
CompletableFuture<NotifyClientTerminationResponse> notifyClientTermination(Context ctx,
NotifyClientTerminationRequest request);
CompletableFuture<ChangeInvisibleDurationResponse> changeInvisibleDuration(Context ctx, ChangeInvisibleDurationRequest request);
CompletableFuture<ChangeInvisibleDurationResponse> changeInvisibleDuration(Context ctx,
ChangeInvisibleDurationRequest request);
StreamObserver<TelemetryCommand> telemetry(Context ctx, StreamObserver<TelemetryCommand> responseObserver);
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/channel/GrpcChannelManager.java
+2 -2
View File
@@ -54,7 +54,7 @@ public class GrpcChannelManager implements StartAndShutdown {
protected void init() {
this.scheduledExecutorService.scheduleAtFixedRate(
this::scanExpireResultFuture,
10, 10, TimeUnit.SECONDS
10, 10, TimeUnit.SECONDS
);
}
@@ -77,7 +77,7 @@ public class GrpcChannelManager implements StartAndShutdown {
return clientIdChannelMap.get(clientId);
}
public GrpcClientChannel removeChannel(String group, String clientId) {
public GrpcClientChannel removeChannel(String group, String clientId) {
AtomicReference<GrpcClientChannel> channelRef = new AtomicReference<>();
this.groupClientIdChannelMap.computeIfPresent(group, (groupKey, clientIdMap) -> {
channelRef.set(clientIdMap.remove(clientId));
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/channel/GrpcClientChannel.java
+2 -1
View File
@@ -50,7 +50,8 @@ public class GrpcClientChannel extends ProxyChannel {
private final String group;
private final String clientId;
public GrpcClientChannel(ProxyRelayService proxyRelayService, GrpcChannelManager grpcChannelManager, Context ctx, String group, String clientId) {
public GrpcClientChannel(ProxyRelayService proxyRelayService, GrpcChannelManager grpcChannelManager, Context ctx,
String group, String clientId) {
super(proxyRelayService, null, new GrpcChannelId(group, clientId),
InterceptorConstants.METADATA.get(ctx).get(InterceptorConstants.REMOTE_ADDRESS),
InterceptorConstants.METADATA.get(ctx).get(InterceptorConstants.LOCAL_ADDRESS));
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/ResponseBuilder.java
+1 -1
View File
@@ -62,7 +62,7 @@ public class ResponseBuilder {
.setMessage(message)
.build();
}
public static Code buildCode(int remotingResponseCode) {
switch (remotingResponseCode) {
case ResponseCode.SUCCESS:
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/PopMessageResultFilterImpl.java
+2 -1
View File
@@ -32,7 +32,8 @@ public class PopMessageResultFilterImpl implements PopMessageResultFilter {
}
@Override
public FilterResult filterMessage(ProxyContext ctx, String consumerGroup, SubscriptionData subscriptionData, MessageExt messageExt) {
public FilterResult filterMessage(ProxyContext ctx, String consumerGroup, SubscriptionData subscriptionData,
MessageExt messageExt) {
int maxAttempts = grpcClientSettingsManager.getClientSettings(ctx).getBackoffPolicy().getMaxAttempts();
if (!FilterUtils.isTagMatched(subscriptionData.getTagsSet(), messageExt.getTags())) {
return FilterResult.NO_MATCH;
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivity.java
+2 -1
View File
@@ -111,7 +111,8 @@ public class ReceiveMessageActivity extends AbstractMessingActivity {
}
}
protected ReceiveMessageResponseStreamWriter createWriter(ProxyContext ctx, StreamObserver<ReceiveMessageResponse> responseObserver) {
protected ReceiveMessageResponseStreamWriter createWriter(ProxyContext ctx,
StreamObserver<ReceiveMessageResponse> responseObserver) {
return new ReceiveMessageResponseStreamWriter(
this.messagingProcessor,
responseObserver
proxy/src/main/java/org/apache/rocketmq/proxy/processor/MessagingProcessor.java
+1 -1
View File
@@ -57,7 +57,7 @@ public interface MessagingProcessor extends StartAndShutdown {
ProxyContext ctx,
List<Address> requestHostAndPortList,
String topicName
) throws Exception;
) throws Exception;
default CompletableFuture<List<SendResult>> sendMessage(
ProxyContext ctx,
proxy/src/main/java/org/apache/rocketmq/proxy/processor/validator/TopicMessageTypeValidator.java
+2 -1
View File
@@ -22,8 +22,9 @@ import org.apache.rocketmq.common.attribute.TopicMessageType;
public interface TopicMessageTypeValidator {
/**
* Will throw {@link org.apache.rocketmq.proxy.common.ProxyException} if validate failed.
*
* @param topicMessageType Target topic
* @param messageType Message's type
* @param messageType Message's type
*/
void validate(TopicMessageType topicMessageType, TopicMessageType messageType);
}
proxy/src/main/java/org/apache/rocketmq/proxy/service/ClusterServiceManager.java
+1 -1
View File
@@ -37,10 +37,10 @@ import org.apache.rocketmq.proxy.service.message.MessageService;
import org.apache.rocketmq.proxy.service.metadata.ClusterMetadataService;
import org.apache.rocketmq.proxy.service.metadata.MetadataService;
import org.apache.rocketmq.proxy.service.mqclient.DoNothingClientRemotingProcessor;
import org.apache.rocketmq.proxy.service.mqclient.MQClientAPIFactory;
import org.apache.rocketmq.proxy.service.mqclient.ProxyClientRemotingProcessor;
import org.apache.rocketmq.proxy.service.relay.ClusterProxyRelayService;
import org.apache.rocketmq.proxy.service.relay.ProxyRelayService;
import org.apache.rocketmq.proxy.service.mqclient.MQClientAPIFactory;
import org.apache.rocketmq.proxy.service.route.ClusterTopicRouteService;
import org.apache.rocketmq.proxy.service.route.TopicRouteService;
import org.apache.rocketmq.proxy.service.transaction.ClusterTransactionService;
proxy/src/main/java/org/apache/rocketmq/proxy/service/LocalServiceManager.java
+4 -2
View File
@@ -115,11 +115,13 @@ public class LocalServiceManager extends AbstractStartAndShutdown implements Ser
}
private class LocalServiceManagerStartAndShutdown implements StartAndShutdown {
@Override public void start() throws Exception {
@Override
public void start() throws Exception {
LocalServiceManager.this.scheduledExecutorService.scheduleWithFixedDelay(channelManager::scanAndCleanChannels, 5, 5, TimeUnit.MINUTES);
}
@Override public void shutdown() throws Exception {
@Override
public void shutdown() throws Exception {
LocalServiceManager.this.scheduledExecutorService.shutdown();
}
}
proxy/src/main/java/org/apache/rocketmq/proxy/service/ServiceManager.java
+1 -1
View File
@@ -20,8 +20,8 @@ import org.apache.rocketmq.broker.client.ConsumerManager;
import org.apache.rocketmq.broker.client.ProducerManager;
import org.apache.rocketmq.proxy.common.StartAndShutdown;
import org.apache.rocketmq.proxy.service.message.MessageService;
import org.apache.rocketmq.proxy.service.relay.ProxyRelayService;
import org.apache.rocketmq.proxy.service.metadata.MetadataService;
import org.apache.rocketmq.proxy.service.relay.ProxyRelayService;
import org.apache.rocketmq.proxy.service.route.TopicRouteService;
import org.apache.rocketmq.proxy.service.transaction.TransactionService;
proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/SimpleChannel.java
+4 -3
View File
@@ -36,6 +36,7 @@ import org.apache.rocketmq.logging.InternalLoggerFactory;
/**
* SimpleChannel is used to handle writeAndFlush situation in processor
*
* @see io.netty.channel.ChannelHandlerContext#writeAndFlush
* @see io.netty.channel.Channel#writeAndFlush
*/
@@ -51,9 +52,9 @@ public class SimpleChannel extends AbstractChannel {
/**
* Creates a new instance.
*
* @param parent the parent of this channel. {@code null} if there's no parent.
* @param remoteAddress Remote address
* @param localAddress Local address
* @param parent the parent of this channel. {@code null} if there's no parent.
* @param remoteAddress Remote address
* @param localAddress Local address
*/
public SimpleChannel(Channel parent, String remoteAddress, String localAddress) {
super(parent);
proxy/src/main/java/org/apache/rocketmq/proxy/service/message/ClusterMessageService.java
+4 -3
View File
@@ -59,12 +59,12 @@ public class ClusterMessageService implements MessageService {
future = this.mqClientAPIFactory.getClient().sendMessageAsync(
messageQueue.getBrokerAddr(),
messageQueue.getBrokerName(), msgList.get(0), requestHeader, timeoutMillis)
.thenApply(Lists::newArrayList);
.thenApply(Lists::newArrayList);
} else {
future = this.mqClientAPIFactory.getClient().sendMessageAsync(
messageQueue.getBrokerAddr(),
messageQueue.getBrokerName(), msgList, requestHeader, timeoutMillis)
.thenApply(Lists::newArrayList);
.thenApply(Lists::newArrayList);
}
return future;
}
@@ -81,7 +81,8 @@ public class ClusterMessageService implements MessageService {
@Override
public void endTransactionOneway(ProxyContext ctx, TransactionId transactionId,
EndTransactionRequestHeader requestHeader, long timeoutMillis) throws MQBrokerException, RemotingException, InterruptedException {
EndTransactionRequestHeader requestHeader,
long timeoutMillis) throws MQBrokerException, RemotingException, InterruptedException {
this.mqClientAPIFactory.getClient().endTransactionOneway(
this.resolveBrokerAddr(transactionId.getBrokerName()),
requestHeader,
proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java
+8 -4
View File
@@ -75,7 +75,8 @@ public class LocalMessageService implements MessageService {
this.channelManager = channelManager;
}
@Override public CompletableFuture<List<SendResult>> sendMessage(ProxyContext ctx, SelectableMessageQueue messageQueue,
@Override
public CompletableFuture<List<SendResult>> sendMessage(ProxyContext ctx, SelectableMessageQueue messageQueue,
List<? extends Message> msgList, SendMessageRequestHeader requestHeader, long timeoutMillis) {
byte[] body;
String messageId;
@@ -162,7 +163,8 @@ public class LocalMessageService implements MessageService {
return future;
}
@Override public void endTransactionOneway(ProxyContext ctx, TransactionId transactionId,
@Override
public void endTransactionOneway(ProxyContext ctx, TransactionId transactionId,
EndTransactionRequestHeader requestHeader, long timeoutMillis) {
SimpleChannel channel = channelManager.createChannel(ctx);
ChannelHandlerContext channelHandlerContext = channel.getChannelHandlerContext();
@@ -175,7 +177,8 @@ public class LocalMessageService implements MessageService {
}
}
@Override public CompletableFuture<PopResult> popMessage(ProxyContext ctx, SelectableMessageQueue messageQueue,
@Override
public CompletableFuture<PopResult> popMessage(ProxyContext ctx, SelectableMessageQueue messageQueue,
PopMessageRequestHeader requestHeader, long timeoutMillis) {
RemotingCommand request = LocalRemotingCommand.createRequestCommand(RequestCode.POP_MESSAGE, requestHeader);
CompletableFuture<RemotingCommand> future = new CompletableFuture<>();
@@ -315,7 +318,8 @@ public class LocalMessageService implements MessageService {
});
}
@Override public CompletableFuture<AckResult> ackMessage(ProxyContext ctx, ReceiptHandle handle, String messageId,
@Override
public CompletableFuture<AckResult> ackMessage(ProxyContext ctx, ReceiptHandle handle, String messageId,
AckMessageRequestHeader requestHeader, long timeoutMillis) {
SimpleChannel channel = channelManager.createChannel(ctx);
ChannelHandlerContext channelHandlerContext = channel.getChannelHandlerContext();
proxy/src/main/java/org/apache/rocketmq/proxy/service/mqclient/MQClientAPIExt.java
+2 -1
View File
@@ -431,7 +431,8 @@ public class MQClientAPIExt extends MQClientAPIImpl {
return future;
}
public CompletableFuture<Long> searchOffsetAsync(String brokerAddr, String topic, int queueId , long timestamp, long timeoutMillis) {
public CompletableFuture<Long> searchOffsetAsync(String brokerAddr, String topic, int queueId, long timestamp,
long timeoutMillis) {
SearchOffsetRequestHeader requestHeader = new SearchOffsetRequestHeader();
requestHeader.setTopic(topic);
requestHeader.setQueueId(queueId);
proxy/src/main/java/org/apache/rocketmq/proxy/service/mqclient/ProxyClientRemotingProcessor.java
+2 -1
View File
@@ -46,7 +46,8 @@ public class ProxyClientRemotingProcessor extends ClientRemotingProcessor {
}
@Override
public RemotingCommand checkTransactionState(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
public RemotingCommand checkTransactionState(ChannelHandlerContext ctx,
RemotingCommand request) throws RemotingCommandException {
final ByteBuffer byteBuffer = ByteBuffer.wrap(request.getBody());
final MessageExt messageExt = MessageDecoder.decode(byteBuffer, true, false, false);
if (messageExt != null) {
proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ClusterProxyRelayService.java
+2 -1
View File
@@ -36,7 +36,8 @@ public class ClusterProxyRelayService implements ProxyRelayService {
return null;
}
@Override public CompletableFuture<ProxyRelayResult<ConsumeMessageDirectlyResult>> processConsumeMessageDirectly(
@Override
public CompletableFuture<ProxyRelayResult<ConsumeMessageDirectlyResult>> processConsumeMessageDirectly(
ProxyContext context, RemotingCommand command,
ConsumeMessageDirectlyResultRequestHeader header) {
return null;
proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyChannel.java
+4 -2
View File
@@ -56,7 +56,8 @@ public abstract class ProxyChannel extends AbstractChannel {
protected final ProxyRelayService proxyRelayService;
protected ProxyChannel(ProxyRelayService proxyRelayService, Channel parent, String remoteAddress, String localAddress) {
protected ProxyChannel(ProxyRelayService proxyRelayService, Channel parent, String remoteAddress,
String localAddress) {
super(parent);
this.proxyRelayService = proxyRelayService;
this.remoteAddress = remoteAddress;
@@ -65,7 +66,8 @@ public abstract class ProxyChannel extends AbstractChannel {
this.localSocketAddress = RemotingUtil.string2SocketAddress(localAddress);
}
protected ProxyChannel(ProxyRelayService proxyRelayService, Channel parent, ChannelId id, String remoteAddress, String localAddress) {
protected ProxyChannel(ProxyRelayService proxyRelayService, Channel parent, ChannelId id, String remoteAddress,
String localAddress) {
super(parent, id);
this.proxyRelayService = proxyRelayService;
this.remoteAddress = remoteAddress;
proxy/src/main/java/org/apache/rocketmq/proxy/service/route/LocalTopicRouteService.java
+2 -1
View File
@@ -56,7 +56,8 @@ public class LocalTopicRouteService extends TopicRouteService {
}
@Override
public ProxyTopicRouteData getTopicRouteForProxy(List<Address> requestHostAndPortList, String topicName) throws Exception {
public ProxyTopicRouteData getTopicRouteForProxy(List<Address> requestHostAndPortList,
String topicName) throws Exception {
MessageQueueView messageQueueView = getAllMessageQueueView(topicName);
TopicRouteData topicRouteData = messageQueueView.getTopicRouteData();
proxy/src/main/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelector.java
+2 -1
View File
@@ -193,7 +193,8 @@ public class MessageQueueSelector {
return Objects.hash(queues, brokerActingQueues);
}
@Override public String toString() {
@Override
public String toString() {
return MoreObjects.toStringHelper(this)
.add("queues", queues)
.add("brokerActingQueues", brokerActingQueues)
proxy/src/main/java/org/apache/rocketmq/proxy/service/route/MessageQueueView.java
+2 -1
View File
@@ -53,7 +53,8 @@ public class MessageQueueView {
return writeSelector;
}
@Override public String toString() {
@Override
public String toString() {
return MoreObjects.toStringHelper(this)
.add("readSelector", readSelector)
.add("writeSelector", writeSelector)
proxy/src/main/java/org/apache/rocketmq/proxy/service/route/SelectableMessageQueue.java
+2 -1
View File
@@ -72,7 +72,8 @@ public class SelectableMessageQueue implements Comparable<SelectableMessageQueue
return brokerAddr;
}
@Override public String toString() {
@Override
public String toString() {
return MoreObjects.toStringHelper(this)
.add("messageQueue", messageQueue)
.add("brokerAddr", brokerAddr)
proxy/src/main/java/org/apache/rocketmq/proxy/service/route/TopicRouteService.java
+4 -2
View File
@@ -82,11 +82,13 @@ public abstract class TopicRouteService extends AbstractStartAndShutdown {
public abstract MessageQueueView getCurrentMessageQueueView(String topicName) throws Exception;
public abstract ProxyTopicRouteData getTopicRouteForProxy(List<Address> requestHostAndPortList, String topicName) throws Exception;
public abstract ProxyTopicRouteData getTopicRouteForProxy(List<Address> requestHostAndPortList,
String topicName) throws Exception;
public abstract String getBrokerAddr(String brokerName) throws Exception;
protected static MessageQueueView getCacheMessageQueueWrapper(LoadingCache<String, MessageQueueView> topicCache, String key) throws Exception {
protected static MessageQueueView getCacheMessageQueueWrapper(LoadingCache<String, MessageQueueView> topicCache,
String key) throws Exception {
MessageQueueView res = topicCache.get(key);
if (res.isEmptyCachedQueue()) {
throw new MQClientException(ResponseCode.TOPIC_NOT_EXIST,
proxy/src/main/java/org/apache/rocketmq/proxy/service/transaction/ClusterTransactionService.java
+3 -2
View File
@@ -57,7 +57,8 @@ public class ClusterTransactionService implements StartAndShutdown, TransactionS
private final Map<String /* group */, Set<ClusterData>/* cluster list */> groupClusterData = new ConcurrentHashMap<>();
private TxHeartbeatServiceThread txHeartbeatServiceThread;
public ClusterTransactionService(TopicRouteService topicRouteService, ProducerManager producerManager, RPCHook rpcHook,
public ClusterTransactionService(TopicRouteService topicRouteService, ProducerManager producerManager,
RPCHook rpcHook,
MQClientAPIFactory mqClientAPIFactory) {
this.topicRouteService = topicRouteService;
this.mqClientAPIFactory = mqClientAPIFactory;
@@ -186,7 +187,7 @@ public class ClusterTransactionService implements StartAndShutdown, TransactionS
protected void sendHeartBeatToCluster(String clusterName, HeartbeatData heartbeatData) {
try {
MessageQueueView messageQueue = this.topicRouteService.getAllMessageQueueView(clusterName);
MessageQueueView messageQueue = this.topicRouteService.getAllMessageQueueView(clusterName);
List<BrokerData> brokerDataList = messageQueue.getTopicRouteData().getBrokerDatas();
if (brokerDataList == null) {
return;
proxy/src/main/java/org/apache/rocketmq/proxy/service/transaction/LocalTransactionService.java
+11 -4
View File
@@ -18,20 +18,27 @@ package org.apache.rocketmq.proxy.service.transaction;
import java.util.List;
/**
* no need to implements, because the channel of producer will put into the broker's producerManager
*/
public class LocalTransactionService implements TransactionService {
@Override public void addTransactionSubscription(String group, List<String> topicList) {
@Override
public void addTransactionSubscription(String group, List<String> topicList) {
}
@Override public void addTransactionSubscription(String group, String topic) {
@Override
public void addTransactionSubscription(String group, String topic) {
}
@Override public void replaceTransactionSubscription(String group, List<String> topicList) {
@Override
public void replaceTransactionSubscription(String group, List<String> topicList) {
}
@Override public void unSubscribeAllTransactionTopic(String group) {
@Override
public void unSubscribeAllTransactionTopic(String group) {
}
}
proxy/src/main/java/org/apache/rocketmq/proxy/service/transaction/TransactionId.java
+4 -2
View File
@@ -181,7 +181,8 @@ public class TransactionId {
this.proxyTransactionId = proxyTransactionId;
}
@Override public String toString() {
@Override
public String toString() {
return MoreObjects.toStringHelper(this)
.add("brokerName", brokerName)
.add("brokerTransactionId", brokerTransactionId)
@@ -230,7 +231,8 @@ public class TransactionId {
return new TransactionId(brokerName, brokerTransactionId, commitLogOffset, tranStateTableOffset, proxyTransactionId);
}
@Override public String toString() {
@Override
public String toString() {
return MoreObjects.toStringHelper(this)
.add("brokerName", brokerName)
.add("brokerTransactionId", brokerTransactionId)
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/BaseActivityTest.java
-1
View File
@@ -31,7 +31,6 @@ import org.apache.rocketmq.proxy.processor.MessagingProcessor;
import org.apache.rocketmq.proxy.service.relay.ProxyRelayService;
import org.junit.Ignore;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import static org.mockito.Mockito.mock;
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/GrpcMessagingApplicationTest.java
+2 -2
View File
@@ -48,7 +48,7 @@ public class GrpcMessagingApplicationTest extends InitConfigAndLoggerTest {
GrpcMessagingApplication grpcMessagingApplication;
private static final String TOPIC = "topic";
private static Endpoints GRPC_ENDPOINTS = Endpoints.newBuilder()
private static Endpoints grpcEndpoints = Endpoints.newBuilder()
.setScheme(AddressScheme.IPv4)
.addAddresses(Address.newBuilder().setHost("127.0.0.1").setPort(8080).build())
.addAddresses(Address.newBuilder().setHost("127.0.0.2").setPort(8080).build())
@@ -64,7 +64,7 @@ public class GrpcMessagingApplicationTest extends InitConfigAndLoggerTest {
public void testQueryRoute() {
CompletableFuture<QueryRouteResponse> future = new CompletableFuture<>();
QueryRouteRequest request = QueryRouteRequest.newBuilder()
.setEndpoints(GRPC_ENDPOINTS)
.setEndpoints(grpcEndpoints)
.setTopic(Resource.newBuilder().setName(TOPIC).build())
.build();
Mockito.when(grpcMessingActivity.queryRoute(Mockito.any(Context.class), Mockito.eq(request)))
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java
+14 -7
View File
@@ -261,13 +261,16 @@ public class ClientActivityTest extends BaseActivityTest {
when(grpcChannelManagerMock.getAndRemoveResponseFuture(anyString())).thenReturn((CompletableFuture) runningInfoFutureMock);
Context context = createContext();
StreamObserver<TelemetryCommand> streamObserver = clientActivity.telemetry(context, new StreamObserver<TelemetryCommand>() {
@Override public void onNext(TelemetryCommand value) {
@Override
public void onNext(TelemetryCommand value) {
}
@Override public void onError(Throwable t) {
@Override
public void onError(Throwable t) {
}
@Override public void onCompleted() {
@Override
public void onCompleted() {
}
});
streamObserver.onNext(TelemetryCommand.newBuilder()
@@ -290,13 +293,16 @@ public class ClientActivityTest extends BaseActivityTest {
when(grpcChannelManagerMock.getAndRemoveResponseFuture(anyString())).thenReturn((CompletableFuture) resultFutureMock);
Context context = createContext();
StreamObserver<TelemetryCommand> streamObserver = clientActivity.telemetry(context, new StreamObserver<TelemetryCommand>() {
@Override public void onNext(TelemetryCommand value) {
@Override
public void onNext(TelemetryCommand value) {
}
@Override public void onError(Throwable t) {
@Override
public void onError(Throwable t) {
}
@Override public void onCompleted() {
@Override
public void onCompleted() {
}
});
streamObserver.onNext(TelemetryCommand.newBuilder()
@@ -326,7 +332,8 @@ public class ClientActivityTest extends BaseActivityTest {
}
@Override public void onCompleted() {
@Override
public void onCompleted() {
}
};
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcClientSettingsManagerTest.java
+1 -1
View File
@@ -67,7 +67,7 @@ public class GrpcClientSettingsManagerTest extends BaseActivityTest {
subscriptionGroupConfig.setRetryMaxTimes(3);
subscriptionGroupConfig.getGroupRetryPolicy().setType(GroupRetryPolicyType.CUSTOMIZED);
subscriptionGroupConfig.getGroupRetryPolicy().setCustomizedRetryPolicy(new CustomizedRetryPolicy(new long[]{1000}));
subscriptionGroupConfig.getGroupRetryPolicy().setCustomizedRetryPolicy(new CustomizedRetryPolicy(new long[] {1000}));
settings = this.grpcClientSettingsManager.getClientSettings(context);
assertEquals(RetryPolicy.newBuilder()
.setMaxAttempts(3)
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/AckMessageActivityTest.java
+1 -2
View File
@@ -24,14 +24,13 @@ import apache.rocketmq.v2.Code;
import java.util.concurrent.CompletableFuture;
import org.apache.rocketmq.client.consumer.AckResult;
import org.apache.rocketmq.client.consumer.AckStatus;
import org.apache.rocketmq.common.protocol.header.ExtraInfoUtil;
import org.apache.rocketmq.proxy.common.ProxyException;
import org.apache.rocketmq.proxy.common.ProxyExceptionCode;
import org.apache.rocketmq.proxy.grpc.v2.BaseActivityTest;
import org.junit.Before;
import org.junit.Test;
import static org.junit.Assert.*;
import static org.junit.Assert.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ChangeInvisibleDurationActivityTest.java
+1 -1
View File
@@ -31,7 +31,7 @@ import org.junit.Before;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import static org.junit.Assert.*;
import static org.junit.Assert.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.when;
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivityTest.java
-1
View File
@@ -42,7 +42,6 @@ import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.grpc.v2.BaseActivityTest;
import org.apache.rocketmq.proxy.service.route.MessageQueueView;
import org.apache.rocketmq.proxy.service.route.SelectableMessageQueue;
import org.assertj.core.util.Lists;
import org.junit.Before;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/ForwardMessageToDLQActivityTest.java
+1 -1
View File
@@ -29,7 +29,7 @@ import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.junit.Before;
import org.junit.Test;
import static org.junit.Assert.*;
import static org.junit.Assert.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.when;
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/route/RouteActivityTest.java
+14 -14
View File
@@ -58,18 +58,18 @@ public class RouteActivityTest extends BaseActivityTest {
private static final Resource GRPC_TOPIC = Resource.newBuilder()
.setName(TOPIC)
.build();
private static Endpoints GRPC_ENDPOINTS = Endpoints.newBuilder()
private static Endpoints grpcEndpoints = Endpoints.newBuilder()
.setScheme(AddressScheme.IPv4)
.addAddresses(Address.newBuilder().setHost("127.0.0.1").setPort(8080).build())
.addAddresses(Address.newBuilder().setHost("127.0.0.2").setPort(8080).build())
.build();
private static List<org.apache.rocketmq.proxy.common.Address> ENDPOINTS_ADDRESS = new ArrayList<>();
private static List<org.apache.rocketmq.proxy.common.Address> addressArrayList = new ArrayList<>();
static {
ENDPOINTS_ADDRESS.add(new org.apache.rocketmq.proxy.common.Address(
addressArrayList.add(new org.apache.rocketmq.proxy.common.Address(
org.apache.rocketmq.proxy.common.Address.AddressScheme.IPv4,
HostAndPort.fromParts("127.0.0.1", 8080)));
ENDPOINTS_ADDRESS.add(new org.apache.rocketmq.proxy.common.Address(
addressArrayList.add(new org.apache.rocketmq.proxy.common.Address(
org.apache.rocketmq.proxy.common.Address.AddressScheme.IPv4,
HostAndPort.fromParts("127.0.0.2", 8080)));
}
@@ -89,16 +89,16 @@ public class RouteActivityTest extends BaseActivityTest {
QueryRouteResponse response = this.routeActivity.queryRoute(
createContext(),
QueryRouteRequest.newBuilder()
.setEndpoints(GRPC_ENDPOINTS)
.setEndpoints(grpcEndpoints)
.setTopic(Resource.newBuilder().setName(TOPIC).build())
.build()
).get();
assertEquals(ENDPOINTS_ADDRESS, addressListCaptor.getValue());
assertEquals(addressArrayList, addressListCaptor.getValue());
assertEquals(Code.OK, response.getStatus().getCode());
assertEquals(4, response.getMessageQueuesCount());
for (MessageQueue messageQueue : response.getMessageQueuesList()) {
assertEquals(GRPC_ENDPOINTS, messageQueue.getBroker().getEndpoints());
assertEquals(grpcEndpoints, messageQueue.getBroker().getEndpoints());
assertEquals(Permission.READ_WRITE, messageQueue.getPermission());
}
}
@@ -111,7 +111,7 @@ public class RouteActivityTest extends BaseActivityTest {
QueryRouteResponse response = this.routeActivity.queryRoute(
createContext(),
QueryRouteRequest.newBuilder()
.setEndpoints(GRPC_ENDPOINTS)
.setEndpoints(grpcEndpoints)
.setTopic(GRPC_TOPIC)
.build()
).get();
@@ -127,7 +127,7 @@ public class RouteActivityTest extends BaseActivityTest {
QueryAssignmentResponse response = this.routeActivity.queryAssignment(
createContext(),
QueryAssignmentRequest.newBuilder()
.setEndpoints(GRPC_ENDPOINTS)
.setEndpoints(grpcEndpoints)
.setTopic(GRPC_TOPIC)
.build()
).get();
@@ -143,7 +143,7 @@ public class RouteActivityTest extends BaseActivityTest {
QueryAssignmentResponse response = this.routeActivity.queryAssignment(
createContext(),
QueryAssignmentRequest.newBuilder()
.setEndpoints(GRPC_ENDPOINTS)
.setEndpoints(grpcEndpoints)
.setTopic(GRPC_TOPIC)
.build()
).get();
@@ -159,14 +159,14 @@ public class RouteActivityTest extends BaseActivityTest {
QueryAssignmentResponse response = this.routeActivity.queryAssignment(
createContext(),
QueryAssignmentRequest.newBuilder()
.setEndpoints(GRPC_ENDPOINTS)
.setEndpoints(grpcEndpoints)
.setTopic(GRPC_TOPIC)
.build()
).get();
assertEquals(Code.OK, response.getStatus().getCode());
assertEquals(1, response.getAssignmentsCount());
assertEquals(GRPC_ENDPOINTS, response.getAssignments(0).getMessageQueue().getBroker().getEndpoints());
assertEquals(grpcEndpoints, response.getAssignments(0).getMessageQueue().getBroker().getEndpoints());
}
private static ProxyTopicRouteData createProxyTopicRouteData(int r, int w, int p) {
@@ -175,8 +175,8 @@ public class RouteActivityTest extends BaseActivityTest {
ProxyTopicRouteData.ProxyBrokerData proxyBrokerData = new ProxyTopicRouteData.ProxyBrokerData();
proxyBrokerData.setCluster(CLUSTER);
proxyBrokerData.setBrokerName(BROKER_NAME);
proxyBrokerData.getBrokerAddrs().put(0L, ENDPOINTS_ADDRESS);
proxyBrokerData.getBrokerAddrs().put(1L, ENDPOINTS_ADDRESS);
proxyBrokerData.getBrokerAddrs().put(0L, addressArrayList);
proxyBrokerData.getBrokerAddrs().put(1L, addressArrayList);
proxyTopicRouteData.getBrokerDatas().add(proxyBrokerData);
return proxyTopicRouteData;
}
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/transaction/EndTransactionActivityTest.java
+1 -1
View File
@@ -91,7 +91,7 @@ public class EndTransactionActivityTest extends BaseActivityTest {
@Parameterized.Parameters
public static Collection<Object[]> parameters() {
Object[][] p = new Object[][]{
Object[][] p = new Object[][] {
{TransactionResolution.COMMIT, TransactionSource.SOURCE_CLIENT, TransactionStatus.COMMIT, false},
{TransactionResolution.ROLLBACK, TransactionSource.SOURCE_SERVER_CHECK, TransactionStatus.ROLLBACK, true},
{TransactionResolution.TRANSACTION_RESOLUTION_UNSPECIFIED, TransactionSource.SOURCE_SERVER_CHECK, TransactionStatus.UNKNOWN, true},
proxy/src/test/java/org/apache/rocketmq/proxy/processor/ConsumerProcessorTest.java
+1 -1
View File
@@ -58,7 +58,7 @@ public class ConsumerProcessorTest extends BaseProcessorTest {
private static final String CONSUMER_GROUP = "consumerGroup";
private static final String TOPIC = "topic";
private ConsumerProcessor consumerProcessor;
@Before
proxy/src/test/java/org/apache/rocketmq/proxy/processor/ProducerProcessorTest.java
+1 -1
View File
@@ -77,7 +77,7 @@ public class ProducerProcessorTest extends BaseProcessorTest {
sendResult.setMsgId(msgId);
ArgumentCaptor<SendMessageRequestHeader> requestHeaderArgumentCaptor = ArgumentCaptor.forClass(SendMessageRequestHeader.class);
when(this.messageService.sendMessage(any(), any(), any(), requestHeaderArgumentCaptor.capture(), anyLong()))
.thenReturn(CompletableFuture.completedFuture(Lists.newArrayList(sendResult)));
.thenReturn(CompletableFuture.completedFuture(Lists.newArrayList(sendResult)));
List<MessageExt> messageExtList = new ArrayList<>();
MessageExt messageExt = createMessageExt(TOPIC, "tag", 0, 0);
proxy/src/test/java/org/apache/rocketmq/proxy/service/metadata/ClusterMetadataServiceTest.java
-3
View File
@@ -18,14 +18,11 @@
package org.apache.rocketmq.proxy.service.metadata;
import java.util.HashMap;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.attribute.TopicMessageType;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.statictopic.TopicConfigAndQueueMapping;
import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.service.BaseServiceTest;
import org.apache.rocketmq.proxy.service.route.MessageQueueView;
import org.junit.Before;
import org.junit.Test;
proxy/src/test/java/org/apache/rocketmq/proxy/service/mqclient/MQClientAPIExtTest.java
-1
View File
@@ -313,7 +313,6 @@ public class MQClientAPIExtTest {
assertEquals(offset, mqClientAPI.getMaxOffsetAsync(BROKER_ADDR, TOPIC, 0, TIMEOUT).get().longValue());
}
@Test
public void testSearchOffsetAsync() throws Exception {
long offset = ThreadLocalRandom.current().nextLong();
proxy/src/test/java/org/apache/rocketmq/proxy/service/relay/ProxyChannelTest.java
+7 -4
View File
@@ -18,7 +18,6 @@
package org.apache.rocketmq.proxy.service.relay;
import io.netty.channel.Channel;
import java.net.SocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
@@ -40,7 +39,9 @@ import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import static org.junit.Assert.*;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.when;
@@ -57,11 +58,13 @@ public class ProxyChannelTest {
super(proxyRelayService, parent, remoteAddress, localAddress);
}
@Override public boolean isOpen() {
@Override
public boolean isOpen() {
return false;
}
@Override public boolean isActive() {
@Override
public boolean isActive() {
return false;
}
}
proxy/src/test/java/org/apache/rocketmq/proxy/service/route/ClusterTopicRouteServiceTest.java
+3 -2
View File
@@ -18,7 +18,6 @@
package org.apache.rocketmq.proxy.service.route;
import com.google.common.net.HostAndPort;
import java.util.ArrayList;
import java.util.List;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.MixAll;
@@ -30,7 +29,9 @@ import org.junit.Before;
import org.junit.Test;
import static org.assertj.core.api.Assertions.catchThrowableOfType;
import static org.junit.Assert.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.when;
proxy/src/test/java/org/apache/rocketmq/proxy/service/transaction/ClusterTransactionServiceTest.java
-3
View File
@@ -30,10 +30,8 @@ import org.apache.rocketmq.proxy.service.route.MessageQueueView;
import org.assertj.core.util.Lists;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import static org.awaitility.Awaitility.await;
import static org.junit.Assert.assertEquals;
@@ -43,7 +41,6 @@ import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.when;
public class ClusterTransactionServiceTest extends BaseServiceTest {
@Mock
proxy/src/test/resources/rmq-proxy-home/conf/logback_proxy.xml
+11 -11
View File
@@ -25,7 +25,7 @@
<rollingPolicy class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
<fileNamePattern>${user.home}/logs/rocketmqlogs/otherdays/proxy.%i.log.gz</fileNamePattern>
<minIndex>1</minIndex>
<maxIndex>20</maxIndex>
<maxIndex>10</maxIndex>
</rollingPolicy>
<triggeringPolicy class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
<maxFileSize>128MB</maxFileSize>
@@ -39,25 +39,25 @@
<appender-ref ref="RocketmqProxyAppender_inner"/>
</appender>
<appender name="RocketmqGrpcAppender_inner"
<appender name="RocketmqProxyWatermarkAppender_inner"
class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>${user.home}/logs/rocketmqlogs/grpc.log</file>
<file>${user.home}/logs/rocketmqlogs/proxy_watermark.log</file>
<append>true</append>
<rollingPolicy class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
<fileNamePattern>${user.home}/logs/rocketmqlogs/otherdays/grpc.%i.log.gz</fileNamePattern>
<fileNamePattern>${user.home}/logs/rocketmqlogs/otherdays/proxy_watermark.%i.log.gz</fileNamePattern>
<minIndex>1</minIndex>
<maxIndex>20</maxIndex>
<maxIndex>10</maxIndex>
</rollingPolicy>
<triggeringPolicy class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
<maxFileSize>128MB</maxFileSize>
</triggeringPolicy>
<encoder>
<pattern>%d{yyy-MM-dd HH:mm:ss,GMT+8} %p %t - %m%n</pattern>
<pattern>%d{yyy-MM-dd HH:mm:ss,GMT+8}%m%n</pattern>
<charset class="java.nio.charset.Charset">UTF-8</charset>
</encoder>
</appender>
<appender name="RocketmqGrpcAppender" class="ch.qos.logback.classic.AsyncAppender">
<appender-ref ref="RocketmqGrpcAppender_inner"/>
<appender name="RocketmqProxyWatermarkAppender" class="ch.qos.logback.classic.AsyncAppender">
<appender-ref ref="RocketmqProxyWatermarkAppender_inner"/>
</appender>
<!-- Below is the logger configuration for broker-->
@@ -312,7 +312,7 @@
<maxIndex>20</maxIndex>
</rollingPolicy>
<triggeringPolicy
class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
<maxFileSize>128MB</maxFileSize>
</triggeringPolicy>
<encoder>
@@ -408,9 +408,9 @@
<appender-ref ref="RocketmqProxyAppender" />
</logger>
<logger name="RocketmqGrpc" additivity="false">
<logger name="RocketmqProxyWatermark" additivity="false">
<level value="INFO" />
<appender-ref ref="RocketmqGrpcAppender" />
<appender-ref ref="RocketmqProxyWatermarkAppender" />
</logger>
<root>