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

Merge pull request #746 from zongtanghu/snode

Browse Source 
[ISSUE #743]Fix code style issue and adjust some unit test in client module for the client.
This commit is contained in:
Heng Du
2019-02-01 16:10:02 +08:00
committed by GitHub
parent 81b4293a47 4f24cd4918
commit e6d0afbf90
13 changed files with 88 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/main/java/org/apache/rocketmq/client/consumer/store/LocalFileOffsetStore.java
+6
View File
@@ -160,6 +160,12 @@ public class LocalFileOffsetStore implements OffsetStore {
}
@Override
public void updateConsumeOffsetToSnode(final MessageQueue mq, final long offset, final boolean isOneway)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
}
@Override
public void updateConsumeOffsetToBroker(final MessageQueue mq, final long offset, final boolean isOneway)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
client/src/main/java/org/apache/rocketmq/client/consumer/store/OffsetStore.java
+8
View File
@@ -71,4 +71,12 @@ public interface OffsetStore {
*/
void updateConsumeOffsetToBroker(MessageQueue mq, long offset, boolean isOneway) throws RemotingException,
MQBrokerException, InterruptedException, MQClientException;
/**
* @param mq
* @param offset
* @param isOneway
*/
void updateConsumeOffsetToSnode(MessageQueue mq, long offset, boolean isOneway) throws RemotingException,
MQBrokerException, InterruptedException, MQClientException;
}
client/src/main/java/org/apache/rocketmq/client/consumer/store/RemoteBrokerOffsetStore.java
+63 -8
View File
@@ -88,7 +88,7 @@ public class RemoteBrokerOffsetStore implements OffsetStore {
}
case READ_FROM_STORE: {
try {
long brokerOffset = this.fetchConsumeOffsetFromBroker(mq);
long brokerOffset = this.fetchConsumeOffsetFromSnode(mq);
AtomicLong offset = new AtomicLong(brokerOffset);
this.updateOffset(mq, offset.get(), false);
return brokerOffset;
@@ -195,20 +195,20 @@ public class RemoteBrokerOffsetStore implements OffsetStore {
}
private void updateConsumeOffsetToSnode(MessageQueue mq, long offset) throws RemotingException,
MQBrokerException, InterruptedException, MQClientException {
updateConsumeOffsetToBroker(mq, offset, true);
updateConsumeOffsetToSnode(mq, offset, true);
}
/**
* Update the Consumer Offset synchronously, once the Master is off, updated to Slave, here need to be optimized.
*/
@Override
public void updateConsumeOffsetToBroker(MessageQueue mq, long offset, boolean isOneway) throws RemotingException,
public void updateConsumeOffsetToSnode(MessageQueue mq, long offset, boolean isOneway) throws RemotingException,
MQBrokerException, InterruptedException, MQClientException {
String snodeAddr = this.mQClientFactory.findSnodeAddressInPublish();
if (null == snodeAddr){
if (null == snodeAddr) {
this.mQClientFactory.updateSnodeInfoFromNameServer();
snodeAddr= this.mQClientFactory.findSnodeAddressInPublish();
snodeAddr = this.mQClientFactory.findSnodeAddressInPublish();
}
if (snodeAddr != null) {
@@ -226,16 +226,71 @@ public class RemoteBrokerOffsetStore implements OffsetStore {
snodeAddr, requestHeader, 1000 * 5);
}
} else {
throw new MQClientException("Update offset to Broker[" + mq.getBrokerName() + "] failed, Snode is null.", null);
throw new MQClientException("Update offset to Snode[" + mq.getBrokerName() + "] failed, Snode is null.", null);
}
}
/**
* Preserved firstly,Compatible with RocketMQ 4.X Version
*/
@Override
public void updateConsumeOffsetToBroker(MessageQueue mq, long offset,
boolean isOneway) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
FindBrokerResult findBrokerResult = this.mQClientFactory.findBrokerAddressInAdmin(mq.getBrokerName());
if (null == findBrokerResult) {
this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic());
findBrokerResult = this.mQClientFactory.findBrokerAddressInAdmin(mq.getBrokerName());
}
if (findBrokerResult != null) {
UpdateConsumerOffsetRequestHeader requestHeader = new UpdateConsumerOffsetRequestHeader();
requestHeader.setTopic(mq.getTopic());
requestHeader.setConsumerGroup(this.groupName);
requestHeader.setQueueId(mq.getQueueId());
requestHeader.setCommitOffset(offset);
requestHeader.setEnodeName(mq.getBrokerName());
if (isOneway) {
this.mQClientFactory.getMQClientAPIImpl().updateConsumerOffsetOneway(
findBrokerResult.getBrokerAddr(), requestHeader, 1000 * 5);
} else {
this.mQClientFactory.getMQClientAPIImpl().updateConsumerOffset(
findBrokerResult.getBrokerAddr(), requestHeader, 1000 * 5);
}
} else {
throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
}
}
private long fetchConsumeOffsetFromSnode(MessageQueue mq) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException {
String snodeAddr = this.mQClientFactory.findSnodeAddressInPublish();
if (null == snodeAddr) {
this.mQClientFactory.updateSnodeInfoFromNameServer();
snodeAddr = this.mQClientFactory.findSnodeAddressInPublish();
}
if (snodeAddr != null) {
QueryConsumerOffsetRequestHeader requestHeader = new QueryConsumerOffsetRequestHeader();
requestHeader.setTopic(mq.getTopic());
requestHeader.setConsumerGroup(this.groupName);
requestHeader.setQueueId(mq.getQueueId());
requestHeader.setEnodeName(mq.getBrokerName());
return this.mQClientFactory.getMQClientAPIImpl().queryConsumerOffset(
snodeAddr, requestHeader, 1000 * 5);
} else {
throw new MQClientException("Get Offset from Snode[" + mq.getBrokerName() + "] failed, Snode is not exist", null);
}
}
/**
* Preserved firstly,Compatible with RocketMQ 4.X Version
*/
private long fetchConsumeOffsetFromBroker(MessageQueue mq) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException {
String snodeAddr = this.mQClientFactory.findSnodeAddressInPublish();
if (null == snodeAddr){
if (null == snodeAddr) {
this.mQClientFactory.updateSnodeInfoFromNameServer();
snodeAddr= this.mQClientFactory.findSnodeAddressInPublish();
snodeAddr = this.mQClientFactory.findSnodeAddressInPublish();
}
if (snodeAddr != null) {
client/src/main/java/org/apache/rocketmq/client/exception/MQSnodeException.java
-3
View File
@@ -16,9 +16,6 @@
*/
package org.apache.rocketmq.client.exception;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.help.FAQUrl;
public class MQSnodeException extends MQBrokerException {
public MQSnodeException(int responseCode, String errorMessage) {
client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
+1 -1
View File
@@ -1194,7 +1194,7 @@ public class MQClientAPIImpl {
public SnodeClusterInfo getSnodeClusterInfo(
//Todo Redifine snode exception
final long timeoutMillis) throws InterruptedException, RemotingTimeoutException,
RemotingSendRequestException, RemotingConnectException , MQBrokerException{
RemotingSendRequestException, RemotingConnectException , MQBrokerException {
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_SNODE_CLUSTER_INFO, null);
RemotingCommand response = this.remotingClient.invokeSync(null, request, timeoutMillis);
client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPullConsumerImpl.java
+2 -2
View File
@@ -480,9 +480,9 @@ public class DefaultMQPullConsumerImpl implements MQConsumerInner {
sendMessageBack(msg, delayLevel, brokerName, this.defaultMQPullConsumer.getConsumerGroup());
}
public void updateConsumeOffsetToBroker(MessageQueue mq, long offset, boolean isOneway) throws RemotingException,
public void updateConsumeOffsetToSnode(MessageQueue mq, long offset, boolean isOneway) throws RemotingException,
MQBrokerException, InterruptedException, MQClientException {
this.offsetStore.updateConsumeOffsetToBroker(mq, offset, isOneway);
this.offsetStore.updateConsumeOffsetToSnode(mq, offset, isOneway);
}
public void sendMessageBack(MessageExt msg, int delayLevel, final String brokerName, String consumerGroup)
client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java
-1
View File
@@ -73,7 +73,6 @@ import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.common.sysflag.PullSysFlag;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
public class DefaultMQPushConsumerImpl implements MQConsumerInner {
client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java
-2
View File
@@ -38,7 +38,6 @@ import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.rocketmq.client.admin.MQAdminExtInner;
import org.apache.rocketmq.client.common.ThreadLocalIndex;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.impl.ClientRemotingProcessor;
import org.apache.rocketmq.client.impl.FindBrokerResult;
@@ -80,7 +79,6 @@ import org.apache.rocketmq.common.protocol.route.TopicRouteData;
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.interceptor.InterceptorGroup;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
client/src/test/java/org/apache/rocketmq/client/consumer/DefaultMQPullConsumerTest.java
+2
View File
@@ -77,6 +77,8 @@ public class DefaultMQPullConsumerTest {
field.set(mQClientFactory, mQClientAPIImpl);
when(mQClientFactory.findBrokerAddressInSubscribe(anyString(), anyLong(), anyBoolean())).thenReturn(new FindBrokerResult("127.0.0.1:10911", false));
when(mQClientFactory.findSnodeAddressInPublish()).thenReturn("127.0.0.1:10911");
}
@After
client/src/test/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumerTest.java
+2
View File
@@ -147,6 +147,8 @@ public class DefaultMQPushConsumerTest {
});
doReturn(new FindBrokerResult("127.0.0.1:10911", false)).when(mQClientFactory).findBrokerAddressInSubscribe(anyString(), anyLong(), anyBoolean());
doReturn("127.0.0.1:10911").when(mQClientFactory).findSnodeAddressInPublish();
Set<MessageQueue> messageQueueSet = new HashSet<MessageQueue>();
messageQueueSet.add(createPullRequest().getMessageQueue());
pushConsumer.getDefaultMQPushConsumerImpl().updateTopicSubscribeInfo(topic, messageQueueSet);
client/src/test/java/org/apache/rocketmq/client/consumer/store/RemoteBrokerOffsetStoreTest.java
+1 -3
View File
@@ -20,7 +20,6 @@ import java.util.Collections;
import java.util.HashSet;
import org.apache.rocketmq.client.ClientConfig;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.impl.FindBrokerResult;
import org.apache.rocketmq.client.impl.MQClientAPIImpl;
import org.apache.rocketmq.client.impl.factory.MQClientInstance;
import org.apache.rocketmq.common.message.MessageQueue;
@@ -34,7 +33,6 @@ import org.mockito.Mock;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.junit.MockitoJUnitRunner;
import org.mockito.stubbing.Answer;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
@@ -58,8 +56,8 @@ public class RemoteBrokerOffsetStoreTest {
System.setProperty("rocketmq.client.localOffsetStoreDir", System.getProperty("java.io.tmpdir") + ".rocketmq_offsets");
String clientId = new ClientConfig().buildMQClientId() + "#TestNamespace" + System.currentTimeMillis();
when(mQClientFactory.getClientId()).thenReturn(clientId);
when(mQClientFactory.findBrokerAddressInAdmin(brokerName)).thenReturn(new FindBrokerResult("127.0.0.1", false));
when(mQClientFactory.getMQClientAPIImpl()).thenReturn(mqClientAPI);
when(mQClientFactory.findSnodeAddressInPublish()).thenReturn("127.0.0.1:10911");
}
@Test
client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java
+2
View File
@@ -106,6 +106,8 @@ public class DefaultMQProducerTest {
when(mQClientAPIImpl.sendMessage(anyString(), anyString(), any(Message.class), any(SendMessageRequestHeader.class), anyLong(), any(CommunicationMode.class),
nullable(SendCallback.class), nullable(TopicPublishInfo.class), nullable(MQClientInstance.class), anyInt(), nullable(SendMessageContext.class), any(DefaultMQProducerImpl.class)))
.thenReturn(createSendResult(SendStatus.SEND_OK));
when(mQClientFactory.findSnodeAddressInPublish()).thenReturn("127.0.0.1:10911");
}
@After
snode/src/main/java/org/apache/rocketmq/snode/service/impl/NnodeServiceImpl.java
+1 -1
View File
@@ -58,7 +58,7 @@ public class NnodeServiceImpl implements NnodeService {
}
@Override
public void registerSnode(SnodeConfig snodeConfig) throws Exception{
public void registerSnode(SnodeConfig snodeConfig) throws Exception {
List<String> nnodeAddressList = this.snodeController.getRemotingClient().getNameServerAddressList();
RemotingCommand remotingCommand = new RemotingCommand();
RegisterSnodeRequestHeader requestHeader = new RegisterSnodeRequestHeader();