Skip to content

fix: Message ordering fix for #1889 #1903

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

Merged
michaelpri10 merged 10 commits into from
Feb 9, 2024
Merged

fix: Message ordering fix for #1889 #1903

Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
b433095
chore: change assignees for issues and PRs to michaelpri10
michaelpri10 Feb 7, 2024
83b6ec3
Merge branch 'googleapis:main' into main
michaelpri10 Feb 7, 2024
b3f1dab
🦉 Updates from OwlBot post-processor
gcf-owl-bot[bot] Feb 8, 2024
7427cc0
fix: Revert PR#1807 and use a LinkedHasMap in the MessageDispatcher
michaelpri10 Feb 8, 2024
9fbb3a7
fix: Make processedReceivedMessages thread-safe
michaelpri10 Feb 8, 2024
962b1f9
Merge branch 'googleapis:main' into message-ordering-fix
michaelpri10 Feb 8, 2024
7ce6868
Merge branch 'googleapis:main' into main
michaelpri10 Feb 9, 2024
2a80a8b
Merge branch 'main' into message-ordering-fix
michaelpri10 Feb 9, 2024
b2fa5dd
fix: Only synchronize on the outstandingReceipts object in the Messag…
michaelpri10 Feb 9, 2024
05d8f0d
Merge branch 'message-ordering-fix' of https://github.com/michaelpri1…
michaelpri10 Feb 9, 2024
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
52 changes: 29 additions & 23 deletions google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
Expand Down Expand Up @@ -92,8 +93,8 @@ class MessageDispatcher {
private final LinkedBlockingQueue<AckRequestData> pendingAcks = new LinkedBlockingQueue<>();
private final LinkedBlockingQueue<AckRequestData> pendingNacks = new LinkedBlockingQueue<>();
private final LinkedBlockingQueue<AckRequestData> pendingReceipts = new LinkedBlockingQueue<>();
private final ConcurrentMap<String, ReceiptCompleteData> outstandingReceipts =
new ConcurrentHashMap<String, ReceiptCompleteData>();
private final LinkedHashMap<String, ReceiptCompleteData> outstandingReceipts =
new LinkedHashMap<String, ReceiptCompleteData>();
private final AtomicInteger messageDeadlineSeconds = new AtomicInteger();
private final AtomicBoolean extendDeadline = new AtomicBoolean(true);
private final Lock jobLock;
Expand Down Expand Up @@ -397,7 +398,9 @@ void processReceivedMessages(List<ReceivedMessage> messages) {
if (this.exactlyOnceDeliveryEnabled.get()) {
// For exactly once deliveries we don't add to outstanding batch because we first
// process the receipt modack. If that is successful then we process the message.
outstandingReceipts.put(message.getAckId(), new ReceiptCompleteData(outstandingMessage));
synchronized (outstandingReceipts) {
outstandingReceipts.put(message.getAckId(), new ReceiptCompleteData(outstandingMessage));
}
} else if (pendingMessages.putIfAbsent(message.getAckId(), ackHandler) != null) {
// putIfAbsent puts ackHandler if ackID isn't previously mapped, then return the
// previously-mapped element.
Expand All @@ -417,33 +420,36 @@ void processReceivedMessages(List<ReceivedMessage> messages) {
}

void notifyAckSuccess(AckRequestData ackRequestData) {

if (outstandingReceipts.containsKey(ackRequestData.getAckId())) {
outstandingReceipts.get(ackRequestData.getAckId()).notifyReceiptComplete();
List<OutstandingMessage> outstandingBatch = new ArrayList<>();

for (Iterator<Entry<String, ReceiptCompleteData>> it =
outstandingReceipts.entrySet().iterator();
it.hasNext(); ) {
Map.Entry<String, ReceiptCompleteData> receipt = it.next();
// If receipt is complete then add to outstandingBatch to process the batch
if (receipt.getValue().isReceiptComplete()) {
it.remove();
if (pendingMessages.putIfAbsent(
receipt.getKey(), receipt.getValue().getOutstandingMessage().ackHandler)
== null) {
outstandingBatch.add(receipt.getValue().getOutstandingMessage());
synchronized (outstandingReceipts) {
if (outstandingReceipts.containsKey(ackRequestData.getAckId())) {
outstandingReceipts.get(ackRequestData.getAckId()).notifyReceiptComplete();
List<OutstandingMessage> outstandingBatch = new ArrayList<>();

for (Iterator<Entry<String, ReceiptCompleteData>> it =
outstandingReceipts.entrySet().iterator();
it.hasNext(); ) {
Map.Entry<String, ReceiptCompleteData> receipt = it.next();
// If receipt is complete then add to outstandingBatch to process the batch
if (receipt.getValue().isReceiptComplete()) {
it.remove();
if (pendingMessages.putIfAbsent(
receipt.getKey(), receipt.getValue().getOutstandingMessage().ackHandler)
== null) {
outstandingBatch.add(receipt.getValue().getOutstandingMessage());
}
} else {
break;
}
} else {
break;
}
processBatch(outstandingBatch);
}
processBatch(outstandingBatch);
}
}

void notifyAckFailed(AckRequestData ackRequestData) {
outstandingReceipts.remove(ackRequestData.getAckId());
synchronized (outstandingReceipts) {
outstandingReceipts.remove(ackRequestData.getAckId());
}
}

private void processBatch(List<OutstandingMessage> batch) {
Expand Down