Skip to content

Commit

Permalink
Update PublisherImplTest.java
Browse files Browse the repository at this point in the history
  • Loading branch information
hannahrogers-google authored Jan 12, 2024
1 parent 88a22eb commit cc32d5b
Showing 1 changed file with 0 additions and 90 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -511,96 +511,6 @@ public void testEnableMessageOrdering_overwritesMaxAttempts() throws Exception {
assertTrue(publisher.awaitTermination(10, TimeUnit.SECONDS));
}

@Test
/**
* Make sure that resume publishing works as expected:
*
* <ol>
* <li>publish with key orderA which returns a failure.
* <li>publish with key orderA again, which should fail immediately
* <li>publish with key orderB, which should succeed
* <li>resume publishing on key orderA
* <li>publish with key orderA, which should now succeed
* </ol>
*/
public void testResumePublish() throws Exception {
Publisher publisher =
getTestPublisherBuilder()
.setBatchingSettings(
Publisher.Builder.DEFAULT_BATCHING_SETTINGS
.toBuilder()
.setElementCountThreshold(2L)
.build())
.setEnableMessageOrdering(true)
.build();

ApiFuture<String> future1 = sendTestMessageWithOrderingKey(publisher, "m1", "orderA");
ApiFuture<String> future2 = sendTestMessageWithOrderingKey(publisher, "m2", "orderA");

fakeExecutor.advanceTime(Duration.ZERO);
assertFalse(future1.isDone());
assertFalse(future2.isDone());

// This exception should stop future publishing to the same key
testPublisherServiceImpl.addPublishError(new StatusException(Status.INVALID_ARGUMENT));

fakeExecutor.advanceTime(Duration.ZERO);

try {
future1.get();
Assert.fail("This should fail.");
} catch (ExecutionException e) {
}

try {
future2.get();
Assert.fail("This should fail.");
} catch (ExecutionException e) {
}

// Submit new requests with orderA that should fail.
ApiFuture<String> future3 = sendTestMessageWithOrderingKey(publisher, "m3", "orderA");
ApiFuture<String> future4 = sendTestMessageWithOrderingKey(publisher, "m4", "orderA");

try {
future3.get();
Assert.fail("This should fail.");
} catch (ExecutionException e) {
assertEquals(SequentialExecutorService.CallbackExecutor.CANCELLATION_EXCEPTION, e.getCause());
}

try {
future4.get();
Assert.fail("This should fail.");
} catch (ExecutionException e) {
assertEquals(SequentialExecutorService.CallbackExecutor.CANCELLATION_EXCEPTION, e.getCause());
}

// Submit a new request with orderB, which should succeed
ApiFuture<String> future5 = sendTestMessageWithOrderingKey(publisher, "m5", "orderB");
ApiFuture<String> future6 = sendTestMessageWithOrderingKey(publisher, "m6", "orderB");

testPublisherServiceImpl.addPublishResponse(
PublishResponse.newBuilder().addMessageIds("5").addMessageIds("6"));

Assert.assertEquals("5", future5.get());
Assert.assertEquals("6", future6.get());

// Resume publishing of "orderA", which should now succeed
publisher.resumePublish("orderA");

ApiFuture<String> future7 = sendTestMessageWithOrderingKey(publisher, "m7", "orderA");
ApiFuture<String> future8 = sendTestMessageWithOrderingKey(publisher, "m8", "orderA");

testPublisherServiceImpl.addPublishResponse(
PublishResponse.newBuilder().addMessageIds("7").addMessageIds("8"));

Assert.assertEquals("7", future7.get());
Assert.assertEquals("8", future8.get());

shutdownTestPublisher(publisher);
}

@Test
public void testErrorPropagation() throws Exception {
Publisher publisher =
Expand Down

0 comments on commit cc32d5b

Please sign in to comment.