You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by di...@apache.org on 2019/11/26 03:40:30 UTC
[rocketmq] branch develop updated (c3b7d18 -> 0bddfe5)
This is an automated email from the ASF dual-hosted git repository.
dinglei pushed a change to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git.
from c3b7d18 [maven-release-plugin] prepare for next development iteration
new a56bec8 feat(PullConsumer) support begin/end seek support for pull consumer
new 0bddfe5 test(PullConsumer) add seek to beigin/end unit test
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:
.../client/consumer/DefaultLitePullConsumer.java | 28 ++++++++++++++-----
.../rocketmq/client/consumer/LitePullConsumer.java | 23 ++++++++++++++++
.../impl/consumer/DefaultLitePullConsumerImpl.java | 32 ++++++++++++++++------
.../consumer/DefaultLitePullConsumerTest.java | 30 ++++++++++++++++++++
4 files changed, 97 insertions(+), 16 deletions(-)
[rocketmq] 02/02: test(PullConsumer) add seek to beigin/end unit
test
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dinglei pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
commit 0bddfe50747c270e6c2312a75ff695361e583225
Author: 翊名 <du...@alibaba-inc.com>
AuthorDate: Mon Nov 25 17:28:03 2019 +0800
test(PullConsumer) add seek to beigin/end unit test
---
.../consumer/DefaultLitePullConsumerTest.java | 30 ++++++++++++++++++++++
1 file changed, 30 insertions(+)
diff --git a/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumerTest.java b/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumerTest.java
index b55f8a0..d2cb057 100644
--- a/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumerTest.java
+++ b/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumerTest.java
@@ -190,6 +190,36 @@ public class DefaultLitePullConsumerTest {
}
@Test
+ public void testSeek_SeekToBegin() throws Exception {
+ DefaultLitePullConsumer litePullConsumer = createStartLitePullConsumer();
+ when(mQAdminImpl.minOffset(any(MessageQueue.class))).thenReturn(0L);
+ when(mQAdminImpl.maxOffset(any(MessageQueue.class))).thenReturn(500L);
+ MessageQueue messageQueue = createMessageQueue();
+ litePullConsumer.assign(Collections.singletonList(messageQueue));
+ litePullConsumer.seekToBegin(messageQueue);
+ Field field = DefaultLitePullConsumerImpl.class.getDeclaredField("assignedMessageQueue");
+ field.setAccessible(true);
+ AssignedMessageQueue assignedMessageQueue = (AssignedMessageQueue) field.get(litePullConsumerImpl);
+ assertEquals(assignedMessageQueue.getSeekOffset(messageQueue), 0L);
+ litePullConsumer.shutdown();
+ }
+
+ @Test
+ public void testSeek_SeekToEnd() throws Exception {
+ DefaultLitePullConsumer litePullConsumer = createStartLitePullConsumer();
+ when(mQAdminImpl.minOffset(any(MessageQueue.class))).thenReturn(0L);
+ when(mQAdminImpl.maxOffset(any(MessageQueue.class))).thenReturn(500L);
+ MessageQueue messageQueue = createMessageQueue();
+ litePullConsumer.assign(Collections.singletonList(messageQueue));
+ litePullConsumer.seekToEnd(messageQueue);
+ Field field = DefaultLitePullConsumerImpl.class.getDeclaredField("assignedMessageQueue");
+ field.setAccessible(true);
+ AssignedMessageQueue assignedMessageQueue = (AssignedMessageQueue) field.get(litePullConsumerImpl);
+ assertEquals(assignedMessageQueue.getSeekOffset(messageQueue), 500L);
+ litePullConsumer.shutdown();
+ }
+
+ @Test
public void testSeek_SeekOffsetIllegal() throws Exception {
DefaultLitePullConsumer litePullConsumer = createStartLitePullConsumer();
when(mQAdminImpl.minOffset(any(MessageQueue.class))).thenReturn(0L);
[rocketmq] 01/02: feat(PullConsumer) support begin/end seek support
for pull consumer
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dinglei pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
commit a56bec8163dd16aa32c498b3b6a7100c612d91c1
Author: 翊名 <du...@alibaba-inc.com>
AuthorDate: Mon Nov 25 16:14:31 2019 +0800
feat(PullConsumer) support begin/end seek support for pull consumer
---
.../client/consumer/DefaultLitePullConsumer.java | 28 ++++++++++++++-----
.../rocketmq/client/consumer/LitePullConsumer.java | 23 ++++++++++++++++
.../impl/consumer/DefaultLitePullConsumerImpl.java | 32 ++++++++++++++++------
3 files changed, 67 insertions(+), 16 deletions(-)
diff --git a/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumer.java b/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumer.java
index 99976d5..09f7e20 100644
--- a/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumer.java
+++ b/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumer.java
@@ -18,7 +18,6 @@ package org.apache.rocketmq.client.consumer;
import java.util.Collection;
import java.util.List;
-
import org.apache.rocketmq.client.ClientConfig;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.consumer.store.OffsetStore;
@@ -35,12 +34,12 @@ public class DefaultLitePullConsumer extends ClientConfig implements LitePullCon
private final DefaultLitePullConsumerImpl defaultLitePullConsumerImpl;
/**
- * Consumers belonging to the same consumer group share a group id. The consumers in a group then
- * divides the topic as fairly amongst themselves as possible by establishing that each queue is only
- * consumed by a single consumer from the group. If all consumers are from the same group, it functions
- * as a traditional message queue. Each message would be consumed by one consumer of the group only.
- * When multiple consumer groups exist, the flow of the data consumption model aligns with the traditional
- * publish-subscribe model. The messages are broadcast to all consumer groups.
+ * Consumers belonging to the same consumer group share a group id. The consumers in a group then divides the topic
+ * as fairly amongst themselves as possible by establishing that each queue is only consumed by a single consumer
+ * from the group. If all consumers are from the same group, it functions as a traditional message queue. Each
+ * message would be consumed by one consumer of the group only. When multiple consumer groups exist, the flow of the
+ * data consumption model aligns with the traditional publish-subscribe model. The messages are broadcast to all
+ * consumer groups.
*/
private String consumerGroup;
@@ -267,6 +266,21 @@ public class DefaultLitePullConsumer extends ClientConfig implements LitePullCon
}
@Override
+ public void updateNameServerAddress(String nameServerAddress) {
+ this.defaultLitePullConsumerImpl.updateNameServerAddr(nameServerAddress);
+ }
+
+ @Override
+ public void seekToBegin(MessageQueue messageQueue) throws MQClientException {
+ this.defaultLitePullConsumerImpl.seekToBegin(messageQueue);
+ }
+
+ @Override
+ public void seekToEnd(MessageQueue messageQueue) throws MQClientException {
+ this.defaultLitePullConsumerImpl.seekToEnd(messageQueue);
+ }
+
+ @Override
public boolean isAutoCommit() {
return autoCommit;
}
diff --git a/client/src/main/java/org/apache/rocketmq/client/consumer/LitePullConsumer.java b/client/src/main/java/org/apache/rocketmq/client/consumer/LitePullConsumer.java
index d6e657f..ce22288 100644
--- a/client/src/main/java/org/apache/rocketmq/client/consumer/LitePullConsumer.java
+++ b/client/src/main/java/org/apache/rocketmq/client/consumer/LitePullConsumer.java
@@ -172,4 +172,27 @@ public interface LitePullConsumer {
*/
void registerTopicMessageQueueChangeListener(String topic,
TopicMessageQueueChangeListener topicMessageQueueChangeListener) throws MQClientException;
+
+ /**
+ * Update name server addresses.
+ */
+ void updateNameServerAddress(String nameServerAddress);
+
+ /**
+ * Overrides the fetch offsets with the begin offset that the consumer will use on the next poll. If this API is
+ * invoked for the same message queue more than once, the latest offset will be used on the next poll(). Note that
+ * you may lose data if this API is arbitrarily used in the middle of consumption.
+ *
+ * @param messageQueue
+ */
+ void seekToBegin(MessageQueue messageQueue)throws MQClientException;
+
+ /**
+ * Overrides the fetch offsets with the end offset that the consumer will use on the next poll. If this API is
+ * invoked for the same message queue more than once, the latest offset will be used on the next poll(). Note that
+ * you may lose data if this API is arbitrarily used in the middle of consumption.
+ *
+ * @param messageQueue
+ */
+ void seekToEnd(MessageQueue messageQueue)throws MQClientException;
}
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java
index 2c673a1..f44eea7 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java
@@ -16,16 +16,16 @@
*/
package org.apache.rocketmq.client.impl.consumer;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
-import java.util.List;
-import java.util.ArrayList;
+import java.util.HashSet;
import java.util.Iterator;
+import java.util.List;
import java.util.Map;
-import java.util.Set;
-import java.util.Collection;
-import java.util.HashSet;
import java.util.Properties;
+import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -35,13 +35,12 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
-
import org.apache.rocketmq.client.Validators;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
-import org.apache.rocketmq.client.consumer.TopicMessageQueueChangeListener;
-import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.MessageQueueListener;
+import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.PullResult;
+import org.apache.rocketmq.client.consumer.TopicMessageQueueChangeListener;
import org.apache.rocketmq.client.consumer.store.LocalFileOffsetStore;
import org.apache.rocketmq.client.consumer.store.OffsetStore;
import org.apache.rocketmq.client.consumer.store.ReadOffsetType;
@@ -164,6 +163,10 @@ public class DefaultLitePullConsumerImpl implements MQConsumerInner {
throw new IllegalStateException(NOT_RUNNING_EXCEPTION_MESSAGE);
}
+ public void updateNameServerAddr(String newAddresses) {
+ this.mQClientFactory.getMQClientAPIImpl().updateNameServerAddressList(newAddresses);
+ }
+
private synchronized void setSubscriptionType(SubscriptionType type) {
if (this.subscriptionType == SubscriptionType.NONE)
this.subscriptionType = type;
@@ -556,6 +559,16 @@ public class DefaultLitePullConsumerImpl implements MQConsumerInner {
}
}
+ public void seekToBegin(MessageQueue messageQueue) throws MQClientException {
+ long begin = minOffset(messageQueue);
+ this.seek(messageQueue, begin);
+ }
+
+ public void seekToEnd(MessageQueue messageQueue) throws MQClientException {
+ long begin = maxOffset(messageQueue);
+ this.seek(messageQueue, begin);
+ }
+
private long maxOffset(MessageQueue messageQueue) throws MQClientException {
checkServiceState();
return this.mQClientFactory.getMQAdminImpl().maxOffset(messageQueue);
@@ -764,8 +777,9 @@ public class DefaultLitePullConsumerImpl implements MQConsumerInner {
subscriptionData = FilterAPI.buildSubscriptionData(defaultLitePullConsumer.getConsumerGroup(),
topic, SubscriptionData.SUB_ALL);
}
-
PullResult pullResult = pull(messageQueue, subscriptionData, offset, nextPullBatchSize());
+
+
switch (pullResult.getPullStatus()) {
case FOUND:
final Object objLock = messageQueueLock.fetchLockObject(messageQueue);