Add more tests

Author: aaron-ai

java/client/src/test/java/org/apache/rocketmq/client/java/impl/consumer/PullConsumerImplTest.java

@@ -18,12 +18,16 @@
 package org.apache.rocketmq.client.java.impl.consumer;
 
 import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
+import apache.rocketmq.v2.UpdateOffsetRequest;
+import apache.rocketmq.v2.UpdateOffsetResponse;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
@@ -38,14 +42,14 @@
 import org.apache.rocketmq.client.java.message.MessageViewImpl;
 import org.apache.rocketmq.client.java.route.MessageQueueImpl;
 import org.apache.rocketmq.client.java.route.TopicRouteData;
+import org.apache.rocketmq.client.java.rpc.RpcFuture;
 import org.apache.rocketmq.client.java.tool.TestBase;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.junit.MockitoJUnitRunner;
 
-@SuppressWarnings("resource")
 @RunWith(MockitoJUnitRunner.class)
 public class PullConsumerImplTest extends TestBase {
 
@@ -252,4 +256,62 @@ public void onChanged(String topic, Set<MessageQueue> messageQueues) {
         pullConsumer.registerMessageQueueChangeListenerByTopic(FAKE_TOPIC_0, listener);
         verify(pullConsumer, times(1)).fetchMessageQueues(anyString());
     }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testSeekWithMessageQueueIsNotContained() {
+        doReturn(true).when(pullConsumer).isRunning();
+        final MessageQueueImpl mq = fakeMessageQueueImpl(FAKE_TOPIC_0);
+        pullConsumer.seek(mq, 1);
+    }
+
+    @Test
+    public void testSeek() {
+        doReturn(true).when(pullConsumer).isRunning();
+        MessageQueueImpl mq = fakeMessageQueueImpl(FAKE_TOPIC_0);
+        List<MessageQueue> mqs = new ArrayList<>();
+        mqs.add(mq);
+        pullConsumer.assign(mqs);
+        pullConsumer.seek(mq, 1);
+        verify(pullConsumer, times(1)).dropProcessQueue(any(MessageQueueImpl.class));
+        verify(pullConsumer, times(1))
+            .tryPullMessageByMessageQueueImmediately(any(MessageQueueImpl.class), any(FilterExpression.class),
+                anyLong());
+    }
+
+    @Test
+    public void testSeekToBegin() throws ClientException {
+        doReturn(true).when(pullConsumer).isRunning();
+        final MessageQueueImpl mq = fakeMessageQueueImpl(FAKE_TOPIC_0);
+        List<MessageQueue> mqs = new ArrayList<>();
+        mqs.add(mq);
+        pullConsumer.assign(mqs);
+        doReturn(okQueryOffsetResponseFuture()).when(pullConsumer).queryOffset(any(MessageQueueImpl.class),
+            any(OffsetPolicy.class));
+        pullConsumer.seekToBegin(mq);
+        verify(pullConsumer, times(1)).queryOffset(any(MessageQueueImpl.class),
+            any(OffsetPolicy.class));
+    }
+
+    @Test
+    public void testSeekToEnd() throws ClientException {
+        doReturn(true).when(pullConsumer).isRunning();
+        final MessageQueueImpl mq = fakeMessageQueueImpl(FAKE_TOPIC_0);
+        List<MessageQueue> mqs = new ArrayList<>();
+        mqs.add(mq);
+        pullConsumer.assign(mqs);
+        doReturn(okQueryOffsetResponseFuture()).when(pullConsumer).queryOffset(any(MessageQueueImpl.class),
+            any(OffsetPolicy.class));
+        pullConsumer.seekToEnd(mq);
+        verify(pullConsumer, times(1)).queryOffset(any(MessageQueueImpl.class),
+            any(OffsetPolicy.class));
+    }
+
+    @Test
+    public void testCommit() throws ClientException {
+        List<RpcFuture<UpdateOffsetRequest, UpdateOffsetResponse>> futures = new ArrayList<>();
+        final RpcFuture<UpdateOffsetRequest, UpdateOffsetResponse> future = okUpdateOffsetResponseFuture();
+        futures.add(future);
+        doReturn(futures).when(pullConsumer).commit0();
+        pullConsumer.commit();
+    }
 }

java/client/src/test/java/org/apache/rocketmq/client/java/tool/TestBase.java

@@ -40,6 +40,8 @@
 import apache.rocketmq.v2.PullMessageResponse;
 import apache.rocketmq.v2.QueryAssignmentRequest;
 import apache.rocketmq.v2.QueryAssignmentResponse;
+import apache.rocketmq.v2.QueryOffsetRequest;
+import apache.rocketmq.v2.QueryOffsetResponse;
 import apache.rocketmq.v2.ReceiveMessageRequest;
 import apache.rocketmq.v2.ReceiveMessageResponse;
 import apache.rocketmq.v2.Resource;
@@ -48,6 +50,8 @@
 import apache.rocketmq.v2.SendResultEntry;
 import apache.rocketmq.v2.Status;
 import apache.rocketmq.v2.SystemProperties;
+import apache.rocketmq.v2.UpdateOffsetRequest;
+import apache.rocketmq.v2.UpdateOffsetResponse;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
@@ -297,6 +301,12 @@ ForwardMessageToDeadLetterQueueResponse> okForwardMessageToDeadLetterQueueRespon
         return new RpcFuture<>(fakeRpcContext(), null, Futures.immediateFuture(response));
     }
 
+    protected RpcFuture<UpdateOffsetRequest, UpdateOffsetResponse> okUpdateOffsetResponseFuture() {
+        final Status status = Status.newBuilder().setCode(Code.OK).build();
+        final UpdateOffsetResponse response = UpdateOffsetResponse.newBuilder().setStatus(status).build();
+        return new RpcFuture<>(fakeRpcContext(), null, Futures.immediateFuture(response));
+    }
+
     protected RpcFuture<ForwardMessageToDeadLetterQueueRequest,
         ForwardMessageToDeadLetterQueueResponse> forwardMessageToDeadLetterQueueResponseFuture(Code code) {
         final Status status = Status.newBuilder().setCode(code).build();
@@ -383,6 +393,12 @@ protected RpcFuture<GetOffsetRequest, GetOffsetResponse> okGetOffsetResponseFutu
         return new RpcFuture<>(fakeRpcContext(), null, Futures.immediateFuture(response));
     }
 
+    protected RpcFuture<QueryOffsetRequest, QueryOffsetResponse> okQueryOffsetResponseFuture() {
+        final Status status = Status.newBuilder().setCode(Code.OK).build();
+        QueryOffsetResponse response = QueryOffsetResponse.newBuilder().setOffset(0).setStatus(status).build();
+        return new RpcFuture<>(fakeRpcContext(), null, Futures.immediateFuture(response));
+    }
+
     protected ListenableFuture<EndTransactionResponse> okEndTransactionResponseFuture() {
         SettableFuture<EndTransactionResponse> future = SettableFuture.create();
         final Status status = Status.newBuilder().setCode(Code.OK).build();