Skip to content

Commit d816138

Browse files
kamalaboulhosngcf-owl-bot[bot]
kamalaboulhosn
and
gcf-owl-bot[bot]
authored
fix: Use message ordering enabled property that comes with streaming pull responses (googleapis#1851)
* samples: schema evolution * samples: schema evolution * Format fixes * Fix documentation for field. * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Add back in working asserts * Formatting fixes * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Version/delete fixes * samples: schema evolution * samples: schema evolution * Format fixes * Fix documentation for field. * Add back in working asserts * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Formatting fixes * Version/delete fixes * samples: Schema evolution (googleapis#1499) * samples: schema evolution * samples: schema evolution * Format fixes * Fix documentation for field. * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Add back in working asserts * Formatting fixes * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Version/delete fixes * samples: schema evolution * samples: schema evolution * Format fixes * Fix documentation for field. * Add back in working asserts * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Formatting fixes * Version/delete fixes --------- Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com> * Minor fixes for comments * samples: Schema evolution (googleapis#1499) * samples: schema evolution * samples: schema evolution * Format fixes * Fix documentation for field. * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Add back in working asserts * Formatting fixes * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Version/delete fixes * samples: schema evolution * samples: schema evolution * Format fixes * Fix documentation for field. * Add back in working asserts * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Formatting fixes * Version/delete fixes --------- Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com> * Fix rollback example * Formatting * Formatting and wording fixes * Add new schemas to test directory * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Samples: Fix exception handling * fix: Set x-goog-request-params for streaming pull request * Revert "fix: Set x-goog-request-params for streaming pull request" This reverts commit 3185a3e. * Revert "Revert "fix: Set x-goog-request-params for streaming pull request"" This reverts commit 3b1f4d9. * Thread example * Add examples for limited and unlimited exeuctors * Add back missing semicolon * Revert changes to original async example * Revert changes to original async example * Add examples of different threading models * Make variables final to conform to style. * Fix catches * Fix ids * Fix naming * Revert "Merge pull request googleapis#2 from kamalaboulhosn/ML_experiments" This reverts commit 5a435fa, reversing changes made to c3a5725. * Set blunderbuss config to auto-assign issues and PRs * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * fix: Swap writer and reader schema to correct places in sample * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * fix: Use message ordering enabled property that comes with streaming pull responses so that messages are only delivered to the callback one at a time in order when ordering is actually enabled --------- Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>
1 parent ca693b1 commit d816138

File tree

3 files changed

+124
-6
lines changed

3 files changed

+124
-6
lines changed

‎google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,7 @@ class MessageDispatcher {
8282
private final FlowController flowController;
8383

8484
private AtomicBoolean exactlyOnceDeliveryEnabled = new AtomicBoolean(false);
85+
private AtomicBoolean messageOrderingEnabled = new AtomicBoolean(false);
8586

8687
private final Waiter messagesWaiter;
8788

@@ -343,6 +344,11 @@ void setExactlyOnceDeliveryEnabled(boolean exactlyOnceDeliveryEnabled) {
343344
}
344345
}
345346

347+
@InternalApi
348+
void setMessageOrderingEnabled(boolean messageOrderingEnabled) {
349+
this.messageOrderingEnabled.set(messageOrderingEnabled);
350+
}
351+
346352
private static class OutstandingMessage {
347353
private final ReceivedMessage receivedMessage;
348354
private final AckHandler ackHandler;
@@ -506,7 +512,7 @@ public void run() {
506512
}
507513
}
508514
};
509-
if (message.getOrderingKey().isEmpty()) {
515+
if (!messageOrderingEnabled.get() || message.getOrderingKey().isEmpty()) {
510516
executor.execute(deliverMessageTask);
511517
} else {
512518
sequentialExecutor.submit(message.getOrderingKey(), deliverMessageTask);

‎google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -236,9 +236,12 @@ public void onResponse(StreamingPullResponse response) {
236236

237237
boolean exactlyOnceDeliveryEnabledResponse =
238238
response.getSubscriptionProperties().getExactlyOnceDeliveryEnabled();
239+
boolean messageOrderingEnabledResponse =
240+
response.getSubscriptionProperties().getMessageOrderingEnabled();
239241

240242
setExactlyOnceDeliveryEnabled(exactlyOnceDeliveryEnabledResponse);
241243
messageDispatcher.setExactlyOnceDeliveryEnabled(exactlyOnceDeliveryEnabledResponse);
244+
messageDispatcher.setMessageOrderingEnabled(messageOrderingEnabledResponse);
242245
messageDispatcher.processReceivedMessages(response.getReceivedMessagesList());
243246

244247
// Only request more if we're not shutdown.

‎google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/MessageDispatcherTest.java‎

Lines changed: 114 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -30,18 +30,41 @@
3030
import java.util.concurrent.*;
3131
import org.junit.Before;
3232
import org.junit.Test;
33+
import org.mockito.invocation.InvocationOnMock;
34+
import org.mockito.stubbing.Answer;
3335
import org.threeten.bp.Duration;
3436

3537
public class MessageDispatcherTest {
3638
private static final ByteString MESSAGE_DATA = ByteString.copyFromUtf8("message-data");
3739
private static final int DELIVERY_INFO_COUNT = 3;
3840
private static final String ACK_ID = "ACK-ID";
41+
private static final String ORDERING_KEY = "KEY";
3942
private static final ReceivedMessage TEST_MESSAGE =
4043
ReceivedMessage.newBuilder()
4144
.setAckId(ACK_ID)
4245
.setMessage(PubsubMessage.newBuilder().setData(MESSAGE_DATA).build())
4346
.setDeliveryAttempt(DELIVERY_INFO_COUNT)
4447
.build();
48+
private static final ByteString ORDERED_MESSAGE_DATA_1 = ByteString.copyFromUtf8("message-data1");
49+
private static final ReceivedMessage ORDERED_TEST_MESSAGE_1 =
50+
ReceivedMessage.newBuilder()
51+
.setAckId("ACK-ID-1")
52+
.setMessage(
53+
PubsubMessage.newBuilder()
54+
.setData(ORDERED_MESSAGE_DATA_1)
55+
.setOrderingKey(ORDERING_KEY)
56+
.build())
57+
.build();
58+
private static final ByteString ORDERED_MESSAGE_DATA_2 = ByteString.copyFromUtf8("message-data2");
59+
private static final ReceivedMessage ORDERED_TEST_MESSAGE_2 =
60+
ReceivedMessage.newBuilder()
61+
.setAckId("ACK-ID-2")
62+
.setMessage(
63+
PubsubMessage.newBuilder()
64+
.setData(ORDERED_MESSAGE_DATA_2)
65+
.setOrderingKey(ORDERING_KEY)
66+
.build())
67+
.build();
4568
private static final int MAX_SECONDS_PER_ACK_EXTENSION = 60;
4669
private static final int MIN_ACK_DEADLINE_SECONDS = 10;
4770
private static final Duration MAX_ACK_EXTENSION_PERIOD = Duration.ofMinutes(60);
@@ -494,6 +517,84 @@ public void testAckExtensionDefaultsExactlyOnceDeliveryEnabledThenDisabled() {
494517
Math.toIntExact(Subscriber.MAX_STREAM_ACK_DEADLINE.getSeconds()));
495518
}
496519

520+
@Test
521+
public void testOrderedDeliveryOrderingDisabled() throws Exception {
522+
MessageReceiver mockMessageReceiver = mock(MessageReceiver.class);
523+
MessageDispatcher messageDispatcher =
524+
getMessageDispatcher(mockMessageReceiver, Executors.newFixedThreadPool(5));
525+
526+
// This would normally be set from the streaming pull response in the
527+
// StreamingSubscriberConnection
528+
messageDispatcher.setMessageOrderingEnabled(false);
529+
530+
CountDownLatch receiveCalls = new CountDownLatch(2);
531+
532+
doAnswer(
533+
new Answer<Void>() {
534+
public Void answer(InvocationOnMock invocation) throws Exception {
535+
Thread.sleep(1000);
536+
receiveCalls.countDown();
537+
return null;
538+
}
539+
})
540+
.when(mockMessageReceiver)
541+
.receiveMessage(eq(ORDERED_TEST_MESSAGE_1.getMessage()), any(AckReplyConsumer.class));
542+
doAnswer(
543+
new Answer<Void>() {
544+
public Void answer(InvocationOnMock invocation) {
545+
// Ensure the previous method didn't finish and we could process in parallel.
546+
assertEquals(2, receiveCalls.getCount());
547+
receiveCalls.countDown();
548+
return null;
549+
}
550+
})
551+
.when(mockMessageReceiver)
552+
.receiveMessage(eq(ORDERED_TEST_MESSAGE_2.getMessage()), any(AckReplyConsumer.class));
553+
554+
messageDispatcher.processReceivedMessages(
555+
Arrays.asList(ORDERED_TEST_MESSAGE_1, ORDERED_TEST_MESSAGE_2));
556+
receiveCalls.await();
557+
}
558+
559+
@Test
560+
public void testOrderedDeliveryOrderingEnabled() throws Exception {
561+
MessageReceiver mockMessageReceiver = mock(MessageReceiver.class);
562+
MessageDispatcher messageDispatcher =
563+
getMessageDispatcher(mockMessageReceiver, Executors.newFixedThreadPool(5));
564+
565+
// This would normally be set from the streaming pull response in the
566+
// StreamingSubscriberConnection
567+
messageDispatcher.setMessageOrderingEnabled(true);
568+
569+
CountDownLatch receiveCalls = new CountDownLatch(2);
570+
571+
doAnswer(
572+
new Answer<Void>() {
573+
public Void answer(InvocationOnMock invocation) throws Exception {
574+
Thread.sleep(1000);
575+
receiveCalls.countDown();
576+
return null;
577+
}
578+
})
579+
.when(mockMessageReceiver)
580+
.receiveMessage(eq(ORDERED_TEST_MESSAGE_1.getMessage()), any(AckReplyConsumer.class));
581+
doAnswer(
582+
new Answer<Void>() {
583+
public Void answer(InvocationOnMock invocation) {
584+
// Ensure the previous method has finished completely.
585+
assertEquals(1, receiveCalls.getCount());
586+
receiveCalls.countDown();
587+
return null;
588+
}
589+
})
590+
.when(mockMessageReceiver)
591+
.receiveMessage(eq(ORDERED_TEST_MESSAGE_2.getMessage()), any(AckReplyConsumer.class));
592+
593+
messageDispatcher.processReceivedMessages(
594+
Arrays.asList(ORDERED_TEST_MESSAGE_1, ORDERED_TEST_MESSAGE_2));
595+
receiveCalls.await();
596+
}
597+
497598
@Test
498599
public void testAckExtensionCustomMinExactlyOnceDeliveryDisabledThenEnabled() {
499600
int customMinSeconds = 30;
@@ -569,20 +670,28 @@ private void assertMinAndMaxAckDeadlines(
569670
}
570671

571672
private MessageDispatcher getMessageDispatcher() {
572-
return getMessageDispatcher(mock(MessageReceiver.class));
673+
return getMessageDispatcher(mock(MessageReceiver.class), MoreExecutors.directExecutor());
573674
}
574675

575676
private MessageDispatcher getMessageDispatcher(MessageReceiver messageReceiver) {
576-
return getMessageDispatcherFromBuilder(MessageDispatcher.newBuilder(messageReceiver));
677+
return getMessageDispatcherFromBuilder(
678+
MessageDispatcher.newBuilder(messageReceiver), MoreExecutors.directExecutor());
679+
}
680+
681+
private MessageDispatcher getMessageDispatcher(
682+
MessageReceiver messageReceiver, Executor executor) {
683+
return getMessageDispatcherFromBuilder(MessageDispatcher.newBuilder(messageReceiver), executor);
577684
}
578685

579686
private MessageDispatcher getMessageDispatcher(
580687
MessageReceiverWithAckResponse messageReceiverWithAckResponse) {
581688
return getMessageDispatcherFromBuilder(
582-
MessageDispatcher.newBuilder(messageReceiverWithAckResponse));
689+
MessageDispatcher.newBuilder(messageReceiverWithAckResponse),
690+
MoreExecutors.directExecutor());
583691
}
584692

585-
private MessageDispatcher getMessageDispatcherFromBuilder(MessageDispatcher.Builder builder) {
693+
private MessageDispatcher getMessageDispatcherFromBuilder(
694+
MessageDispatcher.Builder builder, Executor executor) {
586695
MessageDispatcher messageDispatcher =
587696
builder
588697
.setAckProcessor(mockAckProcessor)
@@ -594,7 +703,7 @@ private MessageDispatcher getMessageDispatcherFromBuilder(MessageDispatcher.Buil
594703
.setMaxDurationPerAckExtensionDefaultUsed(true)
595704
.setAckLatencyDistribution(mock(Distribution.class))
596705
.setFlowController(mock(FlowController.class))
597-
.setExecutor(MoreExecutors.directExecutor())
706+
.setExecutor(executor)
598707
.setSystemExecutor(systemExecutor)
599708
.setApiClock(clock)
600709
.build();

0 commit comments

Comments
 (0)