[ISSUE #9583] Fix ack not working after changeInvisibleDuration in Proxy

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ChangeInvisibleDurationActivity.java

@@ -75,9 +75,25 @@ public CompletableFuture<ChangeInvisibleDurationResponse> changeInvisibleDuratio
     protected ChangeInvisibleDurationResponse convertToChangeInvisibleDurationResponse(ProxyContext ctx,
         ChangeInvisibleDurationRequest request, AckResult ackResult) {
         if (AckStatus.OK.equals(ackResult.getStatus())) {
+            String newHandleStr = ackResult.getExtraInfo();
+            if (newHandleStr != null) {
+                try {
+                    ReceiptHandle newHandle = ReceiptHandle.decode(newHandleStr);
+                    String group = request.getGroup().getName();
+                    String topic = request.getTopic().getName();
+                    MessageReceiptHandle newMessageReceiptHandle = new MessageReceiptHandle(
+                        group, topic, newHandle.getQueueId(), newHandleStr,
+                        request.getMessageId(), newHandle.getOffset(), 0);
+                    messagingProcessor.addReceiptHandle(ctx,
+                        grpcChannelManager.getChannel(ctx.getClientID()),
+                        group, request.getMessageId(), newMessageReceiptHandle);
+                } catch (Exception e) {
+                    // log but do not fail the response
+                }
+            }
             return ChangeInvisibleDurationResponse.newBuilder()
                 .setStatus(ResponseBuilder.getInstance().buildStatus(Code.OK, Code.OK.name()))
-                .setReceiptHandle(ackResult.getExtraInfo())
+                .setReceiptHandle(newHandleStr)
                 .build();
         }
         return ChangeInvisibleDurationResponse.newBuilder()

proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ChangeInvisibleDurationActivityTest.java

@@ -40,6 +40,8 @@
 import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 public class ChangeInvisibleDurationActivityTest extends BaseActivityTest {
@@ -181,6 +183,34 @@ public void testChangeInvisibleDurationInvisibleTimeTooSmall() throws Throwable
         }
     }
 
+    @Test
+    public void testNewHandleIsReRegisteredAfterSuccess() throws Throwable {
+        long newPopTime = System.currentTimeMillis();
+        long newInvisibleTime = 5000;
+        String newHandleStr = buildReceiptHandle(TOPIC, newPopTime, newInvisibleTime);
+        AckResult ackResult = new AckResult();
+        ackResult.setExtraInfo(newHandleStr);
+        ackResult.setStatus(AckStatus.OK);
+        when(this.messagingProcessor.changeInvisibleTime(
+            any(), any(), anyString(), anyString(), anyString(), anyLong(), anyString(), anyLong(), anyBoolean()
+        )).thenReturn(CompletableFuture.completedFuture(ackResult));
+
+        String msgId = "testMsgId";
+        ChangeInvisibleDurationResponse response = this.changeInvisibleDurationActivity.changeInvisibleDuration(
+            createContext(),
+            ChangeInvisibleDurationRequest.newBuilder()
+                .setInvisibleDuration(Durations.fromMillis(newInvisibleTime))
+                .setTopic(Resource.newBuilder().setName(TOPIC).build())
+                .setGroup(Resource.newBuilder().setName(CONSUMER_GROUP).build())
+                .setMessageId(msgId)
+                .setReceiptHandle(buildReceiptHandle(TOPIC, System.currentTimeMillis() - 10000, 3000))
+                .build()
+        ).get();
+
+        assertEquals(Code.OK, response.getStatus().getCode());
+        verify(messagingProcessor).addReceiptHandle(any(), any(), eq(CONSUMER_GROUP), eq(msgId), any(MessageReceiptHandle.class));
+    }
+
     @Test
     public void testChangeInvisibleDurationInvisibleTimeTooLarge() throws Throwable {
         try {