Skip to content

Commit 0e459ae

Browse files
davidtorresgarrettjonesgoogle
davidtorres
authored and
garrettjonesgoogle
committed
---
yaml --- r: 5561 b: refs/heads/master c: 95946ef h: refs/heads/master i: 5559: 9315700
1 parent f07943f commit 0e459ae

7 files changed

Lines changed: 175 additions & 16 deletions

File tree

‎[refs]‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
---
2-
refs/heads/master: e11154e946ac0cf7a517c4688ad1ab14a9565058
2+
refs/heads/master: 95946ef300f3e5a1eb3c2e6194df16ef19b938ed
33
refs/heads/travis: dae77e558b884bc1b165155482d76c8e40b0fca4
44
refs/heads/gh-pages: 049a9668af6afce3b5840dee8be005b0e6b43506
55
refs/tags/0.0.9: 22f1839238f66c39e67ed4dfdcd273b1ae2e8444

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

Lines changed: 25 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,7 @@ class MessageDispatcher {
6161
private final ApiClock clock;
6262

6363
private final Duration ackExpirationPadding;
64+
private final Duration maxAckExtensionPeriod;
6465
private final MessageReceiver receiver;
6566
private final AckProcessor ackProcessor;
6667

@@ -87,20 +88,27 @@ class MessageDispatcher {
8788
// it is not modified while inside the queue.
8889
// The hashcode and equals methods are explicitly not implemented to discourage
8990
// the use of this class as keys in maps or similar containers.
90-
private static class ExtensionJob implements Comparable<ExtensionJob> {
91+
private class ExtensionJob implements Comparable<ExtensionJob> {
92+
Instant creation;
9193
Instant expiration;
9294
int nextExtensionSeconds;
9395
ArrayList<AckHandler> ackHandlers;
9496

9597
ExtensionJob(
96-
Instant expiration, int initialAckDeadlineExtension, ArrayList<AckHandler> ackHandlers) {
98+
Instant creation,
99+
Instant expiration,
100+
int initialAckDeadlineExtension,
101+
ArrayList<AckHandler> ackHandlers) {
102+
this.creation = creation;
97103
this.expiration = expiration;
98104
nextExtensionSeconds = initialAckDeadlineExtension;
99105
this.ackHandlers = ackHandlers;
100106
}
101107

102108
void extendExpiration(Instant now) {
103-
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;
104112
nextExtensionSeconds = Math.min(2 * nextExtensionSeconds, MAX_ACK_DEADLINE_EXTENSION_SECS);
105113
}
106114

@@ -217,12 +225,14 @@ void sendAckOperations(
217225
MessageReceiver receiver,
218226
AckProcessor ackProcessor,
219227
Duration ackExpirationPadding,
228+
Duration maxAckExtensionPeriod,
220229
Distribution ackLatencyDistribution,
221230
FlowController flowController,
222231
ScheduledExecutorService executor,
223232
ApiClock clock) {
224233
this.executor = executor;
225234
this.ackExpirationPadding = ackExpirationPadding;
235+
this.maxAckExtensionPeriod = maxAckExtensionPeriod;
226236
this.receiver = receiver;
227237
this.ackProcessor = ackProcessor;
228238
this.flowController = flowController;
@@ -305,7 +315,11 @@ public void run() {
305315

306316
synchronized (outstandingAckHandlers) {
307317
outstandingAckHandlers.add(
308-
new ExtensionJob(expiration, INITIAL_ACK_DEADLINE_EXTENSION_SECONDS, ackHandlers));
318+
new ExtensionJob(
319+
new Instant(clock.millisTime()),
320+
expiration,
321+
INITIAL_ACK_DEADLINE_EXTENSION_SECONDS,
322+
ackHandlers));
309323
}
310324
setupNextAckDeadlineExtensionAlarm(expiration);
311325

@@ -380,6 +394,13 @@ public void run() {
380394
&& outstandingAckHandlers.peek().expiration.compareTo(cutOverTime) <= 0) {
381395
ExtensionJob job = outstandingAckHandlers.poll();
382396

397+
if (maxAckExtensionPeriod.getMillis() > 0
398+
&& job.creation.plus(maxAckExtensionPeriod).compareTo(now) <= 0) {
399+
// The job has expired, according to the maxAckExtensionPeriod, we are just going to
400+
// drop it.
401+
continue;
402+
}
403+
383404
// If a message has already been acked, remove it, nothing to do.
384405
for (int i = 0; i < job.ackHandlers.size(); ) {
385406
if (job.ackHandlers.get(i).acked.get()) {

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,7 @@ public PollingSubscriberConnection(
6767
Credentials credentials,
6868
MessageReceiver receiver,
6969
Duration ackExpirationPadding,
70+
Duration maxAckExtensionPeriod,
7071
Distribution ackLatencyDistribution,
7172
Channel channel,
7273
FlowController flowController,
@@ -82,6 +83,7 @@ public PollingSubscriberConnection(
8283
receiver,
8384
this,
8485
ackExpirationPadding,
86+
maxAckExtensionPeriod,
8587
ackLatencyDistribution,
8688
flowController,
8789
executor,

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -70,6 +70,7 @@ public StreamingSubscriberConnection(
7070
Credentials credentials,
7171
MessageReceiver receiver,
7272
Duration ackExpirationPadding,
73+
Duration maxAckExtensionPeriod,
7374
int streamAckDeadlineSeconds,
7475
Distribution ackLatencyDistribution,
7576
Channel channel,
@@ -85,6 +86,7 @@ public StreamingSubscriberConnection(
8586
receiver,
8687
this,
8788
ackExpirationPadding,
89+
maxAckExtensionPeriod,
8890
ackLatencyDistribution,
8991
flowController,
9092
executor,

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

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -92,6 +92,7 @@ public class Subscriber extends AbstractApiService {
9292
private final String cachedSubscriptionNameString;
9393
private final FlowControlSettings flowControlSettings;
9494
private final Duration ackExpirationPadding;
95+
private final Duration maxAckExtensionPeriod;
9596
private final ScheduledExecutorService executor;
9697
private final Distribution ackLatencyDistribution =
9798
new Distribution(MAX_ACK_DEADLINE_SECONDS + 1);
@@ -113,6 +114,7 @@ private Subscriber(Builder builder) throws IOException {
113114
subscriptionName = builder.subscriptionName;
114115
cachedSubscriptionNameString = subscriptionName.toString();
115116
ackExpirationPadding = builder.ackExpirationPadding;
117+
maxAckExtensionPeriod = builder.maxAckExtensionPeriod;
116118
streamAckDeadlineSeconds =
117119
Math.max(
118120
INITIAL_ACK_DEADLINE_SECONDS,
@@ -245,6 +247,7 @@ private void startStreamingConnections() {
245247
credentials,
246248
receiver,
247249
ackExpirationPadding,
250+
maxAckExtensionPeriod,
248251
streamAckDeadlineSeconds,
249252
ackLatencyDistribution,
250253
channelBuilder.build(),
@@ -321,6 +324,7 @@ private void startPollingConnections() {
321324
credentials,
322325
receiver,
323326
ackExpirationPadding,
327+
maxAckExtensionPeriod,
324328
ackLatencyDistribution,
325329
channelBuilder.build(),
326330
flowController,
@@ -409,6 +413,7 @@ public void run() {
409413
public static final class Builder {
410414
private static final Duration MIN_ACK_EXPIRATION_PADDING = Duration.millis(100);
411415
private static final Duration DEFAULT_ACK_EXPIRATION_PADDING = Duration.millis(500);
416+
private static final Duration DEFAULT_MAX_ACK_EXTENSION_PERIOD = Duration.standardMinutes(60);
412417

413418
static final ExecutorProvider DEFAULT_EXECUTOR_PROVIDER =
414419
InstantiatingExecutorProvider.newBuilder()
@@ -423,6 +428,7 @@ public static final class Builder {
423428
MessageReceiver receiver;
424429

425430
Duration ackExpirationPadding = DEFAULT_ACK_EXPIRATION_PADDING;
431+
Duration maxAckExtensionPeriod = DEFAULT_MAX_ACK_EXTENSION_PERIOD;
426432

427433
FlowControlSettings flowControlSettings = FlowControlSettings.getDefaultInstance();
428434

@@ -483,6 +489,21 @@ public Builder setAckExpirationPadding(Duration ackExpirationPadding) {
483489
return this;
484490
}
485491

492+
/**
493+
* Set the maximum period a message ack deadline will be extended.
494+
*
495+
* <p>It is recommended to set this value to a reasonable upper bound of the subscriber time to
496+
* process any message. This maximum period avoids messages to be <i>locked</i> by a subscriber
497+
* in cases when the {@link AckReply} is lost.
498+
*
499+
* <p>A zero duration effectively disables auto deadline extensions.
500+
*/
501+
public Builder setMaxAckExtensionPeriod(Duration maxAckExtensionPeriod) {
502+
Preconditions.checkArgument(maxAckExtensionPeriod.getMillis() >= 0);
503+
this.maxAckExtensionPeriod = maxAckExtensionPeriod;
504+
return this;
505+
}
506+
486507
/** Gives the ability to set a custom executor. */
487508
public Builder setExecutorProvider(ExecutorProvider executorProvider) {
488509
this.executorProvider = Preconditions.checkNotNull(executorProvider);
@@ -500,3 +521,4 @@ public Subscriber build() throws IOException {
500521
}
501522
}
502523
}
524+

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

Lines changed: 45 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,8 @@
2020
import com.google.common.primitives.Ints;
2121
import com.google.common.util.concurrent.SettableFuture;
2222
import java.util.ArrayList;
23+
import java.util.Deque;
24+
import java.util.LinkedList;
2325
import java.util.List;
2426
import java.util.PriorityQueue;
2527
import java.util.concurrent.AbstractExecutorService;
@@ -44,6 +46,7 @@ public class FakeScheduledExecutorService extends AbstractExecutorService
4446
private final AtomicBoolean shutdown = new AtomicBoolean(false);
4547
private final PriorityQueue<PendingCallable<?>> pendingCallables = new PriorityQueue<>();
4648
private final FakeClock clock = new FakeClock();
49+
private final Deque<Duration> expectedWorkQueue = new LinkedList<>();
4750

4851
public ApiClock getClock() {
4952
return clock;
@@ -79,6 +82,35 @@ public ScheduledFuture<?> scheduleWithFixedDelay(
7982
Duration.millis(unit.toMillis(initialDelay)), command, PendingCallableType.FIXED_DELAY));
8083
}
8184

85+
/**
86+
* This allows for adding expectations on future work to be scheduled (
87+
* {@link FakeScheduledExecutorService#schedule}
88+
* or {@link FakeScheduledExecutorService#scheduleAtFixedRate}
89+
* or {@link FakeScheduledExecutorService#scheduleWithFixedDelay}) based on its delay.
90+
*/
91+
public void setupScheduleExpectation(Duration delay) {
92+
synchronized (expectedWorkQueue) {
93+
expectedWorkQueue.add(delay);
94+
}
95+
}
96+
97+
/**
98+
* Blocks the current thread until all the work
99+
* {@link FakeScheduledExecutorService#setupScheduleExpectation(Duration) expected} has been
100+
* scheduled in the executor.
101+
*/
102+
public void waitForExpectedWork() {
103+
synchronized (expectedWorkQueue) {
104+
while (!expectedWorkQueue.isEmpty()) {
105+
try {
106+
expectedWorkQueue.wait();
107+
} catch (InterruptedException e) {
108+
// Wait uninterruptibly
109+
}
110+
}
111+
}
112+
}
113+
82114
/**
83115
* This will advance the reference time of the executor and execute (in the same thread) any
84116
* outstanding callable which execution time has passed.
@@ -94,13 +126,14 @@ private void work() {
94126
for (;;) {
95127
PendingCallable<?> callable = null;
96128
synchronized (pendingCallables) {
97-
if (pendingCallables.isEmpty() || pendingCallables.peek().getScheduledTime().isAfter(cmpTime)) {
129+
if (pendingCallables.isEmpty()
130+
|| pendingCallables.peek().getScheduledTime().isAfter(cmpTime)) {
98131
break;
99132
}
100133
callable = pendingCallables.poll();
101134
}
102135
if (callable != null) {
103-
try{
136+
try {
104137
callable.call();
105138
} catch (Exception e) {
106139
// We ignore any callable exception, which should be set to the future but not relevant to
@@ -182,6 +215,16 @@ <V> ScheduledFuture<V> schedulePendingCallable(PendingCallable<V> callable) {
182215
pendingCallables.add(callable);
183216
}
184217
work();
218+
synchronized (expectedWorkQueue) {
219+
// We compare by the callable delay in order decide when to remove expectations from the
220+
// expected work queue, i.e. only the expected work that matches the delay of the scheduled
221+
// callable is removed from the queue.
222+
if (!expectedWorkQueue.isEmpty() && expectedWorkQueue.peek().equals(callable.delay)) {
223+
expectedWorkQueue.poll();
224+
}
225+
expectedWorkQueue.notifyAll();
226+
}
227+
185228
return callable.getScheduledFuture();
186229
}
187230

0 commit comments

Comments
 (0)