You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "mosche (via GitHub)" <gi...@apache.org> on 2023/07/20 10:37:05 UTC
[GitHub] [beam] mosche commented on a diff in pull request #27484: [AWS SQS] Support strict expiration of SQS batches when writing
mosche commented on code in PR #27484:
URL: https://github.com/apache/beam/pull/27484#discussion_r1269273093
##########
sdks/java/io/amazon-web-services2/src/test/java/org/apache/beam/sdk/io/aws2/sqs/SqsIOWriteBatchesTest.java:
##########
@@ -187,24 +237,53 @@ public void testWriteBatchesToDynamic() {
}
@Test
- public void testWriteBatchesWithTimeout() {
+ public void testWriteBatchesToDynamicWithTimeout() {
when(sqs.sendMessageBatch(anyRequest())).thenReturn(completedFuture(SUCCESS));
p.apply(Create.of(5))
.apply(ParDo.of(new CreateMessages()))
.apply(
// simulate delay between messages > batch timeout
- SqsIO.writeBatches(withDelay(millis(200), SET_MESSAGE_BODY))
- .withBatchTimeout(millis(100))
- .to("queue"));
+ SqsIO.writeBatches(withDelay(millis(100), SET_MESSAGE_BODY))
+ .withBatchTimeout(millis(150))
+ .to(msg -> Integer.valueOf(msg) % 2 == 0 ? "even" : "uneven"));
p.run().waitUntilFinish();
SendMessageBatchRequestEntry[] entries = entries(range(0, 5));
- // due to added delay, batches are timed out on arrival of every 2nd msg
- verify(sqs).sendMessageBatch(request("queue", entries[0], entries[1]));
- verify(sqs).sendMessageBatch(request("queue", entries[2], entries[3]));
- verify(sqs).sendMessageBatch(request("queue", entries[4]));
+ // due to added delay, dynamic batches are timed out on arrival of every 2nd msg (per batch)
+ verify(sqs).sendMessageBatch(request("even", entries[0], entries[2]));
+ verify(sqs).sendMessageBatch(request("uneven", entries[1], entries[3]));
+ verify(sqs).sendMessageBatch(request("even", entries[4]));
+ }
+
+ @Test
+ public void testWriteBatchesToDynamicWithStrictTimeout() {
Review Comment:
Thanks for having a look @psolomin :)
The intention of this test was just to check no messages are lost / delivered repeatedly if writing a lot but your concern is absolutely right, this doesn't check it's actually using the strict timeout. Will think about it, but likely I'll just add another test that also verifies the timing (with just a few message as done above) 👍
##########
sdks/java/io/amazon-web-services2/src/test/java/org/apache/beam/sdk/io/aws2/sqs/SqsIOWriteBatchesTest.java:
##########
@@ -187,24 +237,53 @@ public void testWriteBatchesToDynamic() {
}
@Test
- public void testWriteBatchesWithTimeout() {
+ public void testWriteBatchesToDynamicWithTimeout() {
when(sqs.sendMessageBatch(anyRequest())).thenReturn(completedFuture(SUCCESS));
p.apply(Create.of(5))
.apply(ParDo.of(new CreateMessages()))
.apply(
// simulate delay between messages > batch timeout
- SqsIO.writeBatches(withDelay(millis(200), SET_MESSAGE_BODY))
- .withBatchTimeout(millis(100))
- .to("queue"));
+ SqsIO.writeBatches(withDelay(millis(100), SET_MESSAGE_BODY))
+ .withBatchTimeout(millis(150))
+ .to(msg -> Integer.valueOf(msg) % 2 == 0 ? "even" : "uneven"));
p.run().waitUntilFinish();
SendMessageBatchRequestEntry[] entries = entries(range(0, 5));
- // due to added delay, batches are timed out on arrival of every 2nd msg
- verify(sqs).sendMessageBatch(request("queue", entries[0], entries[1]));
- verify(sqs).sendMessageBatch(request("queue", entries[2], entries[3]));
- verify(sqs).sendMessageBatch(request("queue", entries[4]));
+ // due to added delay, dynamic batches are timed out on arrival of every 2nd msg (per batch)
+ verify(sqs).sendMessageBatch(request("even", entries[0], entries[2]));
+ verify(sqs).sendMessageBatch(request("uneven", entries[1], entries[3]));
+ verify(sqs).sendMessageBatch(request("even", entries[4]));
+ }
+
+ @Test
+ public void testWriteBatchesToDynamicWithStrictTimeout() {
Review Comment:
Thanks for having a look @psolomin :)
The intention of this test was just to check no messages are lost / delivered repeatedly if writing a lot but your concern is absolutely right, this doesn't check it's actually using the strict timeout. Will think about it, but likely I'll just add another test that also verifies the timing (with just a few message as done above) 👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org