[ISSUE #9187] The request should be rejected if the queueOffset equals maxOffset when changing the invisible time (#9186)

* When changing the invisible time, the request should be rejected if the queueOffset equals maxOffset

* Add UTs

* Make UTs to pass

* Remove unused imports

Author: RongtongJin

broker/src/main/java/org/apache/rocketmq/broker/processor/ChangeInvisibleTimeProcessor.java

@@ -130,7 +130,7 @@ public CompletableFuture<RemotingCommand> processRequestAsync(final Channel chan
         } catch (ConsumeQueueException e) {
             throw new RemotingCommandException("Failed to get max consume offset", e);
         }
-        if (requestHeader.getOffset() < minOffset || requestHeader.getOffset() > maxOffset) {
+        if (requestHeader.getOffset() < minOffset || requestHeader.getOffset() >= maxOffset) {
             response.setCode(ResponseCode.NO_MESSAGE);
             return CompletableFuture.completedFuture(response);
         }

broker/src/test/java/org/apache/rocketmq/broker/processor/ChangeInvisibleTimeProcessorTest.java

@@ -29,8 +29,6 @@
 import org.apache.rocketmq.common.message.MessageConst;
 import org.apache.rocketmq.common.message.MessageExtBrokerInner;
 import org.apache.rocketmq.remoting.exception.RemotingCommandException;
-import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
-import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
 import org.apache.rocketmq.remoting.netty.NettyClientConfig;
 import org.apache.rocketmq.remoting.netty.NettyServerConfig;
 import org.apache.rocketmq.remoting.protocol.LanguageCode;
@@ -46,6 +44,7 @@
 import org.apache.rocketmq.store.PutMessageResult;
 import org.apache.rocketmq.store.PutMessageStatus;
 import org.apache.rocketmq.store.config.MessageStoreConfig;
+import org.apache.rocketmq.store.exception.ConsumeQueueException;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -56,6 +55,8 @@
 import static org.apache.rocketmq.broker.processor.PullMessageProcessorTest.createConsumerData;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -108,7 +109,8 @@ public void init() throws IllegalAccessException, NoSuchFieldException {
     }
 
     @Test
-    public void testProcessRequest_Success() throws RemotingCommandException, InterruptedException, RemotingTimeoutException, RemotingSendRequestException {
+    public void testProcessRequest_Success() throws RemotingCommandException, ConsumeQueueException {
+        when(messageStore.getMaxOffsetInQueue(anyString(), anyInt())).thenReturn(2L);
         when(escapeBridge.asyncPutMessageToSpecificQueue(any(MessageExtBrokerInner.class))).thenReturn(CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.PUT_OK, new AppendMessageResult(AppendMessageStatus.PUT_OK))));
         int queueId = 0;
         long queueOffset = 0;
@@ -133,4 +135,31 @@ public void testProcessRequest_Success() throws RemotingCommandException, Interr
         assertThat(responseToReturn.getCode()).isEqualTo(ResponseCode.SUCCESS);
         assertThat(responseToReturn.getOpaque()).isEqualTo(request.getOpaque());
     }
+
+    @Test
+    public void testProcessRequest_NoMessage() throws RemotingCommandException, ConsumeQueueException {
+        when(messageStore.getMaxOffsetInQueue(anyString(), anyInt())).thenReturn(2L);
+        int queueId = 0;
+        long queueOffset = 2;
+        long popTime = System.currentTimeMillis() - 1_000;
+        long invisibleTime = 30_000;
+        int reviveQid = 0;
+        String brokerName = "test_broker";
+        String extraInfo = ExtraInfoUtil.buildExtraInfo(queueOffset, popTime, invisibleTime, reviveQid,
+            topic, brokerName, queueId) + MessageConst.KEY_SEPARATOR + queueOffset;
+
+        ChangeInvisibleTimeRequestHeader requestHeader = new ChangeInvisibleTimeRequestHeader();
+        requestHeader.setTopic(topic);
+        requestHeader.setQueueId(queueId);
+        requestHeader.setOffset(queueOffset);
+        requestHeader.setConsumerGroup(group);
+        requestHeader.setExtraInfo(extraInfo);
+        requestHeader.setInvisibleTime(invisibleTime);
+
+        final RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.CHANGE_MESSAGE_INVISIBLETIME, requestHeader);
+        request.makeCustomHeaderToNet();
+        RemotingCommand responseToReturn = changeInvisibleTimeProcessor.processRequest(handlerContext, request);
+        assertThat(responseToReturn.getCode()).isEqualTo(ResponseCode.NO_MESSAGE);
+        assertThat(responseToReturn.getOpaque()).isEqualTo(request.getOpaque());
+    }
 }