Visitar URL original
Setting so the max ack deadline duration is always respected even when · googleapis/google-cloud-java@7779e37 · GitHub
Skip to content

Commit 7779e37

Browse files
committed
Setting so the max ack deadline duration is always respected even when
the next extension surpasses it.
1 parent 32af932 commit 7779e37

2 files changed

Lines changed: 8 additions & 4 deletions

File tree

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

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -88,7 +88,7 @@ class MessageDispatcher {
8888
// it is not modified while inside the queue.
8989
// The hashcode and equals methods are explicitly not implemented to discourage
9090
// the use of this class as keys in maps or similar containers.
91-
private static class ExtensionJob implements Comparable<ExtensionJob> {
91+
private class ExtensionJob implements Comparable<ExtensionJob> {
9292
Instant creation;
9393
Instant expiration;
9494
int nextExtensionSeconds;
@@ -106,7 +106,9 @@ private static class ExtensionJob implements Comparable<ExtensionJob> {
106106
}
107107

108108
void extendExpiration(Instant now) {
109-
expiration = now.plus(Duration.standardSeconds(nextExtensionSeconds));
109+
Instant possibleExtension = now.plus(Duration.standardSeconds(nextExtensionSeconds));
110+
Instant maxExtension = creation.plus(maxAckExtensionPeriod);
111+
expiration = possibleExtension.isBefore(maxExtension) ? possibleExtension : maxExtension;
110112
nextExtensionSeconds = Math.min(2 * nextExtensionSeconds, MAX_ACK_DEADLINE_EXTENSION_SECS);
111113
}
112114

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

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -278,7 +278,7 @@ public void testModifyAckDeadline() throws Exception {
278278
startSubscriber(
279279
getTestSubscriberBuilder(testReceiver)
280280
.setAckExpirationPadding(Duration.standardSeconds(1))
281-
.setMaxAckExtensionPeriod(Duration.standardSeconds(12)));
281+
.setMaxAckExtensionPeriod(Duration.standardSeconds(13)));
282282
// Send messages to be acked
283283
List<String> testAckIdsBundle = ImmutableList.of("A", "B", "C");
284284
testReceiver.setExplicitAck(true);
@@ -306,7 +306,9 @@ public ModifyAckDeadline apply(String ack) {
306306
new Function<String, ModifyAckDeadline>() {
307307
@Override
308308
public ModifyAckDeadline apply(String ack) {
309-
return new ModifyAckDeadline(ack, 4);
309+
return new ModifyAckDeadline(ack, 2); // It is expected that the deadline is renewed
310+
// only two more seconds to not pass the max
311+
// ack deadline ext.
310312
}
311313
});
312314

0 commit comments

Comments
 (0)