Skip to content

[ISSUE #11174] Restore receipt handles when proxy ACK fails #11175

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zmuxuny wants to merge 1 commit into
base: develop
Choose a base branch
from
Open

[ISSUE #11174] Restore receipt handles when proxy ACK fails #11175

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
103 changes: 78 additions & 25 deletions proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/AckMessageActivity.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,23 +70,35 @@ public CompletableFuture<AckMessageResponse> ackMessage(ProxyContext ctx, AckMes

protected CompletableFuture<AckMessageResponse> ackMessageInBatch(ProxyContext ctx, String group, String topic, AckMessageRequest request) {
List<ReceiptHandleMessage> handleMessageList = new ArrayList<>(request.getEntriesCount());
List<ReceiptHandleContext> handleContexts = new ArrayList<>(request.getEntriesCount());

for (AckMessageEntry ackMessageEntry : request.getEntriesList()) {
String handleString = getHandleString(ctx, group, request, ackMessageEntry);
handleMessageList.add(new ReceiptHandleMessage(ReceiptHandle.decode(handleString), ackMessageEntry.getMessageId()));
try {
for (AckMessageEntry ackMessageEntry : request.getEntriesList()) {
ReceiptHandleContext handleContext = getReceiptHandleContext(ctx, group, ackMessageEntry);
handleContexts.add(handleContext);
handleMessageList.add(new ReceiptHandleMessage(ReceiptHandle.decode(handleContext.handleString), ackMessageEntry.getMessageId()));
}
return this.messagingProcessor.batchAckMessage(ctx, handleMessageList, group, topic)
.whenComplete((batchAckResultList, throwable) -> {
if (throwable != null) {
restoreReceiptHandles(ctx, group, request, handleContexts);
}
})
.thenApply(batchAckResultList -> {
AckMessageResponse.Builder responseBuilder = AckMessageResponse.newBuilder();
Set<Code> responseCodes = new HashSet<>();
for (BatchAckResult batchAckResult : batchAckResultList) {
AckMessageResultEntry entry = convertToAckMessageResultEntry(batchAckResult);
responseBuilder.addEntries(entry);
responseCodes.add(entry.getStatus().getCode());
}
setAckResponseStatus(responseBuilder, responseCodes);
return responseBuilder.build();
});
} catch (Throwable t) {
restoreReceiptHandles(ctx, group, request, handleContexts);
throw t;
}
return this.messagingProcessor.batchAckMessage(ctx, handleMessageList, group, topic)
.thenApply(batchAckResultList -> {
AckMessageResponse.Builder responseBuilder = AckMessageResponse.newBuilder();
Set<Code> responseCodes = new HashSet<>();
for (BatchAckResult batchAckResult : batchAckResultList) {
AckMessageResultEntry entry = convertToAckMessageResultEntry(batchAckResult);
responseBuilder.addEntries(entry);
responseCodes.add(entry.getStatus().getCode());
}
setAckResponseStatus(responseBuilder, responseCodes);
return responseBuilder.build();
});
}

protected AckMessageResultEntry convertToAckMessageResultEntry(BatchAckResult batchAckResult) {
Expand Down Expand Up @@ -138,23 +150,28 @@ protected CompletableFuture<AckMessageResultEntry> processAckMessage(ProxyContex
AckMessageEntry ackMessageEntry) {
CompletableFuture<AckMessageResultEntry> future = new CompletableFuture<>();

ReceiptHandleContext handleContext = null;
try {
String handleString = this.getHandleString(ctx, group, request, ackMessageEntry);
handleContext = getReceiptHandleContext(ctx, group, ackMessageEntry);
CompletableFuture<AckResult> ackResultFuture = this.messagingProcessor.ackMessage(
ctx,
ReceiptHandle.decode(handleString),
ReceiptHandle.decode(handleContext.handleString),
ackMessageEntry.getMessageId(),
group,
topic,
ackMessageEntry.hasLiteTopic() ? ackMessageEntry.getLiteTopic() : null
);
ackResultFuture.thenAccept(result -> {
future.complete(convertToAckMessageResultEntry(ctx, ackMessageEntry, result));
}).exceptionally(t -> {
future.complete(convertToAckMessageResultEntry(ctx, ackMessageEntry, t));
return null;
ReceiptHandleContext removedHandleContext = handleContext;
ackResultFuture.whenComplete((result, throwable) -> {
if (throwable != null) {
restoreReceiptHandle(ctx, group, ackMessageEntry.getMessageId(), removedHandleContext);
future.complete(convertToAckMessageResultEntry(ctx, ackMessageEntry, throwable));
} else {
future.complete(convertToAckMessageResultEntry(ctx, ackMessageEntry, result));
}
});
} catch (Throwable t) {
restoreReceiptHandle(ctx, group, ackMessageEntry.getMessageId(), handleContext);
future.complete(convertToAckMessageResultEntry(ctx, ackMessageEntry, t));
}
return future;
Expand Down Expand Up @@ -195,15 +212,51 @@ protected void setAckResponseStatus(AckMessageResponse.Builder responseBuilder,
}
}

protected String getHandleString(ProxyContext ctx, String group, AckMessageRequest request, AckMessageEntry ackMessageEntry) {
protected ReceiptHandleContext getReceiptHandleContext(ProxyContext ctx, String group, AckMessageEntry ackMessageEntry) {
String handleString = ackMessageEntry.getReceiptHandle();
GrpcClientChannel channel = grpcChannelManager.getChannel(ctx.getClientID());
MessageReceiptHandle messageReceiptHandle = null;
if (channel != null) {
MessageReceiptHandle messageReceiptHandle = messagingProcessor.removeReceiptHandle(ctx, channel, group, ackMessageEntry.getMessageId(), ackMessageEntry.getReceiptHandle());
messageReceiptHandle = messagingProcessor.removeReceiptHandle(ctx, channel, group,
ackMessageEntry.getMessageId(), ackMessageEntry.getReceiptHandle());
if (messageReceiptHandle != null) {
handleString = messageReceiptHandle.getReceiptHandleStr();
}
}
return handleString;
return new ReceiptHandleContext(handleString, channel, messageReceiptHandle);
}

protected void restoreReceiptHandles(ProxyContext ctx, String group, AckMessageRequest request,
List<ReceiptHandleContext> handleContexts) {
for (int i = 0; i < handleContexts.size(); i++) {
restoreReceiptHandle(ctx, group, request.getEntries(i).getMessageId(), handleContexts.get(i));
}
}

protected void restoreReceiptHandle(ProxyContext ctx, String group, String messageId,
ReceiptHandleContext handleContext) {
if (handleContext == null || handleContext.channel == null || handleContext.messageReceiptHandle == null) {
return;
}
try {
messagingProcessor.addReceiptHandle(ctx, handleContext.channel, group, messageId,
handleContext.messageReceiptHandle);
} catch (Throwable restoreError) {
log.error("Failed to restore receipt handle after ack failure, group={}, messageId={}",
group, messageId, restoreError);
}
}

protected static class ReceiptHandleContext {
private final String handleString;
private final GrpcClientChannel channel;
private final MessageReceiptHandle messageReceiptHandle;

protected ReceiptHandleContext(String handleString, GrpcClientChannel channel,
MessageReceiptHandle messageReceiptHandle) {
this.handleString = handleString;
this.channel = channel;
this.messageReceiptHandle = messageReceiptHandle;
}
}
}
93 changes: 93 additions & 0 deletions proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/AckMessageActivityTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,22 +30,27 @@
import java.util.concurrent.CompletableFuture;
import org.apache.rocketmq.client.consumer.AckResult;
import org.apache.rocketmq.client.consumer.AckStatus;
import org.apache.rocketmq.proxy.common.MessageReceiptHandle;
import org.apache.rocketmq.proxy.common.ProxyException;
import org.apache.rocketmq.proxy.common.ProxyExceptionCode;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.grpc.v2.BaseActivityTest;
import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcClientChannel;
import org.apache.rocketmq.proxy.processor.BatchAckResult;
import org.apache.rocketmq.proxy.service.message.ReceiptHandleMessage;
import org.junit.Before;
import org.junit.Test;
import org.mockito.stubbing.Answer;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

public class AckMessageActivityTest extends BaseActivityTest {
Expand Down Expand Up @@ -153,6 +158,94 @@ public void testAckMessage() throws Throwable {
}
}

@Test
public void testAckMessageRestoresManagedHandleWhenAckFutureFailsTest() throws Throwable {
ConfigurationManager.getProxyConfig().setEnableBatchAck(false);
String messageId = "msg-retry";
String clientHandle = buildReceiptHandle(TOPIC, System.currentTimeMillis(), 3000);
String latestHandle = buildReceiptHandle(TOPIC, System.currentTimeMillis(), 6000);
GrpcClientChannel channel = grpcChannelManager.createChannel(createContext(), CLIENT_ID);
MessageReceiptHandle managed = new MessageReceiptHandle(GROUP, TOPIC, 0, latestHandle, messageId, 0, 0);
when(messagingProcessor.removeReceiptHandle(any(), eq(channel), eq(GROUP), eq(messageId), eq(clientHandle)))
.thenReturn(managed);
CompletableFuture<AckResult> failed = new CompletableFuture<>();
failed.completeExceptionally(new RuntimeException("transient ack failure"));
when(messagingProcessor.ackMessage(any(), any(), eq(messageId), eq(GROUP), eq(TOPIC), any())).thenReturn(failed);

AckMessageResponse response = ackMessageActivity.ackMessage(createContext(), AckMessageRequest.newBuilder()
.setTopic(Resource.newBuilder().setName(TOPIC)).setGroup(Resource.newBuilder().setName(GROUP))
.addEntries(AckMessageEntry.newBuilder().setMessageId(messageId).setReceiptHandle(clientHandle)).build()).get();

assertEquals(Code.INTERNAL_SERVER_ERROR, response.getStatus().getCode());
verify(messagingProcessor).addReceiptHandle(any(), eq(channel), eq(GROUP), eq(messageId), eq(managed));
}

@Test
public void testAckMessageRestoresManagedHandleWhenAckThrowsTest() throws Throwable {
ConfigurationManager.getProxyConfig().setEnableBatchAck(false);
String messageId = "msg-sync-failure";
String clientHandle = buildReceiptHandle(TOPIC, System.currentTimeMillis(), 3000);
String latestHandle = buildReceiptHandle(TOPIC, System.currentTimeMillis(), 6000);
GrpcClientChannel channel = grpcChannelManager.createChannel(createContext(), CLIENT_ID);
MessageReceiptHandle managed = new MessageReceiptHandle(GROUP, TOPIC, 0, latestHandle, messageId, 0, 0);
when(messagingProcessor.removeReceiptHandle(any(), eq(channel), eq(GROUP), eq(messageId), eq(clientHandle))).thenReturn(managed);
when(messagingProcessor.ackMessage(any(), any(), eq(messageId), eq(GROUP), eq(TOPIC), any()))
.thenThrow(new RuntimeException("sync ack failure"));

AckMessageResponse response = ackMessageActivity.ackMessage(createContext(), AckMessageRequest.newBuilder()
.setTopic(Resource.newBuilder().setName(TOPIC)).setGroup(Resource.newBuilder().setName(GROUP))
.addEntries(AckMessageEntry.newBuilder().setMessageId(messageId).setReceiptHandle(clientHandle)).build()).get();

assertEquals(Code.INTERNAL_SERVER_ERROR, response.getStatus().getCode());
verify(messagingProcessor).addReceiptHandle(any(), eq(channel), eq(GROUP), eq(messageId), eq(managed));
}

@Test
public void testBatchAckRestoresManagedHandleWhenFutureFailsTest() throws Throwable {
ConfigurationManager.getProxyConfig().setEnableBatchAck(true);
String messageId = "msg-batch-failure";
String clientHandle = buildReceiptHandle(TOPIC, System.currentTimeMillis(), 3000);
String latestHandle = buildReceiptHandle(TOPIC, System.currentTimeMillis(), 6000);
GrpcClientChannel channel = grpcChannelManager.createChannel(createContext(), CLIENT_ID);
MessageReceiptHandle managed = new MessageReceiptHandle(GROUP, TOPIC, 0, latestHandle, messageId, 0, 0);
when(messagingProcessor.removeReceiptHandle(any(), eq(channel), eq(GROUP), eq(messageId), eq(clientHandle))).thenReturn(managed);
CompletableFuture<List<BatchAckResult>> failed = new CompletableFuture<>();
failed.completeExceptionally(new RuntimeException("batch ack failure"));
when(messagingProcessor.batchAckMessage(any(), anyList(), eq(GROUP), eq(TOPIC))).thenReturn(failed);

try {
ackMessageActivity.ackMessage(createContext(), AckMessageRequest.newBuilder()
.setTopic(Resource.newBuilder().setName(TOPIC)).setGroup(Resource.newBuilder().setName(GROUP))
.addEntries(AckMessageEntry.newBuilder().setMessageId(messageId).setReceiptHandle(clientHandle)).build()).get();
fail("batch ack should complete exceptionally");
} catch (Exception expected) {
// Expected: batch invocation preserves its existing exceptional-completion semantics.
}
verify(messagingProcessor).addReceiptHandle(any(), eq(channel), eq(GROUP), eq(messageId), eq(managed));
}

@Test
public void testAckMessageDoesNotRestoreManagedHandleWhenAckSucceedsTest() throws Throwable {
ConfigurationManager.getProxyConfig().setEnableBatchAck(false);
String messageId = "msg-success";
String clientHandle = buildReceiptHandle(TOPIC, System.currentTimeMillis(), 3000);
String latestHandle = buildReceiptHandle(TOPIC, System.currentTimeMillis(), 6000);
GrpcClientChannel channel = grpcChannelManager.createChannel(createContext(), CLIENT_ID);
MessageReceiptHandle managed = new MessageReceiptHandle(GROUP, TOPIC, 0, latestHandle, messageId, 0, 0);
when(messagingProcessor.removeReceiptHandle(any(), eq(channel), eq(GROUP), eq(messageId), eq(clientHandle))).thenReturn(managed);
AckResult ackResult = new AckResult();
ackResult.setStatus(AckStatus.OK);
when(messagingProcessor.ackMessage(any(), any(), eq(messageId), eq(GROUP), eq(TOPIC), any()))
.thenReturn(CompletableFuture.completedFuture(ackResult));

AckMessageResponse response = ackMessageActivity.ackMessage(createContext(), AckMessageRequest.newBuilder()
.setTopic(Resource.newBuilder().setName(TOPIC)).setGroup(Resource.newBuilder().setName(GROUP))
.addEntries(AckMessageEntry.newBuilder().setMessageId(messageId).setReceiptHandle(clientHandle)).build()).get();

assertEquals(Code.OK, response.getStatus().getCode());
verify(messagingProcessor, never()).addReceiptHandle(any(), any(), anyString(), anyString(), any());
}

@Test
public void testAckMessageInBatch() throws Throwable {
ConfigurationManager.getProxyConfig().setEnableBatchAck(true);
Expand Down