diff --git a/.github/blunderbuss.yml b/.github/blunderbuss.yml index 44ab091e1..97998e463 100644 --- a/.github/blunderbuss.yml +++ b/.github/blunderbuss.yml @@ -1,9 +1,9 @@ # Configuration for the Blunderbuss GitHub app. For more info see # https://github.com/googleapis/repo-automation-bots/tree/main/packages/blunderbuss assign_issues: - - maitrimangal + - michaelpri10 assign_prs: - - maitrimangal + - michaelpri10 assign_prs_by: - labels: - samples diff --git a/CHANGELOG.md b/CHANGELOG.md index 8721be429..92946f184 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ # Changelog +## [1.126.4](https://github.com/googleapis/java-pubsub/compare/v1.126.3...v1.126.4) (2024-02-09) + + +### Bug Fixes + +* Message ordering fix for [#1889](https://github.com/googleapis/java-pubsub/issues/1889) ([#1903](https://github.com/googleapis/java-pubsub/issues/1903)) ([22a87c6](https://github.com/googleapis/java-pubsub/commit/22a87c67f07b55266e277f83f5ceb17d9f32f67e)) + ## [1.126.3](https://github.com/googleapis/java-pubsub/compare/v1.126.2...v1.126.3) (2024-02-08) diff --git a/README.md b/README.md index 68423c089..89680132d 100644 --- a/README.md +++ b/README.md @@ -44,7 +44,7 @@ If you are using Maven without the BOM, add this to your dependencies: com.google.cloud google-cloud-pubsub - 1.126.2 + 1.126.3 ``` @@ -59,13 +59,13 @@ implementation 'com.google.cloud:google-cloud-pubsub' If you are using Gradle without BOM, add this to your dependencies: ```Groovy -implementation 'com.google.cloud:google-cloud-pubsub:1.126.2' +implementation 'com.google.cloud:google-cloud-pubsub:1.126.3' ``` If you are using SBT, add this to your dependencies: ```Scala -libraryDependencies += "com.google.cloud" % "google-cloud-pubsub" % "1.126.2" +libraryDependencies += "com.google.cloud" % "google-cloud-pubsub" % "1.126.3" ``` @@ -409,7 +409,7 @@ Java is a registered trademark of Oracle and/or its affiliates. [kokoro-badge-link-5]: http://storage.googleapis.com/cloud-devrel-public/java/badges/java-pubsub/java11.html [stability-image]: https://img.shields.io/badge/stability-stable-green [maven-version-image]: https://img.shields.io/maven-central/v/com.google.cloud/google-cloud-pubsub.svg -[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-pubsub/1.126.2 +[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-pubsub/1.126.3 [authentication]: https://github.com/googleapis/google-cloud-java#authentication [auth-scopes]: https://developers.google.com/identity/protocols/oauth2/scopes [predefined-iam-roles]: https://cloud.google.com/iam/docs/understanding-roles#predefined_roles diff --git a/google-cloud-pubsub-bom/pom.xml b/google-cloud-pubsub-bom/pom.xml index cbf0b858c..795bf000c 100644 --- a/google-cloud-pubsub-bom/pom.xml +++ b/google-cloud-pubsub-bom/pom.xml @@ -3,7 +3,7 @@ 4.0.0 com.google.cloud google-cloud-pubsub-bom - 1.126.3 + 1.126.4 pom com.google.cloud @@ -52,17 +52,17 @@ com.google.cloud google-cloud-pubsub - 1.126.3 + 1.126.4 com.google.api.grpc grpc-google-cloud-pubsub-v1 - 1.108.3 + 1.108.4 com.google.api.grpc proto-google-cloud-pubsub-v1 - 1.108.3 + 1.108.4 diff --git a/google-cloud-pubsub/pom.xml b/google-cloud-pubsub/pom.xml index 650cb553f..cb705395d 100644 --- a/google-cloud-pubsub/pom.xml +++ b/google-cloud-pubsub/pom.xml @@ -3,7 +3,7 @@ 4.0.0 com.google.cloud google-cloud-pubsub - 1.126.3 + 1.126.4 jar Google Cloud Pub/Sub https://github.com/googleapis/java-pubsub @@ -11,7 +11,7 @@ com.google.cloud google-cloud-pubsub-parent - 1.126.3 + 1.126.4 google-cloud-pubsub diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java index 635bc92d5..1810badd2 100644 --- a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java +++ b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java @@ -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; @@ -92,8 +93,8 @@ class MessageDispatcher { private final LinkedBlockingQueue pendingAcks = new LinkedBlockingQueue<>(); private final LinkedBlockingQueue pendingNacks = new LinkedBlockingQueue<>(); private final LinkedBlockingQueue pendingReceipts = new LinkedBlockingQueue<>(); - private final ConcurrentMap outstandingReceipts = - new ConcurrentHashMap(); + private final LinkedHashMap outstandingReceipts = + new LinkedHashMap(); private final AtomicInteger messageDeadlineSeconds = new AtomicInteger(); private final AtomicBoolean extendDeadline = new AtomicBoolean(true); private final Lock jobLock; @@ -397,7 +398,9 @@ void processReceivedMessages(List 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. @@ -417,33 +420,36 @@ void processReceivedMessages(List messages) { } void notifyAckSuccess(AckRequestData ackRequestData) { - - if (outstandingReceipts.containsKey(ackRequestData.getAckId())) { - outstandingReceipts.get(ackRequestData.getAckId()).notifyReceiptComplete(); - List outstandingBatch = new ArrayList<>(); - - for (Iterator> it = - outstandingReceipts.entrySet().iterator(); - it.hasNext(); ) { - Map.Entry 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 outstandingBatch = new ArrayList<>(); + + for (Iterator> it = + outstandingReceipts.entrySet().iterator(); + it.hasNext(); ) { + Map.Entry 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 batch) { diff --git a/grpc-google-cloud-pubsub-v1/pom.xml b/grpc-google-cloud-pubsub-v1/pom.xml index 236f1fbe5..ab59fe1ad 100644 --- a/grpc-google-cloud-pubsub-v1/pom.xml +++ b/grpc-google-cloud-pubsub-v1/pom.xml @@ -4,13 +4,13 @@ 4.0.0 com.google.api.grpc grpc-google-cloud-pubsub-v1 - 1.108.3 + 1.108.4 grpc-google-cloud-pubsub-v1 GRPC library for grpc-google-cloud-pubsub-v1 com.google.cloud google-cloud-pubsub-parent - 1.126.3 + 1.126.4 diff --git a/pom.xml b/pom.xml index 321bef25a..6de969e6a 100644 --- a/pom.xml +++ b/pom.xml @@ -4,7 +4,7 @@ com.google.cloud google-cloud-pubsub-parent pom - 1.126.3 + 1.126.4 Google Cloud Pub/Sub Parent https://github.com/googleapis/java-pubsub @@ -69,17 +69,17 @@ com.google.api.grpc proto-google-cloud-pubsub-v1 - 1.108.3 + 1.108.4 com.google.api.grpc grpc-google-cloud-pubsub-v1 - 1.108.3 + 1.108.4 com.google.cloud google-cloud-pubsub - 1.126.3 + 1.126.4 diff --git a/proto-google-cloud-pubsub-v1/pom.xml b/proto-google-cloud-pubsub-v1/pom.xml index 6e94b2ee1..e06710a62 100644 --- a/proto-google-cloud-pubsub-v1/pom.xml +++ b/proto-google-cloud-pubsub-v1/pom.xml @@ -4,13 +4,13 @@ 4.0.0 com.google.api.grpc proto-google-cloud-pubsub-v1 - 1.108.3 + 1.108.4 proto-google-cloud-pubsub-v1 PROTO library for proto-google-cloud-pubsub-v1 com.google.cloud google-cloud-pubsub-parent - 1.126.3 + 1.126.4 diff --git a/samples/install-without-bom/pom.xml b/samples/install-without-bom/pom.xml index 8a23573ba..b6169a62c 100644 --- a/samples/install-without-bom/pom.xml +++ b/samples/install-without-bom/pom.xml @@ -44,7 +44,7 @@ com.google.cloud google-cloud-pubsub - 1.126.2 + 1.126.3 diff --git a/samples/snapshot/pom.xml b/samples/snapshot/pom.xml index b1e557e8b..adcd70028 100644 --- a/samples/snapshot/pom.xml +++ b/samples/snapshot/pom.xml @@ -43,7 +43,7 @@ com.google.cloud google-cloud-pubsub - 1.126.3 + 1.126.4 diff --git a/versions.txt b/versions.txt index 8cce4bec8..451bc8f07 100644 --- a/versions.txt +++ b/versions.txt @@ -1,6 +1,6 @@ # Format: # module:released-version:current-version -google-cloud-pubsub:1.126.3:1.126.3 -grpc-google-cloud-pubsub-v1:1.108.3:1.108.3 -proto-google-cloud-pubsub-v1:1.108.3:1.108.3 +google-cloud-pubsub:1.126.4:1.126.4 +grpc-google-cloud-pubsub-v1:1.108.4:1.108.4 +proto-google-cloud-pubsub-v1:1.108.4:1.108.4