You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by du...@apache.org on 2019/01/15 11:26:47 UTC
[rocketmq] branch snode updated (74a5b3f -> f7c55b9)
This is an automated email from the ASF dual-hosted git repository.
duhengforever pushed a change to branch snode
in repository https://gitbox.apache.org/repos/asf/rocketmq.git.
from 74a5b3f Add request qps and request size per second flow control strategy
new ac25bcc Modify flow control config file
new f7c55b9 Fix unit test ensure mvn install pass
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../rocketmq/broker/client/ClientChannelInfo.java | 21 +++++-----
.../broker/client/ClientHousekeepingService.java | 16 ++++----
.../rocketmq/broker/client/ConsumerGroupInfo.java | 15 ++++---
.../rocketmq/broker/client/ConsumerManager.java | 12 +++---
.../client/DefaultConsumerIdsChangeListener.java | 3 --
.../rocketmq/broker/client/ProducerManager.java | 26 ++++--------
.../rocketmq/broker/client/net/Broker2Client.java | 14 +++----
.../apache/rocketmq/broker/out/BrokerOuterAPI.java | 5 +--
.../broker/processor/AdminBrokerProcessor.java | 6 +--
.../AbstractTransactionalMessageCheckListener.java | 12 +++---
.../broker/client/ProducerManagerTest.java | 26 +++++++-----
.../processor/ClientManageProcessorTest.java | 40 ++++++++++--------
.../processor/EndTransactionProcessorTest.java | 32 ++++++++-------
.../broker/processor/PullMessageProcessorTest.java | 47 +++++++++------------
.../broker/processor/SendMessageProcessorTest.java | 46 ++++++++++-----------
.../consumer/store/RemoteBrokerOffsetStore.java | 3 +-
.../rocketmq/client/impl/MQClientAPIImpl.java | 6 +--
.../client/impl/factory/MQClientInstance.java | 6 +--
.../client/consumer/DefaultMQPushConsumerTest.java | 1 -
.../java/org/apache/rocketmq/common/UtilAll.java | 4 --
.../common/flowcontrol/FlowControlConfig.java | 5 +--
.../common/flowcontrol/FlowControlRule.java | 2 -
.../rocketmq/example/benchmark/Producer.java | 3 +-
.../rocketmq/example/quickstart/Producer.java | 1 -
.../namesrv/processor/DefaultRequestProcessor.java | 3 +-
.../routeinfo/BrokerHousekeepingService.java | 24 +++++++----
.../processor/DefaultRequestProcessorTest.java | 27 ++++++++----
.../apache/rocketmq/remoting/RemotingServer.java | 1 -
.../netty/NettyChannelHandlerContextImpl.java | 2 +-
.../rocketmq/remoting/netty/NettyChannelImpl.java | 2 +-
.../remoting/transport/http2/Http2ClientImpl.java | 16 +++-----
.../remoting/transport/http2/Http2ServerImpl.java | 48 ++++++++++------------
.../transport/rocketmq/NettyRemotingClient.java | 10 +----
.../transport/rocketmq/NettyRemotingServer.java | 3 +-
.../remoting/netty/NettyRemotingClientTest.java | 2 +-
.../remoting/protocol/RemotingCommandTest.java | 2 +-
.../client/DefaultConsumerIdsChangeListener.java | 4 +-
.../apache/rocketmq/snode/config/SnodeConfig.java | 20 +++------
.../rocketmq/snode/constant/SnodeConstant.java | 7 ++--
.../RequestSizeFlowControlServiceImpl.java | 1 -
.../snode/processor/HeartbeatProcessor.java | 2 +-
.../snode/processor/SendMessageProcessor.java | 1 -
.../snode/service/impl/EnodeServiceImpl.java | 25 +++++------
.../snode/service/impl/NnodeServiceImpl.java | 6 +--
.../snode/service/impl/PushServiceImpl.java | 2 +-
.../resources/META-INF/service/flowControl.yml | 18 ++++----
.../snode/processor/SendMessageProcessorTest.java | 4 --
47 files changed, 265 insertions(+), 317 deletions(-)
[rocketmq] 01/02: Modify flow control config file
Posted by du...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
duhengforever pushed a commit to branch snode
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
commit ac25bccca6f73899878eeccbe90c8067c5f1c70e
Author: duhenglucky <du...@gmail.com>
AuthorDate: Tue Jan 15 13:47:32 2019 +0800
Modify flow control config file
---
.../apache/rocketmq/snode/config/SnodeConfig.java | 20 +++++---------------
.../main/resources/META-INF/service/flowControl.yml | 18 +++++++++---------
.../snode/processor/SendMessageProcessorTest.java | 4 ----
3 files changed, 14 insertions(+), 28 deletions(-)
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/config/SnodeConfig.java b/snode/src/main/java/org/apache/rocketmq/snode/config/SnodeConfig.java
index 6ec34a8..7f06ae7 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/config/SnodeConfig.java
+++ b/snode/src/main/java/org/apache/rocketmq/snode/config/SnodeConfig.java
@@ -53,7 +53,7 @@ public class SnodeConfig {
private long snodeHeartBeatInterval = 30 * 1000;
- private boolean fetechNameserver = false;
+ private boolean fetechNameServer = false;
private long houseKeepingInterval = 10 * 1000;
@@ -75,9 +75,6 @@ public class SnodeConfig {
private int listenPort = 11911;
- private double snodeQPSLimit = 10000;
-
-
private boolean vipChannelEnabled = Boolean.parseBoolean(System.getProperty(SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY, "true"));
public void setSnodeHeartBeatInterval(long snodeHeartBeatInterval) {
@@ -92,12 +89,12 @@ public class SnodeConfig {
this.houseKeepingInterval = houseKeepingInterval;
}
- public boolean isFetechNameserver() {
- return fetechNameserver;
+ public boolean isFetechNameServer() {
+ return fetechNameServer;
}
- public void setFetechNameserver(boolean fetechNameserver) {
- this.fetechNameserver = fetechNameserver;
+ public void setFetechNameServer(boolean fetechNameServer) {
+ this.fetechNameServer = fetechNameServer;
}
public long getSnodeHeartBeatInterval() {
@@ -285,11 +282,4 @@ public class SnodeConfig {
return remotingServerInterceptorPath;
}
- public double getSnodeQPSLimit() {
- return snodeQPSLimit;
- }
-
- public void setSnodeQPSLimit(double snodeQPSLimit) {
- this.snodeQPSLimit = snodeQPSLimit;
- }
}
diff --git a/snode/src/main/resources/META-INF/service/flowControl.yml b/snode/src/main/resources/META-INF/service/flowControl.yml
index f76a906..cf71d2d 100644
--- a/snode/src/main/resources/META-INF/service/flowControl.yml
+++ b/snode/src/main/resources/META-INF/service/flowControl.yml
@@ -1,19 +1,19 @@
snode:
countLimit: # flow control type, only requestCount & requestSize support
- flowControlResourceName: 310
- flowControlGrade: directDeny
- flowControlBehavior: flowControlBehavior
- flowControlResourceCount: 100000.00 #QPS
+ flowControlGrade: 1
+ flowControlBehavior: 1
+ flowControlResourceCount: 500.00 #QPS
- flowControlResourceName: overall
- flowControlGrade: directDeny
- flowControlBehavior: flowControlBehavior
- flowControlResourceCount: 100000.00 #QPS
+ flowControlGrade: 1
+ flowControlBehavior: 1
+ flowControlResourceCount: 10000.00 #QPS
sizeLimit:
- flowControlResourceName: 310
- flowControlGrade: directDeny
- flowControlBehavior: flowControlBehavior
- flowControlResourceCount: 10.00 #MB/S
+ flowControlGrade: 1
+ flowControlBehavior: 1
+ flowControlResourceCount: 5.00 #KB/S
topicLimit:
\ No newline at end of file
diff --git a/snode/src/test/java/org/apache/rocketmq/snode/processor/SendMessageProcessorTest.java b/snode/src/test/java/org/apache/rocketmq/snode/processor/SendMessageProcessorTest.java
index 82b4616..448abc9 100644
--- a/snode/src/test/java/org/apache/rocketmq/snode/processor/SendMessageProcessorTest.java
+++ b/snode/src/test/java/org/apache/rocketmq/snode/processor/SendMessageProcessorTest.java
@@ -104,10 +104,6 @@ public class SendMessageProcessorTest {
return request;
}
- private void assertSendMessageResult(int responseCode) throws RemotingCommandException {
-
- }
-
RemotingCommand createSendMessageResponse(int responseCode) {
return RemotingCommand.createResponseCommand(ResponseCode.SUCCESS, null);
}
[rocketmq] 02/02: Fix unit test ensure mvn install pass
Posted by du...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
duhengforever pushed a commit to branch snode
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
commit f7c55b9fd3e0777993f84152b9127cfc07dce18e
Author: duhenglucky <du...@gmail.com>
AuthorDate: Tue Jan 15 19:26:22 2019 +0800
Fix unit test ensure mvn install pass
---
.../rocketmq/broker/client/ClientChannelInfo.java | 21 +++++-----
.../broker/client/ClientHousekeepingService.java | 16 ++++----
.../rocketmq/broker/client/ConsumerGroupInfo.java | 15 ++++---
.../rocketmq/broker/client/ConsumerManager.java | 12 +++---
.../client/DefaultConsumerIdsChangeListener.java | 3 --
.../rocketmq/broker/client/ProducerManager.java | 26 ++++--------
.../rocketmq/broker/client/net/Broker2Client.java | 14 +++----
.../apache/rocketmq/broker/out/BrokerOuterAPI.java | 5 +--
.../broker/processor/AdminBrokerProcessor.java | 6 +--
.../AbstractTransactionalMessageCheckListener.java | 12 +++---
.../broker/client/ProducerManagerTest.java | 26 +++++++-----
.../processor/ClientManageProcessorTest.java | 40 ++++++++++--------
.../processor/EndTransactionProcessorTest.java | 32 ++++++++-------
.../broker/processor/PullMessageProcessorTest.java | 47 +++++++++------------
.../broker/processor/SendMessageProcessorTest.java | 46 ++++++++++-----------
.../consumer/store/RemoteBrokerOffsetStore.java | 3 +-
.../rocketmq/client/impl/MQClientAPIImpl.java | 6 +--
.../client/impl/factory/MQClientInstance.java | 6 +--
.../client/consumer/DefaultMQPushConsumerTest.java | 1 -
.../java/org/apache/rocketmq/common/UtilAll.java | 4 --
.../common/flowcontrol/FlowControlConfig.java | 5 +--
.../common/flowcontrol/FlowControlRule.java | 2 -
.../rocketmq/example/benchmark/Producer.java | 3 +-
.../rocketmq/example/quickstart/Producer.java | 1 -
.../namesrv/processor/DefaultRequestProcessor.java | 3 +-
.../routeinfo/BrokerHousekeepingService.java | 24 +++++++----
.../processor/DefaultRequestProcessorTest.java | 27 ++++++++----
.../apache/rocketmq/remoting/RemotingServer.java | 1 -
.../netty/NettyChannelHandlerContextImpl.java | 2 +-
.../rocketmq/remoting/netty/NettyChannelImpl.java | 2 +-
.../remoting/transport/http2/Http2ClientImpl.java | 16 +++-----
.../remoting/transport/http2/Http2ServerImpl.java | 48 ++++++++++------------
.../transport/rocketmq/NettyRemotingClient.java | 10 +----
.../transport/rocketmq/NettyRemotingServer.java | 3 +-
.../remoting/netty/NettyRemotingClientTest.java | 2 +-
.../remoting/protocol/RemotingCommandTest.java | 2 +-
.../client/DefaultConsumerIdsChangeListener.java | 4 +-
.../rocketmq/snode/constant/SnodeConstant.java | 7 ++--
.../RequestSizeFlowControlServiceImpl.java | 1 -
.../snode/processor/HeartbeatProcessor.java | 2 +-
.../snode/processor/SendMessageProcessor.java | 1 -
.../snode/service/impl/EnodeServiceImpl.java | 25 +++++------
.../snode/service/impl/NnodeServiceImpl.java | 6 +--
.../snode/service/impl/PushServiceImpl.java | 2 +-
44 files changed, 251 insertions(+), 289 deletions(-)
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/client/ClientChannelInfo.java b/broker/src/main/java/org/apache/rocketmq/broker/client/ClientChannelInfo.java
index 192a6f8..ef8be3e 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/client/ClientChannelInfo.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/client/ClientChannelInfo.java
@@ -16,12 +16,11 @@
*/
package org.apache.rocketmq.broker.client;
-import io.netty.channel.Channel;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.serialize.LanguageCode;
public class ClientChannelInfo {
- private final RemotingChannel channel;
+ private final RemotingChannel remotingChannel;
private final String clientId;
private final LanguageCode language;
private final int version;
@@ -31,15 +30,15 @@ public class ClientChannelInfo {
this(channel, null, null, 0);
}
- public ClientChannelInfo(RemotingChannel channel, String clientId, LanguageCode language, int version) {
- this.channel = channel;
+ public ClientChannelInfo(RemotingChannel remotingChannel, String clientId, LanguageCode language, int version) {
+ this.remotingChannel = remotingChannel;
this.clientId = clientId;
this.language = language;
this.version = version;
}
- public RemotingChannel getChannel() {
- return channel;
+ public RemotingChannel getRemotingChannel() {
+ return remotingChannel;
}
public String getClientId() {
@@ -66,7 +65,7 @@ public class ClientChannelInfo {
public int hashCode() {
final int prime = 31;
int result = 1;
- result = prime * result + ((channel == null) ? 0 : channel.hashCode());
+ result = prime * result + ((remotingChannel == null) ? 0 : remotingChannel.hashCode());
result = prime * result + ((clientId == null) ? 0 : clientId.hashCode());
result = prime * result + ((language == null) ? 0 : language.hashCode());
result = prime * result + (int) (lastUpdateTimestamp ^ (lastUpdateTimestamp >>> 32));
@@ -83,10 +82,10 @@ public class ClientChannelInfo {
if (getClass() != obj.getClass())
return false;
ClientChannelInfo other = (ClientChannelInfo) obj;
- if (channel == null) {
- if (other.channel != null)
+ if (remotingChannel == null) {
+ if (other.remotingChannel != null)
return false;
- } else if (this.channel != other.channel) {
+ } else if (this.remotingChannel != other.remotingChannel) {
return false;
}
@@ -95,7 +94,7 @@ public class ClientChannelInfo {
@Override
public String toString() {
- return "ClientChannelInfo [channel=" + channel + ", clientId=" + clientId + ", language=" + language
+ return "ClientChannelInfo [remotingChannel=" + remotingChannel + ", clientId=" + clientId + ", language=" + language
+ ", version=" + version + ", lastUpdateTimestamp=" + lastUpdateTimestamp + "]";
}
}
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/client/ClientHousekeepingService.java b/broker/src/main/java/org/apache/rocketmq/broker/client/ClientHousekeepingService.java
index 5a39e4f..8022a0f 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/client/ClientHousekeepingService.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/client/ClientHousekeepingService.java
@@ -74,28 +74,28 @@ public class ClientHousekeepingService implements ChannelEventListener {
log.info("Remoting channel closed: {}", RemotingHelper.parseChannelRemoteAddr(remotingChannel.remoteAddress()));
NettyChannelImpl nettyChannel = (NettyChannelImpl) remotingChannel;
Channel channel = nettyChannel.getChannel();
- this.brokerController.getProducerManager().doChannelCloseEvent(remoteAddr, channel);
- this.brokerController.getConsumerManager().doChannelCloseEvent(remoteAddr, channel);
+ this.brokerController.getProducerManager().doChannelCloseEvent(remoteAddr, remotingChannel);
+ this.brokerController.getConsumerManager().doChannelCloseEvent(remoteAddr, remotingChannel);
this.brokerController.getFilterServerManager().doChannelCloseEvent(remoteAddr, channel);
}
@Override
public void onChannelException(String remoteAddr, RemotingChannel remotingChannel) {
- log.info("Remoting channel exception: {}", RemotingHelper.parseChannelRemoteAddr(remotingChannel.remoteAddress()));
NettyChannelImpl nettyChannel = (NettyChannelImpl) remotingChannel;
Channel channel = nettyChannel.getChannel();
- this.brokerController.getProducerManager().doChannelCloseEvent(remoteAddr, channel);
- this.brokerController.getConsumerManager().doChannelCloseEvent(remoteAddr, channel);
+ log.info("Remoting channel exception: {}", RemotingHelper.parseChannelRemoteAddr(remotingChannel.remoteAddress()));
+ this.brokerController.getProducerManager().doChannelCloseEvent(remoteAddr, remotingChannel);
+ this.brokerController.getConsumerManager().doChannelCloseEvent(remoteAddr, remotingChannel);
this.brokerController.getFilterServerManager().doChannelCloseEvent(remoteAddr, channel);
}
@Override
public void onChannelIdle(String remoteAddr, RemotingChannel remotingChannel) {
- log.info("Remoting channel idle: {}", RemotingHelper.parseChannelRemoteAddr(remotingChannel.remoteAddress()));
NettyChannelImpl nettyChannel = (NettyChannelImpl) remotingChannel;
Channel channel = nettyChannel.getChannel();
- this.brokerController.getProducerManager().doChannelCloseEvent(remoteAddr, channel);
- this.brokerController.getConsumerManager().doChannelCloseEvent(remoteAddr, channel);
+ log.info("Remoting channel idle: {}", RemotingHelper.parseChannelRemoteAddr(remotingChannel.remoteAddress()));
+ this.brokerController.getProducerManager().doChannelCloseEvent(remoteAddr, remotingChannel);
+ this.brokerController.getConsumerManager().doChannelCloseEvent(remoteAddr, remotingChannel);
this.brokerController.getFilterServerManager().doChannelCloseEvent(remoteAddr, channel);
}
}
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/client/ConsumerGroupInfo.java b/broker/src/main/java/org/apache/rocketmq/broker/client/ConsumerGroupInfo.java
index 3631424..1e44df2 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/client/ConsumerGroupInfo.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/client/ConsumerGroupInfo.java
@@ -16,7 +16,6 @@
*/
package org.apache.rocketmq.broker.client;
-import io.netty.channel.Channel;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
@@ -26,11 +25,11 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
-import org.apache.rocketmq.logging.InternalLogger;
-import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.common.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.RemotingChannel;
public class ConsumerGroupInfo {
@@ -99,13 +98,13 @@ public class ConsumerGroupInfo {
}
public void unregisterChannel(final ClientChannelInfo clientChannelInfo) {
- ClientChannelInfo old = this.channelInfoTable.remove(clientChannelInfo.getChannel());
+ ClientChannelInfo old = this.channelInfoTable.remove(clientChannelInfo.getRemotingChannel());
if (old != null) {
log.info("unregister a consumer[{}] from consumerGroupInfo {}", this.groupName, old.toString());
}
}
- public boolean doChannelCloseEvent(final String remoteAddr, final Channel channel) {
+ public boolean doChannelCloseEvent(final String remoteAddr, final RemotingChannel channel) {
final ClientChannelInfo info = this.channelInfoTable.remove(channel);
if (info != null) {
log.warn(
@@ -124,9 +123,9 @@ public class ConsumerGroupInfo {
this.messageModel = messageModel;
this.consumeFromWhere = consumeFromWhere;
- ClientChannelInfo infoOld = this.channelInfoTable.get(infoNew.getChannel());
+ ClientChannelInfo infoOld = this.channelInfoTable.get(infoNew.getRemotingChannel());
if (null == infoOld) {
- ClientChannelInfo prev = this.channelInfoTable.put(infoNew.getChannel(), infoNew);
+ ClientChannelInfo prev = this.channelInfoTable.put(infoNew.getRemotingChannel(), infoNew);
if (null == prev) {
log.info("new consumer connected, group: {} {} {} channel: {}", this.groupName, consumeType,
messageModel, infoNew.toString());
@@ -140,7 +139,7 @@ public class ConsumerGroupInfo {
this.groupName,
infoOld.toString(),
infoNew.toString());
- this.channelInfoTable.put(infoNew.getChannel(), infoNew);
+ this.channelInfoTable.put(infoNew.getRemotingChannel(), infoNew);
}
}
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/client/ConsumerManager.java b/broker/src/main/java/org/apache/rocketmq/broker/client/ConsumerManager.java
index f621c1d..c39d7b4 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/client/ConsumerManager.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/client/ConsumerManager.java
@@ -16,7 +16,6 @@
*/
package org.apache.rocketmq.broker.client;
-import io.netty.channel.Channel;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map.Entry;
@@ -25,14 +24,13 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
-import org.apache.rocketmq.logging.InternalLogger;
-import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.common.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
-import org.apache.rocketmq.remoting.common.RemotingUtil;
public class ConsumerManager {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
@@ -75,7 +73,7 @@ public class ConsumerManager {
return 0;
}
- public void doChannelCloseEvent(final String remoteAddr, final Channel channel) {
+ public void doChannelCloseEvent(final String remoteAddr, final RemotingChannel channel) {
Iterator<Entry<String, ConsumerGroupInfo>> it = this.consumerTable.entrySet().iterator();
while (it.hasNext()) {
Entry<String, ConsumerGroupInfo> next = it.next();
@@ -159,9 +157,9 @@ public class ConsumerManager {
if (diff > CHANNEL_EXPIRED_TIMEOUT) {
log.warn(
"SCAN: remove expired channel from ConsumerManager consumerTable. channel={}, consumerGroup={}",
- RemotingHelper.parseChannelRemoteAddr(clientChannelInfo.getChannel().remoteAddress()), group);
+ RemotingHelper.parseChannelRemoteAddr(clientChannelInfo.getRemotingChannel().remoteAddress()), group);
- clientChannelInfo.getChannel().close();
+ clientChannelInfo.getRemotingChannel().close();
itChannel.remove();
}
}
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/client/DefaultConsumerIdsChangeListener.java b/broker/src/main/java/org/apache/rocketmq/broker/client/DefaultConsumerIdsChangeListener.java
index e2174dc..d583453 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/client/DefaultConsumerIdsChangeListener.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/client/DefaultConsumerIdsChangeListener.java
@@ -16,11 +16,8 @@
*/
package org.apache.rocketmq.broker.client;
-import io.netty.channel.Channel;
-
import java.util.Collection;
import java.util.List;
-
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.remoting.RemotingChannel;
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/client/ProducerManager.java b/broker/src/main/java/org/apache/rocketmq/broker/client/ProducerManager.java
index 1c9a557..11b2c15 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/client/ProducerManager.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/client/ProducerManager.java
@@ -16,8 +16,6 @@
*/
package org.apache.rocketmq.broker.client;
-import io.netty.channel.Channel;
-
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
@@ -27,14 +25,12 @@ import java.util.Map.Entry;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
-
import org.apache.rocketmq.broker.util.PositiveAtomicCounter;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
-import org.apache.rocketmq.remoting.common.RemotingUtil;
public class ProducerManager {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
@@ -86,8 +82,8 @@ public class ProducerManager {
it.remove();
log.warn(
"SCAN: remove expired channel[{}] from ProducerManager groupChannelTable, producer group name: {}",
- RemotingHelper.parseChannelRemoteAddr(info.getChannel().remoteAddress()), group);
- info.getChannel().close();
+ RemotingHelper.parseChannelRemoteAddr(info.getRemotingChannel().remoteAddress()), group);
+ info.getRemotingChannel().close();
}
}
}
@@ -102,7 +98,7 @@ public class ProducerManager {
}
}
- public void doChannelCloseEvent(final String remoteAddr, final Channel channel) {
+ public void doChannelCloseEvent(final String remoteAddr, final RemotingChannel channel) {
if (channel != null) {
try {
if (this.groupChannelLock.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
@@ -110,16 +106,12 @@ public class ProducerManager {
for (final Map.Entry<String, HashMap<RemotingChannel, ClientChannelInfo>> entry : this.groupChannelTable
.entrySet()) {
final String group = entry.getKey();
- final HashMap<RemotingChannel, ClientChannelInfo> clientChannelInfoTable =
- entry.getValue();
- final ClientChannelInfo clientChannelInfo =
- clientChannelInfoTable.remove(channel);
+ final HashMap<RemotingChannel, ClientChannelInfo> clientChannelInfoTable = entry.getValue();
+ final ClientChannelInfo clientChannelInfo = clientChannelInfoTable.remove(channel);
if (clientChannelInfo != null) {
- log.info(
- "NETTY EVENT: remove channel[{}][{}] from ProducerManager groupChannelTable, producer group: {}",
+ log.info("NETTY EVENT: remove channel[{}][{}] from ProducerManager groupChannelTable, producer group: {}",
clientChannelInfo.toString(), remoteAddr, group);
}
-
}
} finally {
this.groupChannelLock.unlock();
@@ -145,9 +137,9 @@ public class ProducerManager {
this.groupChannelTable.put(group, channelTable);
}
- clientChannelInfoFound = channelTable.get(clientChannelInfo.getChannel());
+ clientChannelInfoFound = channelTable.get(clientChannelInfo.getRemotingChannel());
if (null == clientChannelInfoFound) {
- channelTable.put(clientChannelInfo.getChannel(), clientChannelInfo);
+ channelTable.put(clientChannelInfo.getRemotingChannel(), clientChannelInfo);
log.info("new producer connected, group: {} channel: {}", group,
clientChannelInfo.toString());
}
@@ -172,7 +164,7 @@ public class ProducerManager {
try {
HashMap<RemotingChannel, ClientChannelInfo> channelTable = this.groupChannelTable.get(group);
if (null != channelTable && !channelTable.isEmpty()) {
- ClientChannelInfo old = channelTable.remove(clientChannelInfo.getChannel());
+ ClientChannelInfo old = channelTable.remove(clientChannelInfo.getRemotingChannel());
if (old != null) {
log.info("unregister a producer[{}] from groupChannelTable {}", group,
clientChannelInfo.toString());
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/client/net/Broker2Client.java b/broker/src/main/java/org/apache/rocketmq/broker/client/net/Broker2Client.java
index 4eee9db..370fee4 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/client/net/Broker2Client.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/client/net/Broker2Client.java
@@ -16,7 +16,12 @@
*/
package org.apache.rocketmq.broker.client.net;
-import io.netty.channel.Channel;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentMap;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.client.ClientChannelInfo;
import org.apache.rocketmq.broker.client.ConsumerGroupInfo;
@@ -45,13 +50,6 @@ import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.concurrent.ConcurrentMap;
-
public class Broker2Client {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
private final BrokerController brokerController;
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java b/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java
index d8e96a8..001058c 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java
@@ -47,6 +47,7 @@ import org.apache.rocketmq.common.protocol.header.namesrv.RegisterBrokerResponse
import org.apache.rocketmq.common.protocol.header.namesrv.UnRegisterBrokerRequestHeader;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
+import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.RemotingClient;
import org.apache.rocketmq.remoting.RemotingClientFactory;
@@ -55,7 +56,6 @@ import org.apache.rocketmq.remoting.exception.RemotingConnectException;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
-import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
public class BrokerOuterAPI {
@@ -193,9 +193,6 @@ public class BrokerOuterAPI {
RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, timeoutMills);
assert response != null;
- if (response == null){
- System.out.println("ssssssssssssss");
- }
switch (response.getCode()) {
case ResponseCode.SUCCESS: {
RegisterBrokerResponseHeader responseHeader =
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
index a83c488..914bb5b 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
@@ -603,7 +603,7 @@ public class AdminBrokerProcessor implements RequestProcessor {
connection.setClientId(info.getClientId());
connection.setLanguage(info.getLanguage());
connection.setVersion(info.getVersion());
- connection.setClientAddr(RemotingHelper.parseChannelRemoteAddr(info.getChannel().remoteAddress()));
+ connection.setClientAddr(RemotingHelper.parseChannelRemoteAddr(info.getRemotingChannel().remoteAddress()));
bodydata.getConnectionSet().add(connection);
}
@@ -638,7 +638,7 @@ public class AdminBrokerProcessor implements RequestProcessor {
connection.setClientId(info.getClientId());
connection.setLanguage(info.getLanguage());
connection.setVersion(info.getVersion());
- connection.setClientAddr(RemotingHelper.parseChannelRemoteAddr(info.getChannel().remoteAddress()));
+ connection.setClientAddr(RemotingHelper.parseChannelRemoteAddr(info.getRemotingChannel().remoteAddress()));
bodydata.getConnectionSet().add(connection);
}
@@ -1278,7 +1278,7 @@ public class AdminBrokerProcessor implements RequestProcessor {
newRequest.setExtFields(request.getExtFields());
newRequest.setBody(request.getBody());
- return this.brokerController.getBroker2Client().callClient(clientChannelInfo.getChannel(), newRequest);
+ return this.brokerController.getBroker2Client().callClient(clientChannelInfo.getRemotingChannel(), newRequest);
} catch (RemotingTimeoutException e) {
response.setCode(ResponseCode.CONSUME_MSG_TIMEOUT);
response
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/transaction/AbstractTransactionalMessageCheckListener.java b/broker/src/main/java/org/apache/rocketmq/broker/transaction/AbstractTransactionalMessageCheckListener.java
index 152e067..1a6c4be 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/transaction/AbstractTransactionalMessageCheckListener.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/transaction/AbstractTransactionalMessageCheckListener.java
@@ -16,7 +16,11 @@
*/
package org.apache.rocketmq.broker.transaction;
-import io.netty.channel.Channel;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.MessageConst;
@@ -24,12 +28,6 @@ import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.header.CheckTransactionStateRequestHeader;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
-
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.remoting.RemotingChannel;
public abstract class AbstractTransactionalMessageCheckListener {
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/client/ProducerManagerTest.java b/broker/src/test/java/org/apache/rocketmq/broker/client/ProducerManagerTest.java
index 2b01e7a..c52ee2e 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/client/ProducerManagerTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/client/ProducerManagerTest.java
@@ -21,6 +21,7 @@ import io.netty.channel.ChannelFuture;
import java.lang.reflect.Field;
import java.util.HashMap;
import org.apache.rocketmq.remoting.RemotingChannel;
+import org.apache.rocketmq.remoting.netty.NettyChannelImpl;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -37,35 +38,38 @@ public class ProducerManagerTest {
private String group = "FooBar";
private ClientChannelInfo clientInfo;
+ private RemotingChannel remotingChannel;
+
@Mock
- private RemotingChannel channel;
+ private Channel channel;
@Before
public void init() {
producerManager = new ProducerManager();
- clientInfo = new ClientChannelInfo(channel);
+ remotingChannel = new NettyChannelImpl(channel);
+ clientInfo = new ClientChannelInfo(remotingChannel);
}
@Test
public void scanNotActiveChannel() throws Exception {
producerManager.registerProducer(group, clientInfo);
- assertThat(producerManager.getGroupChannelTable().get(group).get(channel)).isNotNull();
-
+ assertThat(producerManager.getGroupChannelTable().get(group).get(remotingChannel)).isNotNull();
Field field = ProducerManager.class.getDeclaredField("CHANNEL_EXPIRED_TIMEOUT");
field.setAccessible(true);
long CHANNEL_EXPIRED_TIMEOUT = field.getLong(producerManager);
clientInfo.setLastUpdateTimestamp(System.currentTimeMillis() - CHANNEL_EXPIRED_TIMEOUT - 10);
-// when(channel.close()).thenReturn(mock(ChannelFuture.class));
+ when(channel.close()).thenReturn(mock(ChannelFuture.class));
producerManager.scanNotActiveChannel();
- assertThat(producerManager.getGroupChannelTable().get(group).get(channel)).isNull();
+ assertThat(producerManager.getGroupChannelTable().get(group).get(remotingChannel)).isNull();
}
@Test
public void doChannelCloseEvent() throws Exception {
producerManager.registerProducer(group, clientInfo);
- assertThat(producerManager.getGroupChannelTable().get(group).get(channel)).isNotNull();
-// producerManager.doChannelCloseEvent("127.0.0.1", channel);
- assertThat(producerManager.getGroupChannelTable().get(group).get(channel)).isNull();
+ assertThat(producerManager.getGroupChannelTable().get(group).get(remotingChannel)).isNotNull();
+ producerManager.doChannelCloseEvent("127.0.0.1", remotingChannel);
+ assertThat(producerManager.getGroupChannelTable().get(group).get(remotingChannel)).isNull();
+
}
@Test
@@ -73,7 +77,7 @@ public class ProducerManagerTest {
producerManager.registerProducer(group, clientInfo);
HashMap<RemotingChannel, ClientChannelInfo> channelMap = producerManager.getGroupChannelTable().get(group);
assertThat(channelMap).isNotNull();
- assertThat(channelMap.get(channel)).isEqualTo(clientInfo);
+ assertThat(channelMap.get(remotingChannel)).isEqualTo(clientInfo);
}
@Test
@@ -81,7 +85,7 @@ public class ProducerManagerTest {
producerManager.registerProducer(group, clientInfo);
HashMap<RemotingChannel, ClientChannelInfo> channelMap = producerManager.getGroupChannelTable().get(group);
assertThat(channelMap).isNotNull();
- assertThat(channelMap.get(channel)).isEqualTo(clientInfo);
+ assertThat(channelMap.get(remotingChannel)).isEqualTo(clientInfo);
producerManager.unregisterProducer(group, clientInfo);
channelMap = producerManager.getGroupChannelTable().get(group);
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/processor/ClientManageProcessorTest.java b/broker/src/test/java/org/apache/rocketmq/broker/processor/ClientManageProcessorTest.java
index 1be0309..ef00598 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/processor/ClientManageProcessorTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/processor/ClientManageProcessorTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.rocketmq.broker.processor;
+import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import java.util.HashMap;
import java.util.UUID;
@@ -27,14 +28,15 @@ import org.apache.rocketmq.common.protocol.RequestCode;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.header.UnregisterClientRequestHeader;
import org.apache.rocketmq.common.protocol.heartbeat.ConsumerData;
+import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.RemotingChannel;
+import org.apache.rocketmq.remoting.ServerConfig;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+import org.apache.rocketmq.remoting.netty.CodecHelper;
import org.apache.rocketmq.remoting.netty.NettyChannelHandlerContextImpl;
import org.apache.rocketmq.remoting.netty.NettyChannelImpl;
-import org.apache.rocketmq.remoting.ClientConfig;
-import org.apache.rocketmq.remoting.ServerConfig;
-import org.apache.rocketmq.remoting.serialize.LanguageCode;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.serialize.LanguageCode;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.junit.Before;
import org.junit.Test;
@@ -45,6 +47,7 @@ import org.mockito.junit.MockitoJUnitRunner;
import static org.apache.rocketmq.broker.processor.PullMessageProcessorTest.createConsumerData;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.when;
@RunWith(MockitoJUnitRunner.class)
public class ClientManageProcessorTest {
@@ -52,9 +55,15 @@ public class ClientManageProcessorTest {
@Spy
private BrokerController brokerController = new BrokerController(new BrokerConfig(), new ServerConfig(), new ClientConfig(), new MessageStoreConfig());
@Mock
- private ChannelHandlerContext handlerContext;
+ private NettyChannelHandlerContextImpl handlerContext;
+
@Mock
- private NettyChannelImpl channel;
+ private ChannelHandlerContext channelHandlerContext;
+
+ private RemotingChannel remotingChannel;
+
+ @Mock
+ private Channel channel;
private ClientChannelInfo clientChannelInfo;
private String clientId = UUID.randomUUID().toString();
@@ -63,11 +72,12 @@ public class ClientManageProcessorTest {
@Before
public void init() {
-// when(handlerContext.channel()).thenReturn(channel);
+ when(handlerContext.getChannelHandlerContext()).thenReturn(channelHandlerContext);
+ when(channelHandlerContext.channel()).thenReturn(channel);
clientManageProcessor = new ClientManageProcessor(brokerController);
- clientChannelInfo = new ClientChannelInfo(channel, clientId, LanguageCode.JAVA, 100);
+ remotingChannel = new NettyChannelImpl(channel);
+ clientChannelInfo = new ClientChannelInfo(remotingChannel, clientId, LanguageCode.JAVA, 100);
brokerController.getProducerManager().registerProducer(group, clientChannelInfo);
-
ConsumerData consumerData = createConsumerData(group, topic);
brokerController.getConsumerManager().registerConsumer(
consumerData.getGroupName(),
@@ -84,11 +94,10 @@ public class ClientManageProcessorTest {
brokerController.getProducerManager().registerProducer(group, clientChannelInfo);
HashMap<RemotingChannel, ClientChannelInfo> channelMap = brokerController.getProducerManager().getGroupChannelTable().get(group);
assertThat(channelMap).isNotNull();
- assertThat(channelMap.get(channel)).isEqualTo(clientChannelInfo);
+ assertThat(channelMap.get(remotingChannel)).isEqualTo(clientChannelInfo);
RemotingCommand request = createUnRegisterProducerCommand();
- NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
- RemotingCommand response = clientManageProcessor.processRequest(nettyChannelHandlerContext, request);
+ RemotingCommand response = clientManageProcessor.processRequest(handlerContext, request);
assertThat(response).isNotNull();
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
@@ -102,9 +111,8 @@ public class ClientManageProcessorTest {
assertThat(consumerGroupInfo).isNotNull();
RemotingCommand request = createUnRegisterConsumerCommand();
-
- NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
- RemotingCommand response = clientManageProcessor.processRequest(nettyChannelHandlerContext, request); assertThat(response).isNotNull();
+ RemotingCommand response = clientManageProcessor.processRequest(handlerContext, request);
+ assertThat(response).isNotNull();
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
consumerGroupInfo = brokerController.getConsumerManager().getConsumerGroupInfo(group);
@@ -118,7 +126,7 @@ public class ClientManageProcessorTest {
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UNREGISTER_CLIENT, requestHeader);
request.setLanguage(LanguageCode.JAVA);
request.setVersion(100);
-// request.makeCustomHeaderToNet();
+ CodecHelper.makeCustomHeaderToNet(request);
return request;
}
@@ -129,7 +137,7 @@ public class ClientManageProcessorTest {
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UNREGISTER_CLIENT, requestHeader);
request.setLanguage(LanguageCode.JAVA);
request.setVersion(100);
-// request.makeCustomHeaderToNet();
+ CodecHelper.makeCustomHeaderToNet(request);
return request;
}
}
\ No newline at end of file
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/processor/EndTransactionProcessorTest.java b/broker/src/test/java/org/apache/rocketmq/broker/processor/EndTransactionProcessorTest.java
index 82c0a7d..6da0798 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/processor/EndTransactionProcessorTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/processor/EndTransactionProcessorTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.rocketmq.broker.processor;
+import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.transaction.OperationResult;
@@ -28,10 +29,11 @@ import org.apache.rocketmq.common.protocol.RequestCode;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.header.EndTransactionRequestHeader;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
-import org.apache.rocketmq.remoting.exception.RemotingCommandException;
-import org.apache.rocketmq.remoting.netty.NettyChannelHandlerContextImpl;
import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.ServerConfig;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+import org.apache.rocketmq.remoting.netty.CodecHelper;
+import org.apache.rocketmq.remoting.netty.NettyChannelHandlerContextImpl;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.store.AppendMessageResult;
import org.apache.rocketmq.store.AppendMessageStatus;
@@ -49,6 +51,7 @@ import org.mockito.junit.MockitoJUnitRunner;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@RunWith(MockitoJUnitRunner.class)
@@ -57,7 +60,10 @@ public class EndTransactionProcessorTest {
private EndTransactionProcessor endTransactionProcessor;
@Mock
- private ChannelHandlerContext handlerContext;
+ private NettyChannelHandlerContextImpl handlerContext;
+
+ @Mock
+ private ChannelHandlerContext channelHandlerContext;
@Spy
private BrokerController
@@ -74,10 +80,13 @@ public class EndTransactionProcessorTest {
public void init() {
brokerController.setMessageStore(messageStore);
brokerController.setTransactionalMessageService(transactionMsgService);
+ Channel mockChannel = mock(Channel.class);
+ when(handlerContext.getChannelHandlerContext()).thenReturn(channelHandlerContext);
+ when(channelHandlerContext.channel()).thenReturn(mockChannel);
endTransactionProcessor = new EndTransactionProcessor(brokerController);
}
- private OperationResult createResponse(int status) {
+ private OperationResult createResponse(int status){
OperationResult response = new OperationResult();
response.setPrepareMessage(createDefaultMessageExt());
response.setResponseCode(status);
@@ -91,9 +100,7 @@ public class EndTransactionProcessorTest {
when(messageStore.putMessage(any(MessageExtBrokerInner.class))).thenReturn(new PutMessageResult
(PutMessageStatus.PUT_OK, new AppendMessageResult(AppendMessageStatus.PUT_OK)));
RemotingCommand request = createEndTransactionMsgCommand(MessageSysFlag.TRANSACTION_COMMIT_TYPE, false);
-
- NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
- RemotingCommand response = endTransactionProcessor.processRequest(nettyChannelHandlerContext, request);
+ RemotingCommand response = endTransactionProcessor.processRequest(handlerContext, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
@@ -103,16 +110,14 @@ public class EndTransactionProcessorTest {
when(messageStore.putMessage(any(MessageExtBrokerInner.class))).thenReturn(new PutMessageResult
(PutMessageStatus.PUT_OK, new AppendMessageResult(AppendMessageStatus.PUT_OK)));
RemotingCommand request = createEndTransactionMsgCommand(MessageSysFlag.TRANSACTION_COMMIT_TYPE, true);
- NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
- RemotingCommand response = endTransactionProcessor.processRequest(nettyChannelHandlerContext, request);
+ RemotingCommand response = endTransactionProcessor.processRequest(handlerContext, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
@Test
public void testProcessRequest_NotType() throws RemotingCommandException {
RemotingCommand request = createEndTransactionMsgCommand(MessageSysFlag.TRANSACTION_NOT_TYPE, true);
- NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
- RemotingCommand response = endTransactionProcessor.processRequest(nettyChannelHandlerContext, request);
+ RemotingCommand response = endTransactionProcessor.processRequest(handlerContext, request);
assertThat(response).isNull();
}
@@ -120,8 +125,7 @@ public class EndTransactionProcessorTest {
public void testProcessRequest_RollBack() throws RemotingCommandException {
when(transactionMsgService.rollbackMessage(any(EndTransactionRequestHeader.class))).thenReturn(createResponse(ResponseCode.SUCCESS));
RemotingCommand request = createEndTransactionMsgCommand(MessageSysFlag.TRANSACTION_ROLLBACK_TYPE, true);
- NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
- RemotingCommand response = endTransactionProcessor.processRequest(nettyChannelHandlerContext, request);
+ RemotingCommand response = endTransactionProcessor.processRequest(handlerContext, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
@@ -152,7 +156,7 @@ public class EndTransactionProcessorTest {
private RemotingCommand createEndTransactionMsgCommand(int status, boolean isCheckMsg) {
EndTransactionRequestHeader header = createEndTransactionRequestHeader(status, isCheckMsg);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.END_TRANSACTION, header);
-// request.makeCustomHeaderToNet();
+ CodecHelper.makeCustomHeaderToNet(request);
return request;
}
}
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/processor/PullMessageProcessorTest.java b/broker/src/test/java/org/apache/rocketmq/broker/processor/PullMessageProcessorTest.java
index a5a32c4..fd99c08 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/processor/PullMessageProcessorTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/processor/PullMessageProcessorTest.java
@@ -38,11 +38,12 @@ import org.apache.rocketmq.common.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.common.protocol.heartbeat.ConsumerData;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
+import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.RemotingChannel;
+import org.apache.rocketmq.remoting.ServerConfig;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+import org.apache.rocketmq.remoting.netty.CodecHelper;
import org.apache.rocketmq.remoting.netty.NettyChannelHandlerContextImpl;
-import org.apache.rocketmq.remoting.ClientConfig;
-import org.apache.rocketmq.remoting.ServerConfig;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.store.GetMessageResult;
import org.apache.rocketmq.store.GetMessageStatus;
@@ -69,9 +70,11 @@ public class PullMessageProcessorTest {
@Spy
private BrokerController brokerController = new BrokerController(new BrokerConfig(), new ServerConfig(), new ClientConfig(), new MessageStoreConfig());
@Mock
- private ChannelHandlerContext handlerContext;
+ private NettyChannelHandlerContextImpl handlerContext;
+
@Mock
- private Channel channel;
+ private ChannelHandlerContext channelHandlerContext;
+
@Mock
private MessageStore messageStore;
private ClientChannelInfo clientChannelInfo;
@@ -82,11 +85,13 @@ public class PullMessageProcessorTest {
public void init() {
brokerController.setMessageStore(messageStore);
pullMessageProcessor = new PullMessageProcessor(brokerController);
- RemotingChannel mockChannel = mock(RemotingChannel.class);
+ Channel mockChannel = mock(Channel.class);
+ RemotingChannel remotingChannel = mock(RemotingChannel.class);
when(mockChannel.remoteAddress()).thenReturn(new InetSocketAddress(1024));
-// when(handlerContext.channel()).thenReturn(mockChannel);
+ when(handlerContext.getChannelHandlerContext()).thenReturn(channelHandlerContext);
+ when(channelHandlerContext.channel()).thenReturn(mockChannel);
brokerController.getTopicConfigManager().getTopicConfigTable().put(topic, new TopicConfig());
- clientChannelInfo = new ClientChannelInfo(mockChannel);
+ clientChannelInfo = new ClientChannelInfo(remotingChannel);
ConsumerData consumerData = createConsumerData(group, topic);
brokerController.getConsumerManager().registerConsumer(
consumerData.getGroupName(),
@@ -102,8 +107,7 @@ public class PullMessageProcessorTest {
public void testProcessRequest_TopicNotExist() throws RemotingCommandException {
brokerController.getTopicConfigManager().getTopicConfigTable().remove(topic);
final RemotingCommand request = createPullMsgCommand(RequestCode.PULL_MESSAGE);
- NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
- RemotingCommand response = pullMessageProcessor.processRequest(nettyChannelHandlerContext, request);
+ RemotingCommand response = pullMessageProcessor.processRequest(handlerContext, request);
assertThat(response).isNotNull();
assertThat(response.getCode()).isEqualTo(ResponseCode.TOPIC_NOT_EXIST);
assertThat(response.getRemark()).contains("topic[" + topic + "] not exist");
@@ -113,9 +117,7 @@ public class PullMessageProcessorTest {
public void testProcessRequest_SubNotExist() throws RemotingCommandException {
brokerController.getConsumerManager().unregisterConsumer(group, clientChannelInfo, false);
final RemotingCommand request = createPullMsgCommand(RequestCode.PULL_MESSAGE);
-
- NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
- RemotingCommand response = pullMessageProcessor.processRequest(nettyChannelHandlerContext, request);
+ RemotingCommand response = pullMessageProcessor.processRequest(handlerContext, request);
assertThat(response).isNotNull();
assertThat(response.getCode()).isEqualTo(ResponseCode.SUBSCRIPTION_NOT_EXIST);
assertThat(response.getRemark()).contains("consumer's group info not exist");
@@ -125,8 +127,7 @@ public class PullMessageProcessorTest {
public void testProcessRequest_SubNotLatest() throws RemotingCommandException {
final RemotingCommand request = createPullMsgCommand(RequestCode.PULL_MESSAGE);
request.addExtField("subVersion", String.valueOf(101));
- NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
- RemotingCommand response = pullMessageProcessor.processRequest(nettyChannelHandlerContext, request);
+ RemotingCommand response = pullMessageProcessor.processRequest(handlerContext, request);
assertThat(response).isNotNull();
assertThat(response.getCode()).isEqualTo(ResponseCode.SUBSCRIPTION_NOT_LATEST);
assertThat(response.getRemark()).contains("subscription not latest");
@@ -138,9 +139,7 @@ public class PullMessageProcessorTest {
when(messageStore.getMessage(anyString(), anyString(), anyInt(), anyLong(), anyInt(), any(ExpressionMessageFilter.class))).thenReturn(getMessageResult);
final RemotingCommand request = createPullMsgCommand(RequestCode.PULL_MESSAGE);
-
- NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
- RemotingCommand response = pullMessageProcessor.processRequest(nettyChannelHandlerContext, request);
+ RemotingCommand response = pullMessageProcessor.processRequest(handlerContext, request);
assertThat(response).isNotNull();
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
@@ -169,9 +168,7 @@ public class PullMessageProcessorTest {
consumeMessageHookList.add(consumeMessageHook);
pullMessageProcessor.registerConsumeMessageHook(consumeMessageHookList);
final RemotingCommand request = createPullMsgCommand(RequestCode.PULL_MESSAGE);
-
- NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
- RemotingCommand response = pullMessageProcessor.processRequest(nettyChannelHandlerContext, request);
+ RemotingCommand response = pullMessageProcessor.processRequest(handlerContext, request);
assertThat(response).isNotNull();
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
assertThat(messageContext[0]).isNotNull();
@@ -187,9 +184,7 @@ public class PullMessageProcessorTest {
when(messageStore.getMessage(anyString(), anyString(), anyInt(), anyLong(), anyInt(), any(ExpressionMessageFilter.class))).thenReturn(getMessageResult);
final RemotingCommand request = createPullMsgCommand(RequestCode.PULL_MESSAGE);
-
- NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
- RemotingCommand response = pullMessageProcessor.processRequest(nettyChannelHandlerContext, request);
+ RemotingCommand response = pullMessageProcessor.processRequest(handlerContext, request);
assertThat(response).isNotNull();
assertThat(response.getCode()).isEqualTo(ResponseCode.PULL_RETRY_IMMEDIATELY);
}
@@ -201,9 +196,7 @@ public class PullMessageProcessorTest {
when(messageStore.getMessage(anyString(), anyString(), anyInt(), anyLong(), anyInt(), any(ExpressionMessageFilter.class))).thenReturn(getMessageResult);
final RemotingCommand request = createPullMsgCommand(RequestCode.PULL_MESSAGE);
-
- NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
- RemotingCommand response = pullMessageProcessor.processRequest(nettyChannelHandlerContext, request);
+ RemotingCommand response = pullMessageProcessor.processRequest(handlerContext, request);
assertThat(response).isNotNull();
assertThat(response.getCode()).isEqualTo(ResponseCode.PULL_OFFSET_MOVED);
}
@@ -220,7 +213,7 @@ public class PullMessageProcessorTest {
requestHeader.setSysFlag(0);
requestHeader.setSubVersion(100L);
RemotingCommand request = RemotingCommand.createRequestCommand(requestCode, requestHeader);
-// request.makeCustomHeaderToNet();
+ CodecHelper.makeCustomHeaderToNet(request);
return request;
}
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/processor/SendMessageProcessorTest.java b/broker/src/test/java/org/apache/rocketmq/broker/processor/SendMessageProcessorTest.java
index ac8a106..c6a5b51 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/processor/SendMessageProcessorTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/processor/SendMessageProcessorTest.java
@@ -18,6 +18,10 @@ package org.apache.rocketmq.broker.processor;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.mqtrace.SendMessageContext;
import org.apache.rocketmq.broker.mqtrace.SendMessageHook;
@@ -32,10 +36,11 @@ import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.header.ConsumerSendMsgBackRequestHeader;
import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
-import org.apache.rocketmq.remoting.exception.RemotingCommandException;
-import org.apache.rocketmq.remoting.netty.NettyChannelHandlerContextImpl;
import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.ServerConfig;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+import org.apache.rocketmq.remoting.netty.CodecHelper;
+import org.apache.rocketmq.remoting.netty.NettyChannelHandlerContextImpl;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.store.AppendMessageResult;
import org.apache.rocketmq.store.AppendMessageStatus;
@@ -53,11 +58,6 @@ import org.mockito.invocation.InvocationOnMock;
import org.mockito.junit.MockitoJUnitRunner;
import org.mockito.stubbing.Answer;
-import java.net.InetSocketAddress;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
@@ -69,7 +69,10 @@ import static org.mockito.Mockito.when;
public class SendMessageProcessorTest {
private SendMessageProcessor sendMessageProcessor;
@Mock
- private ChannelHandlerContext handlerContext;
+ private NettyChannelHandlerContextImpl handlerContext;
+
+ @Mock
+ private ChannelHandlerContext channelHandlerContext;
@Spy
private BrokerController brokerController = new BrokerController(new BrokerConfig(), new ServerConfig(), new ClientConfig(), new MessageStoreConfig());
@Mock
@@ -87,7 +90,8 @@ public class SendMessageProcessorTest {
when(messageStore.now()).thenReturn(System.currentTimeMillis());
Channel mockChannel = mock(Channel.class);
when(mockChannel.remoteAddress()).thenReturn(new InetSocketAddress(1024));
- when(handlerContext.channel()).thenReturn(mockChannel);
+ when(handlerContext.getChannelHandlerContext()).thenReturn(channelHandlerContext);
+ when(channelHandlerContext.channel()).thenReturn(mockChannel);
when(messageStore.lookMessageByOffset(anyLong())).thenReturn(new MessageExt());
sendMessageProcessor = new SendMessageProcessor(brokerController);
}
@@ -182,9 +186,7 @@ public class SendMessageProcessorTest {
final RemotingCommand request = createSendMsgBackCommand(RequestCode.CONSUMER_SEND_MSG_BACK);
sendMessageProcessor = new SendMessageProcessor(brokerController);
-
- NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
- RemotingCommand response = sendMessageProcessor.processRequest(nettyChannelHandlerContext, request);
+ final RemotingCommand response = sendMessageProcessor.processRequest(handlerContext, request);
assertThat(response).isNotNull();
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
}
@@ -201,11 +203,8 @@ public class SendMessageProcessorTest {
response[0] = invocation.getArgument(0);
return null;
}
- }).when(handlerContext).writeAndFlush(any(Object.class));
-
- NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
- RemotingCommand responseToReturn = sendMessageProcessor.processRequest(nettyChannelHandlerContext, request);
-
+ }).when(channelHandlerContext).writeAndFlush(any(Object.class));
+ RemotingCommand responseToReturn = sendMessageProcessor.processRequest(handlerContext, request);
if (responseToReturn != null) {
assertThat(response[0]).isNull();
response[0] = responseToReturn;
@@ -213,7 +212,6 @@ public class SendMessageProcessorTest {
assertThat(response[0].getCode()).isEqualTo(ResponseCode.SUCCESS);
}
-
private RemotingCommand createSendTransactionMsgCommand(int requestCode) {
SendMessageRequestHeader header = createSendMsgRequestHeader();
int sysFlag = header.getSysFlag();
@@ -224,7 +222,7 @@ public class SendMessageProcessorTest {
header.setSysFlag(sysFlag);
RemotingCommand request = RemotingCommand.createRequestCommand(requestCode, header);
request.setBody(new byte[] {'a'});
-// request.makeCustomHeaderToNet();
+ CodecHelper.makeCustomHeaderToNet(request);
return request;
}
@@ -247,7 +245,7 @@ public class SendMessageProcessorTest {
RemotingCommand request = RemotingCommand.createRequestCommand(requestCode, requestHeader);
request.setBody(new byte[] {'a'});
-// request.makeCustomHeaderToNet();
+ CodecHelper.makeCustomHeaderToNet(request);
return request;
}
@@ -260,7 +258,7 @@ public class SendMessageProcessorTest {
requestHeader.setOffset(123L);
RemotingCommand request = RemotingCommand.createRequestCommand(requestCode, requestHeader);
-// request.makeCustomHeaderToNet();
+ CodecHelper.makeCustomHeaderToNet(request);
return request;
}
@@ -273,10 +271,8 @@ public class SendMessageProcessorTest {
response[0] = invocation.getArgument(0);
return null;
}
- }).when(handlerContext).writeAndFlush(any(Object.class));
-
- NettyChannelHandlerContextImpl nettyChannelHandlerContext = new NettyChannelHandlerContextImpl(handlerContext);
- RemotingCommand responseToReturn = sendMessageProcessor.processRequest(nettyChannelHandlerContext, request);
+ }).when(channelHandlerContext).writeAndFlush(any(Object.class));
+ RemotingCommand responseToReturn = sendMessageProcessor.processRequest(handlerContext, request);
if (responseToReturn != null) {
assertThat(response[0]).isNull();
response[0] = responseToReturn;
diff --git a/client/src/main/java/org/apache/rocketmq/client/consumer/store/RemoteBrokerOffsetStore.java b/client/src/main/java/org/apache/rocketmq/client/consumer/store/RemoteBrokerOffsetStore.java
index f3e2f4e..9087505 100644
--- a/client/src/main/java/org/apache/rocketmq/client/consumer/store/RemoteBrokerOffsetStore.java
+++ b/client/src/main/java/org/apache/rocketmq/client/consumer/store/RemoteBrokerOffsetStore.java
@@ -16,7 +16,6 @@
*/
package org.apache.rocketmq.client.consumer.store;
-import com.sun.org.apache.regexp.internal.RE;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
@@ -31,10 +30,10 @@ import org.apache.rocketmq.client.impl.factory.MQClientInstance;
import org.apache.rocketmq.client.log.ClientLogger;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.UtilAll;
-import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.header.QueryConsumerOffsetRequestHeader;
import org.apache.rocketmq.common.protocol.header.UpdateConsumerOffsetRequestHeader;
+import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.remoting.exception.RemotingException;
/**
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
index 98a626c..5b4d318 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
@@ -135,8 +135,8 @@ import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.InvokeCallback;
-import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.RemotingClient;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.exception.RemotingConnectException;
@@ -144,12 +144,10 @@ import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
-import org.apache.rocketmq.remoting.ClientConfig;
-import org.apache.rocketmq.remoting.interceptor.Interceptor;
import org.apache.rocketmq.remoting.interceptor.InterceptorGroup;
import org.apache.rocketmq.remoting.netty.ResponseFuture;
-import org.apache.rocketmq.remoting.serialize.LanguageCode;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.serialize.LanguageCode;
import org.apache.rocketmq.remoting.serialize.RemotingSerializable;
import org.apache.rocketmq.remoting.transport.rocketmq.NettyRemotingClient;
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java b/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java
index 769164b..321f804 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java
@@ -62,7 +62,6 @@ import org.apache.rocketmq.common.ServiceState;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.constant.PermName;
import org.apache.rocketmq.common.filter.ExpressionType;
-import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
@@ -75,11 +74,10 @@ import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.common.protocol.route.QueueData;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
-import org.apache.rocketmq.remoting.RPCHook;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
-import org.apache.rocketmq.remoting.ClientConfig;
-import org.apache.rocketmq.remoting.interceptor.Interceptor;
import org.apache.rocketmq.remoting.interceptor.InterceptorGroup;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
diff --git a/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumerTest.java b/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumerTest.java
index ff2fb78..7224794 100644
--- a/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumerTest.java
+++ b/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumerTest.java
@@ -147,7 +147,6 @@ public class DefaultMQPushConsumerTest {
});
doReturn(new FindBrokerResult("127.0.0.1:10911", false)).when(mQClientFactory).findBrokerAddressInSubscribe(anyString(), anyLong(), anyBoolean());
- doReturn(Collections.singletonList(mQClientFactory.getClientId())).when(mQClientFactory).findConsumerIdList(anyString(), anyString());
Set<MessageQueue> messageQueueSet = new HashSet<MessageQueue>();
messageQueueSet.add(createPullRequest().getMessageQueue());
pushConsumer.getDefaultMQPushConsumerImpl().updateTopicSubscribeInfo(topic, messageQueueSet);
diff --git a/common/src/main/java/org/apache/rocketmq/common/UtilAll.java b/common/src/main/java/org/apache/rocketmq/common/UtilAll.java
index 3b1866e..b1a8d41 100644
--- a/common/src/main/java/org/apache/rocketmq/common/UtilAll.java
+++ b/common/src/main/java/org/apache/rocketmq/common/UtilAll.java
@@ -16,8 +16,6 @@
*/
package org.apache.rocketmq.common;
-import com.alibaba.fastjson.parser.ParserConfig;
-import com.alibaba.fastjson.util.TypeUtils;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
@@ -31,12 +29,10 @@ import java.net.NetworkInterface;
import java.text.NumberFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
-import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.Enumeration;
import java.util.Iterator;
-import java.util.List;
import java.util.Map;
import java.util.zip.CRC32;
import java.util.zip.DeflaterOutputStream;
diff --git a/common/src/main/java/org/apache/rocketmq/common/flowcontrol/FlowControlConfig.java b/common/src/main/java/org/apache/rocketmq/common/flowcontrol/FlowControlConfig.java
index 58bbade..57e9aa8 100644
--- a/common/src/main/java/org/apache/rocketmq/common/flowcontrol/FlowControlConfig.java
+++ b/common/src/main/java/org/apache/rocketmq/common/flowcontrol/FlowControlConfig.java
@@ -37,14 +37,11 @@ public class FlowControlConfig {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);
private String flowControlFileHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY, System.getenv(MixAll.ROCKETMQ_HOME_ENV));
+
private static final String DEFAULT_FLOW_CONTROL_FILE = "conf/flowControl.yml";
private String flowControlFileName = System.getProperty("rocketmq.flow.control.file", DEFAULT_FLOW_CONTROL_FILE);
- private List<FlowControlRule> rules;
-
- public static final String defaultResourceName = "overallFlowControl";
-
private Map<String/*server name*/, Map<String/*flowControlType*/, List<FlowControlRule>>> plainFlowControlRules;
public FlowControlConfig() {
diff --git a/common/src/main/java/org/apache/rocketmq/common/flowcontrol/FlowControlRule.java b/common/src/main/java/org/apache/rocketmq/common/flowcontrol/FlowControlRule.java
index 8735ff9..c286c87 100644
--- a/common/src/main/java/org/apache/rocketmq/common/flowcontrol/FlowControlRule.java
+++ b/common/src/main/java/org/apache/rocketmq/common/flowcontrol/FlowControlRule.java
@@ -16,8 +16,6 @@
*/
package org.apache.rocketmq.common.flowcontrol;
-import org.apache.rocketmq.remoting.interceptor.Interceptor;
-
public class FlowControlRule {
private String flowControlResourceName;
private Integer flowControlGrade;
diff --git a/example/src/main/java/org/apache/rocketmq/example/benchmark/Producer.java b/example/src/main/java/org/apache/rocketmq/example/benchmark/Producer.java
index ce2b83f..1ba42ea 100644
--- a/example/src/main/java/org/apache/rocketmq/example/benchmark/Producer.java
+++ b/example/src/main/java/org/apache/rocketmq/example/benchmark/Producer.java
@@ -24,7 +24,6 @@ import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;
-
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
@@ -33,8 +32,8 @@ import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.log.ClientLogger;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
-import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.srvutil.ServerUtil;
diff --git a/example/src/main/java/org/apache/rocketmq/example/quickstart/Producer.java b/example/src/main/java/org/apache/rocketmq/example/quickstart/Producer.java
index b2609c8..b87aada 100644
--- a/example/src/main/java/org/apache/rocketmq/example/quickstart/Producer.java
+++ b/example/src/main/java/org/apache/rocketmq/example/quickstart/Producer.java
@@ -20,7 +20,6 @@ import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
-import org.apache.rocketmq.common.utils.ThreadUtils;
import org.apache.rocketmq.remoting.common.RemotingHelper;
/**
diff --git a/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessor.java b/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessor.java
index a7da99f..e3136fa 100644
--- a/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessor.java
+++ b/namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessor.java
@@ -295,10 +295,9 @@ public class DefaultRequestProcessor implements RequestProcessor {
public RemotingCommand registerBroker(ChannelHandlerContext ctx,
RemotingCommand request) throws RemotingCommandException {
final RemotingCommand response = RemotingCommand.createResponseCommand(RegisterBrokerResponseHeader.class);
- final RegisterBrokerResponseHeader responseHeader = (RegisterBrokerResponseHeader) response.readCustomHeader();
final RegisterBrokerRequestHeader requestHeader =
(RegisterBrokerRequestHeader) request.decodeCommandCustomHeader(RegisterBrokerRequestHeader.class);
- log.info("requestHeader： " + requestHeader);
+ final RegisterBrokerResponseHeader responseHeader = (RegisterBrokerResponseHeader) response.readCustomHeader();
if (!checksum(ctx, request, requestHeader)) {
response.setCode(ResponseCode.SYSTEM_ERROR);
response.setRemark("crc32 not match");
diff --git a/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/BrokerHousekeepingService.java b/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/BrokerHousekeepingService.java
index e0cdd95..77f387c 100644
--- a/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/BrokerHousekeepingService.java
+++ b/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/BrokerHousekeepingService.java
@@ -39,22 +39,28 @@ public class BrokerHousekeepingService implements ChannelEventListener {
@Override
public void onChannelClose(String remoteAddr, RemotingChannel remotingChannel) {
- NettyChannelImpl nettyChannel = (NettyChannelImpl) remotingChannel;
- Channel channel = nettyChannel.getChannel();
- this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
+ if (remotingChannel != null) {
+ NettyChannelImpl nettyChannel = (NettyChannelImpl) remotingChannel;
+ Channel channel = nettyChannel.getChannel();
+ this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
+ }
}
@Override
public void onChannelException(String remoteAddr, RemotingChannel remotingChannel) {
- NettyChannelImpl nettyChannel = (NettyChannelImpl) remotingChannel;
- Channel channel = nettyChannel.getChannel();
- this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
+ if (remotingChannel != null) {
+ NettyChannelImpl nettyChannel = (NettyChannelImpl) remotingChannel;
+ Channel channel = nettyChannel.getChannel();
+ this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
+ }
}
@Override
public void onChannelIdle(String remoteAddr, RemotingChannel remotingChannel) {
- NettyChannelImpl nettyChannel = (NettyChannelImpl) remotingChannel;
- Channel channel = nettyChannel.getChannel();
- this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
+ if (remotingChannel != null) {
+ NettyChannelImpl nettyChannel = (NettyChannelImpl) remotingChannel;
+ Channel channel = nettyChannel.getChannel();
+ this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
+ }
}
}
diff --git a/namesrv/src/test/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessorTest.java b/namesrv/src/test/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessorTest.java
index 8cd75f8..a2609ee 100644
--- a/namesrv/src/test/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessorTest.java
+++ b/namesrv/src/test/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessorTest.java
@@ -25,7 +25,6 @@ import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.rocketmq.common.TopicConfig;
-import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.common.namesrv.NamesrvConfig;
import org.apache.rocketmq.common.namesrv.RegisterBrokerResult;
import org.apache.rocketmq.common.protocol.RequestCode;
@@ -37,11 +36,12 @@ import org.apache.rocketmq.common.protocol.header.namesrv.GetKVConfigResponseHea
import org.apache.rocketmq.common.protocol.header.namesrv.PutKVConfigRequestHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.RegisterBrokerRequestHeader;
import org.apache.rocketmq.common.protocol.route.BrokerData;
+import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.namesrv.NamesrvController;
import org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager;
+import org.apache.rocketmq.remoting.ServerConfig;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.netty.NettyChannelHandlerContextImpl;
-import org.apache.rocketmq.remoting.ServerConfig;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.assertj.core.util.Maps;
import org.junit.Before;
@@ -91,8 +91,10 @@ public class DefaultRequestProcessorTest {
request.addExtField("namespace", "namespace");
request.addExtField("key", "key");
request.addExtField("value", "value");
-
- RemotingCommand response = defaultRequestProcessor.processRequest(null, request);
+ NettyChannelHandlerContextImpl remotingChannel = mock(NettyChannelHandlerContextImpl.class);
+ ChannelHandlerContext channelHandlerContext = mock(ChannelHandlerContext.class);
+ when(remotingChannel.getChannelHandlerContext()).thenReturn(channelHandlerContext);
+ RemotingCommand response = defaultRequestProcessor.processRequest(remotingChannel, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
assertThat(response.getRemark()).isNull();
@@ -111,7 +113,10 @@ public class DefaultRequestProcessorTest {
request.addExtField("namespace", "namespace");
request.addExtField("key", "key");
- RemotingCommand response = defaultRequestProcessor.processRequest(null, request);
+ NettyChannelHandlerContextImpl remotingChannel = mock(NettyChannelHandlerContextImpl.class);
+ ChannelHandlerContext channelHandlerContext = mock(ChannelHandlerContext.class);
+ when(remotingChannel.getChannelHandlerContext()).thenReturn(channelHandlerContext);
+ RemotingCommand response = defaultRequestProcessor.processRequest(remotingChannel, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
assertThat(response.getRemark()).isNull();
@@ -130,7 +135,10 @@ public class DefaultRequestProcessorTest {
request.addExtField("namespace", "namespace");
request.addExtField("key", "key");
- RemotingCommand response = defaultRequestProcessor.processRequest(null, request);
+ NettyChannelHandlerContextImpl remotingChannel = mock(NettyChannelHandlerContextImpl.class);
+ ChannelHandlerContext channelHandlerContext = mock(ChannelHandlerContext.class);
+ when(remotingChannel.getChannelHandlerContext()).thenReturn(channelHandlerContext);
+ RemotingCommand response = defaultRequestProcessor.processRequest(remotingChannel, request);
assertThat(response.getCode()).isEqualTo(ResponseCode.QUERY_NOT_FOUND);
assertThat(response.getRemark()).isEqualTo("No config item, Namespace: namespace Key: key");
@@ -151,8 +159,11 @@ public class DefaultRequestProcessorTest {
request.addExtField("namespace", "namespace");
request.addExtField("key", "key");
- RemotingCommand response = defaultRequestProcessor.processRequest(null, request);
-
+ NettyChannelHandlerContextImpl remotingChannel = mock(NettyChannelHandlerContextImpl.class);
+ ChannelHandlerContext channelHandlerContext = mock(ChannelHandlerContext.class);
+ when(remotingChannel.getChannelHandlerContext()).thenReturn(channelHandlerContext);
+ RemotingCommand response = defaultRequestProcessor.processRequest(remotingChannel, request);
+
assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
assertThat(response.getRemark()).isNull();
diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/RemotingServer.java b/remoting/src/main/java/org/apache/rocketmq/remoting/RemotingServer.java
index c0b8558..a042065 100644
--- a/remoting/src/main/java/org/apache/rocketmq/remoting/RemotingServer.java
+++ b/remoting/src/main/java/org/apache/rocketmq/remoting/RemotingServer.java
@@ -21,7 +21,6 @@ import org.apache.rocketmq.remoting.common.Pair;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
-import org.apache.rocketmq.remoting.interceptor.InterceptorGroup;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
public interface RemotingServer extends RemotingService {
diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyChannelHandlerContextImpl.java b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyChannelHandlerContextImpl.java
index a00ef05..f1746fa 100644
--- a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyChannelHandlerContextImpl.java
+++ b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyChannelHandlerContextImpl.java
@@ -99,7 +99,7 @@ public class NettyChannelHandlerContextImpl implements RemotingChannel {
final NettyChannelHandlerContextImpl that = (NettyChannelHandlerContextImpl) o;
- return channelHandlerContext != null ? channelHandlerContext.equals(that.channelHandlerContext) : that.channelHandlerContext == null;
+ return channelHandlerContext.channel() != null ? channelHandlerContext.channel().equals(that.channelHandlerContext.channel()) : that.channelHandlerContext.channel() == null;
}
diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyChannelImpl.java b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyChannelImpl.java
index fe33a52..90f2b31 100644
--- a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyChannelImpl.java
+++ b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyChannelImpl.java
@@ -32,7 +32,7 @@ public class NettyChannelImpl implements RemotingChannel {
private static final InternalLogger log = InternalLoggerFactory.getLogger(ROCKETMQ_REMOTING);
- private final io.netty.channel.Channel channel;
+ private final Channel channel;
public NettyChannelImpl(Channel channel) {
this.channel = channel;
diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2ClientImpl.java b/remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2ClientImpl.java
index 4aa5d68..761d2bf 100644
--- a/remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2ClientImpl.java
+++ b/remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2ClientImpl.java
@@ -17,7 +17,6 @@
package org.apache.rocketmq.remoting.transport.http2;
import io.netty.bootstrap.Bootstrap;
-import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
@@ -41,26 +40,21 @@ import javax.net.ssl.SSLException;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.ChannelEventListener;
+import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.InvokeCallback;
-import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.RemotingClient;
-import org.apache.rocketmq.remoting.common.Pair;
+import org.apache.rocketmq.remoting.RequestProcessor;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingConnectException;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
-import org.apache.rocketmq.remoting.ClientConfig;
-import org.apache.rocketmq.remoting.interceptor.Interceptor;
import org.apache.rocketmq.remoting.interceptor.InterceptorGroup;
-import org.apache.rocketmq.remoting.interceptor.InterceptorInvoker;
-import org.apache.rocketmq.remoting.netty.NettyChannelImpl;
-import org.apache.rocketmq.remoting.transport.rocketmq.NettyDecoder;
-import org.apache.rocketmq.remoting.transport.rocketmq.NettyEncoder;
-import org.apache.rocketmq.remoting.RequestProcessor;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.transport.NettyRemotingClientAbstract;
+import org.apache.rocketmq.remoting.transport.rocketmq.NettyDecoder;
+import org.apache.rocketmq.remoting.transport.rocketmq.NettyEncoder;
import org.apache.rocketmq.remoting.util.ThreadUtils;
public class Http2ClientImpl extends NettyRemotingClientAbstract implements RemotingClient {
@@ -248,7 +242,7 @@ public class Http2ClientImpl extends NettyRemotingClientAbstract implements Remo
@Override
public void registerProcessor(int requestCode, RequestProcessor processor, ExecutorService executor) {
- executor = (executor == null ? this.publicExecutor : executor);
+ executor = executor == null ? this.publicExecutor : executor;
registerNettyProcessor(requestCode, processor, executor);
}
diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2ServerImpl.java b/remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2ServerImpl.java
index d86e4ca..7791392 100644
--- a/remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2ServerImpl.java
+++ b/remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2ServerImpl.java
@@ -48,24 +48,22 @@ import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.ChannelEventListener;
import org.apache.rocketmq.remoting.InvokeCallback;
-import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.RemotingServer;
+import org.apache.rocketmq.remoting.RequestProcessor;
+import org.apache.rocketmq.remoting.ServerConfig;
import org.apache.rocketmq.remoting.common.Pair;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
-import org.apache.rocketmq.remoting.interceptor.Interceptor;
import org.apache.rocketmq.remoting.interceptor.InterceptorGroup;
import org.apache.rocketmq.remoting.netty.ChannelStatisticsHandler;
import org.apache.rocketmq.remoting.netty.NettyChannelImpl;
-import org.apache.rocketmq.remoting.transport.rocketmq.NettyDecoder;
-import org.apache.rocketmq.remoting.transport.rocketmq.NettyEncoder;
-import org.apache.rocketmq.remoting.RequestProcessor;
-import org.apache.rocketmq.remoting.ServerConfig;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.transport.NettyRemotingServerAbstract;
+import org.apache.rocketmq.remoting.transport.rocketmq.NettyDecoder;
+import org.apache.rocketmq.remoting.transport.rocketmq.NettyEncoder;
import org.apache.rocketmq.remoting.util.JvmUtils;
import org.apache.rocketmq.remoting.util.ThreadUtils;
@@ -139,7 +137,7 @@ public class Http2ServerImpl extends NettyRemotingServerAbstract implements Remo
@Override
public void registerProcessor(int requestCode, RequestProcessor processor, ExecutorService executor) {
- executor = (executor == null ? this.publicExecutor : executor);
+ executor = executor == null ? this.publicExecutor : executor;
registerNettyProcessor(requestCode, processor, executor);
}
@@ -204,25 +202,23 @@ public class Http2ServerImpl extends NettyRemotingServerAbstract implements Remo
public void start() {
super.start();
final ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
- this.serverBootstrap.group(this.bossGroup, this.ioGroup).
- channel(socketChannelClass).childHandler(new ChannelInitializer<SocketChannel>() {
- @Override
- public void initChannel(SocketChannel ch) throws Exception {
- channels.add(ch);
-
- ChannelPipeline cp = ch.pipeline();
- cp.addLast(ChannelStatisticsHandler.NAME, new ChannelStatisticsHandler(channels));
- cp.addLast(workerGroup,
- Http2Handler.newHandler(true),
- new NettyEncoder(),
- new NettyDecoder(),
- new IdleStateHandler(serverConfig.getConnectionChannelReaderIdleSeconds(),
- serverConfig.getConnectionChannelWriterIdleSeconds(),
- serverConfig.getServerChannelMaxIdleTimeSeconds()),
- new NettyConnectManageHandler(),
- new NettyServerHandler());
- }
- });
+ this.serverBootstrap.group(this.bossGroup, this.ioGroup).channel(socketChannelClass).childHandler(new ChannelInitializer<SocketChannel>() {
+ @Override
+ public void initChannel(SocketChannel ch) {
+ channels.add(ch);
+ ChannelPipeline cp = ch.pipeline();
+ cp.addLast(ChannelStatisticsHandler.NAME, new ChannelStatisticsHandler(channels));
+ cp.addLast(workerGroup,
+ Http2Handler.newHandler(true),
+ new NettyEncoder(),
+ new NettyDecoder(),
+ new IdleStateHandler(serverConfig.getConnectionChannelReaderIdleSeconds(),
+ serverConfig.getConnectionChannelWriterIdleSeconds(),
+ serverConfig.getServerChannelMaxIdleTimeSeconds()),
+ new NettyConnectManageHandler(),
+ new NettyServerHandler());
+ }
+ });
applyOptions(serverBootstrap);
ChannelFuture channelFuture = this.serverBootstrap.bind(this.port).syncUninterruptibly();
diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/transport/rocketmq/NettyRemotingClient.java b/remoting/src/main/java/org/apache/rocketmq/remoting/transport/rocketmq/NettyRemotingClient.java
index 4658147..b161638 100644
--- a/remoting/src/main/java/org/apache/rocketmq/remoting/transport/rocketmq/NettyRemotingClient.java
+++ b/remoting/src/main/java/org/apache/rocketmq/remoting/transport/rocketmq/NettyRemotingClient.java
@@ -36,10 +36,11 @@ import java.util.concurrent.ExecutorService;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.ChannelEventListener;
+import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.InvokeCallback;
-import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.RemotingClient;
+import org.apache.rocketmq.remoting.RequestProcessor;
import org.apache.rocketmq.remoting.common.Pair;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingConnectException;
@@ -47,14 +48,7 @@ import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
-import org.apache.rocketmq.remoting.ClientConfig;
-import org.apache.rocketmq.remoting.RequestProcessor;
-import org.apache.rocketmq.remoting.interceptor.ExceptionContext;
-import org.apache.rocketmq.remoting.interceptor.Interceptor;
import org.apache.rocketmq.remoting.interceptor.InterceptorGroup;
-import org.apache.rocketmq.remoting.interceptor.InterceptorInvoker;
-import org.apache.rocketmq.remoting.interceptor.RequestContext;
-import org.apache.rocketmq.remoting.interceptor.ResponseContext;
import org.apache.rocketmq.remoting.netty.NettyChannelImpl;
import org.apache.rocketmq.remoting.netty.TlsHelper;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/transport/rocketmq/NettyRemotingServer.java b/remoting/src/main/java/org/apache/rocketmq/remoting/transport/rocketmq/NettyRemotingServer.java
index ec02a28..fa3673c 100644
--- a/remoting/src/main/java/org/apache/rocketmq/remoting/transport/rocketmq/NettyRemotingServer.java
+++ b/remoting/src/main/java/org/apache/rocketmq/remoting/transport/rocketmq/NettyRemotingServer.java
@@ -52,7 +52,6 @@ import org.apache.rocketmq.remoting.common.TlsMode;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
-import org.apache.rocketmq.remoting.interceptor.Interceptor;
import org.apache.rocketmq.remoting.interceptor.InterceptorGroup;
import org.apache.rocketmq.remoting.netty.FileRegionEncoder;
import org.apache.rocketmq.remoting.netty.NettyChannelImpl;
@@ -222,7 +221,7 @@ public class NettyRemotingServer extends NettyRemotingServerAbstract implements
@Override
public void registerProcessor(int requestCode, RequestProcessor processor, ExecutorService executor) {
- executor = (executor == null ? this.publicExecutor : executor);
+ executor = executor == null ? this.publicExecutor : executor;
registerNettyProcessor(requestCode, processor, executor);
}
diff --git a/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingClientTest.java b/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingClientTest.java
index 13bccca..3c5b9f0 100644
--- a/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingClientTest.java
+++ b/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingClientTest.java
@@ -31,7 +31,7 @@ public class NettyRemotingClientTest {
private NettyRemotingClient remotingClient = new NettyRemotingClient(new ClientConfig());
@Test
- public void testSetCallbackExecutor() throws NoSuchFieldException, IllegalAccessException {
+ public void testSetCallbackExecutor() {
ExecutorService customized = Executors.newCachedThreadPool();
remotingClient.setCallbackExecutor(customized);
diff --git a/remoting/src/test/java/org/apache/rocketmq/remoting/protocol/RemotingCommandTest.java b/remoting/src/test/java/org/apache/rocketmq/remoting/protocol/RemotingCommandTest.java
index 243d7f3..c9bf0b8 100644
--- a/remoting/src/test/java/org/apache/rocketmq/remoting/protocol/RemotingCommandTest.java
+++ b/remoting/src/test/java/org/apache/rocketmq/remoting/protocol/RemotingCommandTest.java
@@ -109,7 +109,7 @@ public class RemotingCommandTest {
@Test
public void testNotNullField() throws Exception {
RemotingCommand remotingCommand = new RemotingCommand();
- Method method = RemotingCommand.class.getDeclaredMethod("isFieldNullable", Field.class);
+ Method method = CodecHelper.class.getDeclaredMethod("isFieldNullable", Field.class);
method.setAccessible(true);
Field nullString = FieldTestClass.class.getDeclaredField("nullString");
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/client/DefaultConsumerIdsChangeListener.java b/snode/src/main/java/org/apache/rocketmq/snode/client/DefaultConsumerIdsChangeListener.java
index df74e93..90a5e99 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/client/DefaultConsumerIdsChangeListener.java
+++ b/snode/src/main/java/org/apache/rocketmq/snode/client/DefaultConsumerIdsChangeListener.java
@@ -20,7 +20,9 @@ import java.util.List;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.snode.SnodeController;
-//TODO Filter implementation
+/**
+ * TODO Filter
+ */
public class DefaultConsumerIdsChangeListener implements ConsumerIdsChangeListener {
private final SnodeController snodeController;
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/constant/SnodeConstant.java b/snode/src/main/java/org/apache/rocketmq/snode/constant/SnodeConstant.java
index 65838f3..ea1fbfa 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/constant/SnodeConstant.java
+++ b/snode/src/main/java/org/apache/rocketmq/snode/constant/SnodeConstant.java
@@ -15,12 +15,13 @@
* limitations under the License.
*/
package org.apache.rocketmq.snode.constant;
+
public class SnodeConstant {
- public static final long heartbeatTimeout = 3000;
+ public static final long HEARTBEAT_TIME_OUT = 3000;
- public static final long oneWaytimeout = 10;
+ public static final long ONE_WAY_TIMEOUT = 10;
- public static final long defaultTimeoutMills = 3000L;
+ public static final long DEFAULT_TIMEOUT_MILLS = 3000L;
public static final long CONSUMER_TIMEOUT_MILLIS_WHEN_SUSPEND = 1000 * 30;
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/flowcontrol/RequestSizeFlowControlServiceImpl.java b/snode/src/main/java/org/apache/rocketmq/snode/flowcontrol/RequestSizeFlowControlServiceImpl.java
index 9088694..956e79f 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/flowcontrol/RequestSizeFlowControlServiceImpl.java
+++ b/snode/src/main/java/org/apache/rocketmq/snode/flowcontrol/RequestSizeFlowControlServiceImpl.java
@@ -16,7 +16,6 @@
*/
package org.apache.rocketmq.snode.flowcontrol;
-import com.sun.org.apache.bcel.internal.generic.IF_ACMPEQ;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.flowcontrol.AbstractFlowControlService;
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/processor/HeartbeatProcessor.java b/snode/src/main/java/org/apache/rocketmq/snode/processor/HeartbeatProcessor.java
index 5bdae5c..d66c027 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/processor/HeartbeatProcessor.java
+++ b/snode/src/main/java/org/apache/rocketmq/snode/processor/HeartbeatProcessor.java
@@ -61,7 +61,7 @@ public class HeartbeatProcessor implements RequestProcessor {
private RemotingCommand heartbeat(RemotingChannel remotingChannel, RemotingCommand request) {
HeartbeatData heartbeatData = HeartbeatData.decode(request.getBody(), HeartbeatData.class);
ClientChannelInfo clientChannelInfo = new ClientChannelInfo(
- new NettyChannelImpl((((NettyChannelHandlerContextImpl) remotingChannel).getChannelHandlerContext().channel())),
+ new NettyChannelImpl(((NettyChannelHandlerContextImpl) remotingChannel).getChannelHandlerContext().channel()),
heartbeatData.getClientID(),
request.getLanguage(),
request.getVersion()
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/processor/SendMessageProcessor.java b/snode/src/main/java/org/apache/rocketmq/snode/processor/SendMessageProcessor.java
index 09cd90b..f9f6be2 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/processor/SendMessageProcessor.java
+++ b/snode/src/main/java/org/apache/rocketmq/snode/processor/SendMessageProcessor.java
@@ -54,7 +54,6 @@ public class SendMessageProcessor implements RequestProcessor {
boolean isSendBack = false;
if (request.getCode() == RequestCode.SEND_MESSAGE_V2) {
sendMessageRequestHeaderV2 = (SendMessageRequestHeaderV2) request.decodeCommandCustomHeader(SendMessageRequestHeaderV2.class);
- System.out.println("sendMessageRequestHeaderV2: " + sendMessageRequestHeaderV2);
enodeName = sendMessageRequestHeaderV2.getN();
} else {
isSendBack = true;
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/service/impl/EnodeServiceImpl.java b/snode/src/main/java/org/apache/rocketmq/snode/service/impl/EnodeServiceImpl.java
index abbddd7..7a70997 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/service/impl/EnodeServiceImpl.java
+++ b/snode/src/main/java/org/apache/rocketmq/snode/service/impl/EnodeServiceImpl.java
@@ -29,13 +29,10 @@ import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.protocol.RequestCode;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.header.ConsumerSendMsgBackRequestHeader;
import org.apache.rocketmq.common.protocol.header.CreateTopicRequestHeader;
import org.apache.rocketmq.common.protocol.header.GetMinOffsetRequestHeader;
import org.apache.rocketmq.common.protocol.header.NotifyConsumerIdsChangedRequestHeader;
-import org.apache.rocketmq.common.protocol.header.PullMessageRequestHeader;
import org.apache.rocketmq.common.protocol.header.QueryConsumerOffsetRequestHeader;
-import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeaderV2;
import org.apache.rocketmq.common.protocol.header.UpdateConsumerOffsetRequestHeader;
import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.logging.InternalLogger;
@@ -71,7 +68,7 @@ public class EnodeServiceImpl implements EnodeService {
String enodeAddr = entry.getValue().get(MixAll.MASTER_ID);
if (enodeAddr != null) {
try {
- this.snodeController.getRemotingClient().invokeSync(enodeAddr, remotingCommand, SnodeConstant.defaultTimeoutMills);
+ this.snodeController.getRemotingClient().invokeSync(enodeAddr, remotingCommand, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
} catch (Exception ex) {
log.warn("Send heart beat faild:{} ,ex:{}", enodeAddr, ex);
}
@@ -114,7 +111,7 @@ public class EnodeServiceImpl implements EnodeService {
CompletableFuture<RemotingCommand> future = new CompletableFuture<>();
try {
String enodeAddress = this.snodeController.getNnodeService().getAddressByEnodeName(enodeName, false);
- this.snodeController.getRemotingClient().invokeAsync(enodeAddress, request, SnodeConstant.defaultTimeoutMills, (responseFuture) -> {
+ this.snodeController.getRemotingClient().invokeAsync(enodeAddress, request, SnodeConstant.DEFAULT_TIMEOUT_MILLS, (responseFuture) -> {
future.complete(responseFuture.getResponseCommand());
});
} catch (Exception ex) {
@@ -139,7 +136,7 @@ public class EnodeServiceImpl implements EnodeService {
RemotingCommand.createRequestCommand(RequestCode.NOTIFY_CONSUMER_IDS_CHANGED, requestHeader);
try {
- this.snodeController.getSnodeServer().invokeOneway(channel, request, SnodeConstant.oneWaytimeout);
+ this.snodeController.getSnodeServer().invokeOneway(channel, request, SnodeConstant.ONE_WAY_TIMEOUT);
} catch (Exception e) {
log.error("NotifyConsumerIdsChanged consumer group: {} exception ", consumerGroup, e);
}
@@ -164,7 +161,7 @@ public class EnodeServiceImpl implements EnodeService {
public void updateEnodeAddr(String clusterName) throws InterruptedException, RemotingTimeoutException,
RemotingSendRequestException, RemotingConnectException, MQBrokerException {
synchronized (this) {
- ClusterInfo clusterInfo = getBrokerClusterInfo(SnodeConstant.defaultTimeoutMills);
+ ClusterInfo clusterInfo = getBrokerClusterInfo(SnodeConstant.DEFAULT_TIMEOUT_MILLS);
if (clusterInfo != null) {
HashMap<String, Set<String>> enodeAddress = clusterInfo.getClusterAddrTable();
for (Map.Entry<String, Set<String>> entry : enodeAddress.entrySet()) {
@@ -189,7 +186,7 @@ public class EnodeServiceImpl implements EnodeService {
String enodeAddress = entry.getValue().get(MixAll.MASTER_ID);
try {
RemotingCommand response = this.snodeController.getRemotingClient().invokeSync(enodeAddress,
- request, SnodeConstant.defaultTimeoutMills);
+ request, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
if (response != null && response.getCode() == ResponseCode.SUCCESS) {
persist = true;
} else {
@@ -215,7 +212,7 @@ public class EnodeServiceImpl implements EnodeService {
requestHeader.setCommitOffset(offset);
requestHeader.setEnodeName(enodeName);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UPDATE_CONSUMER_OFFSET, requestHeader);
- this.snodeController.getRemotingClient().invokeOneway(address, request, SnodeConstant.defaultTimeoutMills);
+ this.snodeController.getRemotingClient().invokeOneway(address, request, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
} catch (Exception ex) {
log.error("Persist offset to Enode error!");
}
@@ -231,7 +228,7 @@ public class EnodeServiceImpl implements EnodeService {
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_MIN_OFFSET, requestHeader);
String addr = this.snodeController.getNnodeService().getAddressByEnodeName(enodeName, false);
return this.snodeController.getRemotingClient().invokeSync(MixAll.brokerVIPChannel(snodeController.getSnodeConfig().isVipChannelEnabled(), addr),
- request, SnodeConstant.defaultTimeoutMills);
+ request, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
}
@Override
@@ -246,7 +243,7 @@ public class EnodeServiceImpl implements EnodeService {
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.QUERY_CONSUMER_OFFSET, requestHeader);
String addr = this.snodeController.getNnodeService().getAddressByEnodeName(enodeName, false);
return this.snodeController.getRemotingClient().invokeSync(MixAll.brokerVIPChannel(this.snodeController.getSnodeConfig().isVipChannelEnabled(), addr),
- request, SnodeConstant.defaultTimeoutMills);
+ request, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
}
@Override
@@ -254,7 +251,7 @@ public class EnodeServiceImpl implements EnodeService {
RemotingCommand request) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, RemotingCommandException {
String addr = this.snodeController.getNnodeService().getAddressByEnodeName(enodeName, false);
return this.snodeController.getRemotingClient().invokeSync(MixAll.brokerVIPChannel(snodeController.getSnodeConfig().isVipChannelEnabled(), addr),
- request, SnodeConstant.defaultTimeoutMills);
+ request, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
}
@Override
@@ -262,7 +259,7 @@ public class EnodeServiceImpl implements EnodeService {
RemotingCommand request) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
String addr = this.snodeController.getNnodeService().getAddressByEnodeName(enodeName, false);
return this.snodeController.getRemotingClient().invokeSync(MixAll.brokerVIPChannel(snodeController.getSnodeConfig().isVipChannelEnabled(), addr),
- request, SnodeConstant.defaultTimeoutMills);
+ request, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
}
@Override
@@ -279,6 +276,6 @@ public class EnodeServiceImpl implements EnodeService {
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UPDATE_AND_CREATE_TOPIC, requestHeader);
String address = this.snodeController.getNnodeService().getAddressByEnodeName(enodeName, false);
return this.snodeController.getRemotingClient().invokeSync(address,
- request, SnodeConstant.defaultTimeoutMills);
+ request, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
}
}
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/service/impl/NnodeServiceImpl.java b/snode/src/main/java/org/apache/rocketmq/snode/service/impl/NnodeServiceImpl.java
index e1c2017..8853b1f 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/service/impl/NnodeServiceImpl.java
+++ b/snode/src/main/java/org/apache/rocketmq/snode/service/impl/NnodeServiceImpl.java
@@ -66,7 +66,7 @@ public class NnodeServiceImpl implements NnodeService {
if (nnodeAddressList != null && nnodeAddressList.size() > 0) {
for (String nodeAddress : nnodeAddressList) {
try {
- this.snodeController.getRemotingClient().invokeSync(nodeAddress, remotingCommand, SnodeConstant.heartbeatTimeout);
+ this.snodeController.getRemotingClient().invokeSync(nodeAddress, remotingCommand, SnodeConstant.HEARTBEAT_TIME_OUT);
} catch (Exception ex) {
log.warn("Register Snode to Nnode addr: {} error, ex:{} ", nodeAddress, ex);
}
@@ -93,7 +93,7 @@ public class NnodeServiceImpl implements NnodeService {
requestHeader.setTopic(topic);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ROUTEINTO_BY_TOPIC, requestHeader);
- RemotingCommand response = this.snodeController.getRemotingClient().invokeSync(null, request, SnodeConstant.defaultTimeoutMills);
+ RemotingCommand response = this.snodeController.getRemotingClient().invokeSync(null, request, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
log.info("GetTopicRouteInfoFromNameServer response: " + response);
assert response != null;
switch (response.getCode()) {
@@ -167,7 +167,7 @@ public class NnodeServiceImpl implements NnodeService {
RemotingSendRequestException, RemotingConnectException {
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_BROKER_CLUSTER_INFO, null);
- RemotingCommand response = this.snodeController.getRemotingClient().invokeSync(null, request, SnodeConstant.defaultTimeoutMills);
+ RemotingCommand response = this.snodeController.getRemotingClient().invokeSync(null, request, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
switch (response.getCode()) {
case ResponseCode.SUCCESS: {
ClusterInfo clusterInfo = ClusterInfo.decode(response.getBody(), ClusterInfo.class);
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/service/impl/PushServiceImpl.java b/snode/src/main/java/org/apache/rocketmq/snode/service/impl/PushServiceImpl.java
index 1ea56c1..8a42b1c 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/service/impl/PushServiceImpl.java
+++ b/snode/src/main/java/org/apache/rocketmq/snode/service/impl/PushServiceImpl.java
@@ -90,7 +90,7 @@ public class PushServiceImpl implements PushService {
RemotingChannel remotingChannel = clientChannelInfoEntry.getValue().getChannel();
if (remotingChannel.isWritable()) {
log.warn("Push message to topic: {} queueId: {} consumer group:{}, message:{}", topic, queueId, clientChannelInfoEntry.getKey(), pushMessage);
- snodeController.getSnodeServer().push(remotingChannel, pushMessage, SnodeConstant.defaultTimeoutMills);
+ snodeController.getSnodeServer().push(remotingChannel, pushMessage, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
}
}
} else {