rocketmq-commits mailing list archives

This is an automated email from the ASF dual-hosted git repository.

jinrongtong pushed a commit to branch test-release
in repository https://gitbox.apache.org/repos/asf/rocketmq.git

commit 45f6eb1e33a1c21395d4995fd1e0002fb44dd10d
Author: 翊名 <duheng.dh@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);