[ISSUE #10410] Fix RocksDBOptionsFactoryTest compile failure (#10411)

Author: RongtongJin

broker/src/test/java/org/apache/rocketmq/broker/BrokerOuterAPITest.java

@@ -44,6 +44,7 @@
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.message.MessageDecoder;
 import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.remoting.InvokeCallback;
 import org.apache.rocketmq.remoting.common.SemaphoreReleaseOnlyOnce;
 import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
 import org.apache.rocketmq.remoting.netty.NettyClientConfig;
@@ -334,12 +335,7 @@ public void testPullMessageFromSpecificBrokerAsync_timeout() throws Exception {
 
     @Test
     public void testPullMessageFromSpecificBrokerAsync_brokerReturn_pullStatusCode() throws Exception {
-        Channel channel = Mockito.mock(Channel.class);
-        when(channel.isActive()).thenReturn(true);
-        NettyRemotingClient mockClient = PowerMockito.spy(new NettyRemotingClient(new NettyClientConfig()));
-        DefaultChannelPromise promise = PowerMockito.spy(new DefaultChannelPromise(PowerMockito.mock(Channel.class), new DefaultEventExecutor()));
-        PowerMockito.when(mockClient, "getAndCreateChannelAsync", any()).thenReturn(promise);
-        when(promise.channel()).thenReturn(channel);
+        NettyRemotingClient mockClient = Mockito.mock(NettyRemotingClient.class);
         BrokerOuterAPI api = new BrokerOuterAPI(new NettyClientConfig(), new AuthConfig());
         Field field = BrokerOuterAPI.class.getDeclaredField("remotingClient");
         field.setAccessible(true);
@@ -348,14 +344,12 @@ public void testPullMessageFromSpecificBrokerAsync_brokerReturn_pullStatusCode()
         int[] respCodes = new int[] {ResponseCode.SUCCESS, ResponseCode.PULL_NOT_FOUND, ResponseCode.PULL_RETRY_IMMEDIATELY, ResponseCode.PULL_OFFSET_MOVED};
         PullStatus[] respStatus = new PullStatus[] {PullStatus.FOUND, PullStatus.NO_NEW_MSG, PullStatus.NO_MATCHED_MSG, PullStatus.OFFSET_ILLEGAL};
         for (int i = 0; i < respCodes.length; i++) {
-            CompletableFuture<ResponseFuture> future = new CompletableFuture<>();
-            doReturn(future).when(mockClient).invokeImpl(any(Channel.class), any(RemotingCommand.class), anyLong());
             RemotingCommand response = mockPullMessageResponse(respCodes[i]);
-            ResponseFuture responseFuture = new ResponseFuture(channel, 0, null, 1000,
-                    resp -> { }, new SemaphoreReleaseOnlyOnce(new Semaphore(1)));
-            responseFuture.setResponseCommand(response);
-            promise.trySuccess(null);
-            future.complete(responseFuture);
+            Mockito.doAnswer(invocation -> {
+                InvokeCallback callback = invocation.getArgument(3);
+                callback.operationSucceed(response);
+                return null;
+            }).when(mockClient).invokeAsync(anyString(), any(RemotingCommand.class), anyLong(), any(InvokeCallback.class));
 
             Triple<PullResult, String, Boolean> rst = api.pullMessageFromSpecificBrokerAsync("", "", "", "", 1, 1, 1, 3000L).join();
             Assert.assertEquals(respStatus[i], rst.getLeft().getPullStatus());

client/src/test/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumerTest.java

@@ -75,7 +75,6 @@
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.nullable;
-import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.spy;
@@ -727,7 +726,7 @@ private void initDefaultLitePullConsumer(DefaultLitePullConsumer litePullConsume
         field.setAccessible(true);
         field.set(litePullConsumerImpl, offsetStore);
 
-        when(mQClientFactory.getMQClientAPIImpl().pullMessage(anyString(), any(PullMessageRequestHeader.class),
+        when(mQClientAPIImpl.pullMessage(anyString(), any(PullMessageRequestHeader.class),
             anyLong(), any(CommunicationMode.class), nullable(PullCallback.class)))
             .thenAnswer(new Answer<PullResult>() {
                 @Override
@@ -746,9 +745,11 @@ public PullResult answer(InvocationOnMock mock) throws Throwable {
                 }
             });
 
-        doAnswer(x -> new FindBrokerResult("127.0.0.1:10911", false)).when(mQClientFactory).findBrokerAddressInSubscribe(anyString(), anyLong(), anyBoolean());
+        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());
+        String clientId = mQClientFactory.getClientId();
+        doReturn(Collections.singletonList(clientId)).when(mQClientFactory).findConsumerIdList(anyString(), anyString());
 
         doReturn(123L).when(offsetStore).readOffset(any(MessageQueue.class), any(ReadOffsetType.class));
     }
@@ -787,7 +788,7 @@ private void initDefaultLitePullConsumerWithTag(DefaultLitePullConsumer litePull
         field.setAccessible(true);
         field.set(litePullConsumerImpl, offsetStore);
 
-        when(mQClientFactory.getMQClientAPIImpl().pullMessage(anyString(), any(PullMessageRequestHeader.class),
+        when(mQClientAPIImpl.pullMessage(anyString(), any(PullMessageRequestHeader.class),
             anyLong(), any(CommunicationMode.class), nullable(PullCallback.class)))
             .thenAnswer(new Answer<PullResult>() {
                 @Override
@@ -807,7 +808,8 @@ public PullResult answer(InvocationOnMock mock) throws Throwable {
                 }
             });
 
-        when(mQClientFactory.findBrokerAddressInSubscribe(anyString(), anyLong(), anyBoolean())).thenReturn(new FindBrokerResult("127.0.0.1:10911", false));
+        doReturn(new FindBrokerResult("127.0.0.1:10911", false))
+            .when(mQClientFactory).findBrokerAddressInSubscribe(anyString(), anyLong(), anyBoolean());
 
         doReturn(123L).when(offsetStore).readOffset(any(MessageQueue.class), any(ReadOffsetType.class));
     }

store/src/test/java/org/apache/rocketmq/store/rocksdb/RocksDBOptionsFactoryTest.java

@@ -52,8 +52,7 @@ public void testConsumeQueueUniversalCompactionMaxSizeAmplificationPercent() {
         MessageStore messageStore = mock(MessageStore.class);
         when(messageStore.getMessageStoreConfig()).thenReturn(config);
 
-        ConsumeQueueCompactionFilterFactory compactionFilterFactory = new ConsumeQueueCompactionFilterFactory(() -> 0);
-        try (ColumnFamilyOptions options = RocksDBOptionsFactory.createCQCFOptions(messageStore, compactionFilterFactory);
+        try (ColumnFamilyOptions options = RocksDBOptionsFactory.createCQCFOptions(messageStore);
             CompactionOptionsUniversal compactionOptions = options.compactionOptionsUniversal()) {
             Assert.assertEquals(50, compactionOptions.maxSizeAmplificationPercent());
         }