Skip to content

Commit bf1dad9

Browse files
anuchandyanuchandy
authored
Stabilizing the WindowedSubscriber test that uses VTS time advancing, and test comments (Azure#39418)
1 parent f96205e commit bf1dad9

File tree

2 files changed

+124
-87
lines changed

2 files changed

+124
-87
lines changed

‎sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/WindowedSubscriberFluxWindowIsolatedTest.java

Lines changed: 120 additions & 86 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33

44
package com.azure.messaging.servicebus;
55

6+
import com.azure.core.util.logging.ClientLogger;
67
import org.junit.jupiter.api.Assertions;
78
import org.junit.jupiter.api.Test;
89
import org.junit.jupiter.api.parallel.Execution;
@@ -19,6 +20,7 @@
1920
import java.util.Deque;
2021
import java.util.List;
2122
import java.util.concurrent.ConcurrentLinkedDeque;
23+
import java.util.concurrent.atomic.AtomicBoolean;
2224
import java.util.concurrent.atomic.AtomicReference;
2325
import java.util.function.Consumer;
2426
import java.util.function.Supplier;
@@ -33,6 +35,8 @@
3335
@Execution(ExecutionMode.SAME_THREAD)
3436
@Isolated
3537
public final class WindowedSubscriberFluxWindowIsolatedTest {
38+
private final ClientLogger logger = new ClientLogger(WindowedSubscriberFluxWindowIsolatedTest.class);
39+
3640
@Test
3741
@Execution(ExecutionMode.SAME_THREAD)
3842
public void shouldCloseEmptyWindowOnTimeout() {
@@ -44,13 +48,14 @@ public void shouldCloseEmptyWindowOnTimeout() {
4448
upstream.subscribe(subscriber);
4549

4650
final AtomicReference<EnqueueResult<Integer>> rRef = new AtomicReference<>();
47-
final Supplier<Publisher<Integer>> scenario = () -> {
48-
final EnqueueResult<Integer> r = subscriber.enqueueRequestImpl(windowSize, windowTimeout);
49-
rRef.set(r);
50-
return r.getWindowFlux();
51-
};
52-
5351
try (VirtualTimeStepVerifier verifier = new VirtualTimeStepVerifier()) {
52+
final Supplier<Publisher<Integer>> scenario = () -> {
53+
verifier.logIfClosedUnexpectedly(logger);
54+
final EnqueueResult<Integer> r = subscriber.enqueueRequestImpl(windowSize, windowTimeout);
55+
rRef.set(r);
56+
return r.getWindowFlux();
57+
};
58+
5459
verifier.create(scenario)
5560
// Forward time to timeout empty windowTimeout.
5661
.thenAwait(windowTimeout.plusSeconds(10))
@@ -74,13 +79,14 @@ public void shouldCloseStreamingWindowOnTimeout() {
7479
upstream.subscribe(subscriber);
7580

7681
final AtomicReference<EnqueueResult<Integer>> rRef = new AtomicReference<>();
77-
final Supplier<Publisher<Integer>> scenario = () -> {
78-
final EnqueueResult<Integer> r = subscriber.enqueueRequestImpl(windowSize, windowTimeout);
79-
rRef.set(r);
80-
return r.getWindowFlux();
81-
};
82-
8382
try (VirtualTimeStepVerifier verifier = new VirtualTimeStepVerifier()) {
83+
final Supplier<Publisher<Integer>> scenario = () -> {
84+
verifier.logIfClosedUnexpectedly(logger);
85+
final EnqueueResult<Integer> r = subscriber.enqueueRequestImpl(windowSize, windowTimeout);
86+
rRef.set(r);
87+
return r.getWindowFlux();
88+
};
89+
8490
verifier.create(scenario)
8591
.then(() -> upstream.next(1))
8692
.then(() -> upstream.next(2))
@@ -108,17 +114,18 @@ public void shouldContinueToNextWindowWhenEmptyWindowTimeout() {
108114

109115
final AtomicReference<EnqueueResult<Integer>> r0Ref = new AtomicReference<>();
110116
final AtomicReference<EnqueueResult<Integer>> r1Ref = new AtomicReference<>();
111-
final Supplier<Publisher<Integer>> scenario = () -> {
112-
final EnqueueResult<Integer> r0 = subscriber.enqueueRequestImpl(windowSize, windowTimeout);
113-
final EnqueueResult<Integer> r1 = subscriber.enqueueRequestImpl(windowSize, windowTimeout);
114-
r0Ref.set(r0);
115-
r1Ref.set(r1);
116-
final Flux<Integer> window0Flux = r0.getWindowFlux();
117-
final Flux<Integer> window1Flux = r1.getWindowFlux();
118-
return window0Flux.concatWith(window1Flux);
119-
};
120-
121117
try (VirtualTimeStepVerifier verifier = new VirtualTimeStepVerifier()) {
118+
final Supplier<Publisher<Integer>> scenario = () -> {
119+
verifier.logIfClosedUnexpectedly(logger);
120+
final EnqueueResult<Integer> r0 = subscriber.enqueueRequestImpl(windowSize, windowTimeout);
121+
final EnqueueResult<Integer> r1 = subscriber.enqueueRequestImpl(windowSize, windowTimeout);
122+
r0Ref.set(r0);
123+
r1Ref.set(r1);
124+
final Flux<Integer> window0Flux = r0.getWindowFlux();
125+
final Flux<Integer> window1Flux = r1.getWindowFlux();
126+
return window0Flux.concatWith(window1Flux);
127+
};
128+
122129
verifier.create(scenario)
123130
// Forward time to timeout empty window0Flux,
124131
.thenAwait(windowTimeout.plusSeconds(10))
@@ -154,17 +161,18 @@ public void shouldContinueToNextWindowWhenStreamingWindowTimeout() {
154161

155162
final AtomicReference<EnqueueResult<Integer>> r0Ref = new AtomicReference<>();
156163
final AtomicReference<EnqueueResult<Integer>> r1Ref = new AtomicReference<>();
157-
final Supplier<Publisher<Integer>> scenario = () -> {
158-
final EnqueueResult<Integer> r0 = subscriber.enqueueRequestImpl(windowSize, windowTimeout);
159-
final EnqueueResult<Integer> r1 = subscriber.enqueueRequestImpl(windowSize, windowTimeout);
160-
r0Ref.set(r0);
161-
r1Ref.set(r1);
162-
final Flux<Integer> window0Flux = r0.getWindowFlux();
163-
final Flux<Integer> window1Flux = r1.getWindowFlux();
164-
return window0Flux.concatWith(window1Flux);
165-
};
166-
167164
try (VirtualTimeStepVerifier verifier = new VirtualTimeStepVerifier()) {
165+
final Supplier<Publisher<Integer>> scenario = () -> {
166+
verifier.logIfClosedUnexpectedly(logger);
167+
final EnqueueResult<Integer> r0 = subscriber.enqueueRequestImpl(windowSize, windowTimeout);
168+
final EnqueueResult<Integer> r1 = subscriber.enqueueRequestImpl(windowSize, windowTimeout);
169+
r0Ref.set(r0);
170+
r1Ref.set(r1);
171+
final Flux<Integer> window0Flux = r0.getWindowFlux();
172+
final Flux<Integer> window1Flux = r1.getWindowFlux();
173+
return window0Flux.concatWith(window1Flux);
174+
};
175+
168176
verifier.create(scenario)
169177
.then(() -> upstream.next(1))
170178
.then(() -> upstream.next(2))
@@ -186,9 +194,19 @@ public void shouldContinueToNextWindowWhenStreamingWindowTimeout() {
186194
Assertions.assertFalse(work0.isCanceled());
187195

188196
final WindowWork<Integer> work1 = r1Ref.get().getInnerWork();
189-
Assertions.assertNotEquals(windowSize, work1.getPending());
190197
Assertions.assertTrue(work1.hasTimedOut());
191198
Assertions.assertFalse(work1.isCanceled());
199+
final boolean hasWindow1ReceivedNothing = work1.getPending() == windowSize;
200+
if (hasWindow1ReceivedNothing) {
201+
// The combination of VirtualTimeScheduler and WindowedSubscriber.drain() sometimes delays arrival of timeout
202+
// signaling for window0, resulting window0 to timeout only after the emission of 3 (and 4). This result in
203+
// window0 to receive 1, 2, 3 and 4, and window1 to receive nothing. Here asserting that, when/if this happens
204+
// application still gets emitted events via window0.
205+
//
206+
final boolean hasWindow0ReceivedAll = work0.getPending() == windowSize - 4; // (demanded - received)
207+
Assertions.assertTrue(hasWindow0ReceivedAll,
208+
String.format("window0 pending: %d, window1 pending: %d", work0.getPending(), work1.getPending()));
209+
}
192210
}
193211

194212
@Test
@@ -204,19 +222,20 @@ public void shouldContinueToNextWindowWhenStreamingWindowCancels() {
204222

205223
final AtomicReference<EnqueueResult<Integer>> r0Ref = new AtomicReference<>();
206224
final AtomicReference<EnqueueResult<Integer>> r1Ref = new AtomicReference<>();
207-
final Supplier<Publisher<Integer>> scenario = () -> {
208-
final EnqueueResult<Integer> r0 = subscriber.enqueueRequestImpl(windowSize, windowTimeout);
209-
final EnqueueResult<Integer> r1 = subscriber.enqueueRequestImpl(windowSize, windowTimeout);
210-
r0Ref.set(r0);
211-
r1Ref.set(r1);
212-
final Flux<Integer> window0Flux = r0.getWindowFlux();
213-
final Flux<Integer> window1Flux = r1.getWindowFlux();
214-
return window0Flux
215-
.take(cancelAfter)
216-
.concatWith(window1Flux);
217-
};
218-
219225
try (VirtualTimeStepVerifier verifier = new VirtualTimeStepVerifier()) {
226+
final Supplier<Publisher<Integer>> scenario = () -> {
227+
verifier.logIfClosedUnexpectedly(logger);
228+
final EnqueueResult<Integer> r0 = subscriber.enqueueRequestImpl(windowSize, windowTimeout);
229+
final EnqueueResult<Integer> r1 = subscriber.enqueueRequestImpl(windowSize, windowTimeout);
230+
r0Ref.set(r0);
231+
r1Ref.set(r1);
232+
final Flux<Integer> window0Flux = r0.getWindowFlux();
233+
final Flux<Integer> window1Flux = r1.getWindowFlux();
234+
return window0Flux
235+
.take(cancelAfter)
236+
.concatWith(window1Flux);
237+
};
238+
220239
verifier.create(scenario)
221240
.then(() -> upstream.next(1))
222241
.then(() -> upstream.next(2))
@@ -248,13 +267,14 @@ public void shouldRequestWindowDemand() {
248267
upstream.subscribe(subscriber);
249268

250269
final AtomicReference<EnqueueResult<Integer>> rRef = new AtomicReference<>();
251-
final Supplier<Publisher<Integer>> scenario = () -> {
252-
final EnqueueResult<Integer> r = subscriber.enqueueRequestImpl(windowSize, windowTimeout);
253-
rRef.set(r);
254-
return r.getWindowFlux();
255-
};
256-
257270
try (VirtualTimeStepVerifier verifier = new VirtualTimeStepVerifier()) {
271+
final Supplier<Publisher<Integer>> scenario = () -> {
272+
verifier.logIfClosedUnexpectedly(logger);
273+
final EnqueueResult<Integer> r = subscriber.enqueueRequestImpl(windowSize, windowTimeout);
274+
rRef.set(r);
275+
return r.getWindowFlux();
276+
};
277+
258278
verifier.create(scenario)
259279
.thenAwait(windowTimeout.plusSeconds(10))
260280
.verifyComplete();
@@ -279,18 +299,18 @@ public void shouldAccountPendingRequestWhenServingNextWindowDemand() {
279299

280300
final AtomicReference<EnqueueResult<Integer>> r0Ref = new AtomicReference<>();
281301
final AtomicReference<EnqueueResult<Integer>> r1Ref = new AtomicReference<>();
282-
final Supplier<Publisher<Integer>> scenario = () -> {
283-
final EnqueueResult<Integer> r0 = subscriber.enqueueRequestImpl(window0Size, windowTimeout);
284-
final EnqueueResult<Integer> r1 = subscriber.enqueueRequestImpl(window1Size, windowTimeout);
285-
r0Ref.set(r0);
286-
r1Ref.set(r1);
287-
final Flux<Integer> window0Flux = r0.getWindowFlux();
288-
final Flux<Integer> window1Flux = r1.getWindowFlux();
289-
return window0Flux.concatWith(window1Flux);
290-
};
291-
292-
293302
try (VirtualTimeStepVerifier verifier = new VirtualTimeStepVerifier()) {
303+
final Supplier<Publisher<Integer>> scenario = () -> {
304+
verifier.logIfClosedUnexpectedly(logger);
305+
final EnqueueResult<Integer> r0 = subscriber.enqueueRequestImpl(window0Size, windowTimeout);
306+
final EnqueueResult<Integer> r1 = subscriber.enqueueRequestImpl(window1Size, windowTimeout);
307+
r0Ref.set(r0);
308+
r1Ref.set(r1);
309+
final Flux<Integer> window0Flux = r0.getWindowFlux();
310+
final Flux<Integer> window1Flux = r1.getWindowFlux();
311+
return window0Flux.concatWith(window1Flux);
312+
};
313+
294314
verifier.create(scenario)
295315
// timeout window0Flux without receiving (so pending request become 'windowSize0'), and pick next work.
296316
.thenAwait(windowTimeout.plusSeconds(10))
@@ -322,17 +342,19 @@ public void shouldPickEnqueuedWindowRequestsOnSubscriptionReady() {
322342

323343
final AtomicReference<EnqueueResult<Integer>> r0Ref = new AtomicReference<>();
324344
final AtomicReference<EnqueueResult<Integer>> r1Ref = new AtomicReference<>();
325-
final Supplier<Publisher<Integer>> scenario = () -> {
326-
final EnqueueResult<Integer> r0 = subscriber.enqueueRequestImpl(window0Size, windowTimeout);
327-
final EnqueueResult<Integer> r1 = subscriber.enqueueRequestImpl(window1Size, windowTimeout);
328-
r0Ref.set(r0);
329-
r1Ref.set(r1);
330-
final Flux<Integer> window0Flux = r0.getWindowFlux();
331-
final Flux<Integer> window1Flux = r1.getWindowFlux();
332-
return window0Flux.concatWith(window1Flux);
333-
};
334345

335346
try (VirtualTimeStepVerifier verifier = new VirtualTimeStepVerifier()) {
347+
final Supplier<Publisher<Integer>> scenario = () -> {
348+
verifier.logIfClosedUnexpectedly(logger);
349+
final EnqueueResult<Integer> r0 = subscriber.enqueueRequestImpl(window0Size, windowTimeout);
350+
final EnqueueResult<Integer> r1 = subscriber.enqueueRequestImpl(window1Size, windowTimeout);
351+
r0Ref.set(r0);
352+
r1Ref.set(r1);
353+
final Flux<Integer> window0Flux = r0.getWindowFlux();
354+
final Flux<Integer> window1Flux = r1.getWindowFlux();
355+
return window0Flux.concatWith(window1Flux);
356+
};
357+
336358
verifier.create(scenario)
337359
// subscribe after enqueuing requests in 'scenario' (mimicking late arrival of subscription).
338360
.then(() -> upstream.subscribe(subscriber))
@@ -366,14 +388,13 @@ public void shouldInvokeReleaserWhenNoWindowToService() {
366388
final WindowedSubscriber<Integer> subscriber = createSubscriber(options.setReleaser(releaser));
367389
upstream.subscribe(subscriber);
368390

369-
final AtomicReference<EnqueueResult<Integer>> rRef = new AtomicReference<>();
370-
final Supplier<Publisher<Integer>> scenario = () -> {
371-
final EnqueueResult<Integer> r = subscriber.enqueueRequestImpl(windowSize, windowTimeout);
372-
rRef.set(r);
373-
return r.getWindowFlux();
374-
};
375-
376391
try (VirtualTimeStepVerifier verifier = new VirtualTimeStepVerifier()) {
392+
final Supplier<Publisher<Integer>> scenario = () -> {
393+
verifier.logIfClosedUnexpectedly(logger);
394+
final EnqueueResult<Integer> r = subscriber.enqueueRequestImpl(windowSize, windowTimeout);
395+
return r.getWindowFlux();
396+
};
397+
377398
verifier.create(scenario)
378399
// forward time to timeout windowFlux without receiving.
379400
.thenAwait(windowTimeout.plusSeconds(10))
@@ -400,14 +421,13 @@ public void shouldStopInvokingReleaserOnUpstreamTermination() {
400421
final WindowedSubscriber<Integer> subscriber = createSubscriber(options.setReleaser(releaser));
401422
upstream.subscribe(subscriber);
402423

403-
final AtomicReference<EnqueueResult<Integer>> rRef = new AtomicReference<>();
404-
final Supplier<Publisher<Integer>> scenario = () -> {
405-
final EnqueueResult<Integer> r = subscriber.enqueueRequestImpl(windowSize, windowTimeout);
406-
rRef.set(r);
407-
return r.getWindowFlux();
408-
};
409-
410424
try (VirtualTimeStepVerifier verifier = new VirtualTimeStepVerifier()) {
425+
final Supplier<Publisher<Integer>> scenario = () -> {
426+
verifier.logIfClosedUnexpectedly(logger);
427+
final EnqueueResult<Integer> r = subscriber.enqueueRequestImpl(windowSize, windowTimeout);
428+
return r.getWindowFlux();
429+
};
430+
411431
verifier.create(scenario)
412432
// forward time to timeout windowFlux without receiving.
413433
.thenAwait(windowTimeout.plusSeconds(10))
@@ -424,10 +444,11 @@ public void shouldStopInvokingReleaserOnUpstreamTermination() {
424444
Assertions.assertEquals(Arrays.asList(1, 2), released);
425445
}
426446

427-
private static final class VirtualTimeStepVerifier implements AutoCloseable {
447+
private static final class VirtualTimeStepVerifier extends AtomicBoolean implements AutoCloseable {
428448
private final VirtualTimeScheduler scheduler;
429449

430450
VirtualTimeStepVerifier() {
451+
super(false);
431452
scheduler = VirtualTimeScheduler.create();
432453
}
433454

@@ -437,8 +458,21 @@ <T> StepVerifier.Step<T> create(Supplier<Publisher<T>> scenarioSupplier) {
437458

438459
@Override
439460
public void close() {
461+
super.set(true);
440462
scheduler.dispose();
441463
}
464+
465+
void logIfClosedUnexpectedly(ClientLogger logger) {
466+
final boolean wasAutoClosed = get();
467+
final boolean isSchedulerDisposed = scheduler.isDisposed();
468+
if (wasAutoClosed || isSchedulerDisposed) {
469+
if (!wasAutoClosed) {
470+
logger.atError().log("VirtualTimeScheduler unavailable (unexpected close from outside of the test).");
471+
} else {
472+
logger.atError().log("VirtualTimeScheduler unavailable (unexpected close by the test).");
473+
}
474+
}
475+
}
442476
}
443477

444478
private static class Releaser<T> implements Consumer<T> {

‎sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/WindowedSubscriberIterableWindowTest.java

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -655,7 +655,8 @@ public void shouldContinueToNextWindowOnEmptyWindowTimeout() {
655655
final IterableStream<Integer> window0Iterable = r0.getWindowIterable();
656656
final IterableStream<Integer> window1Iterable = r1.getWindowIterable();
657657

658-
658+
// When time out triggers, it will close the window0Iterable stream, which will end the blocking collect() call
659+
// by returning "empty" list since "no events were received" within the timeout.
659660
final List<Integer> list0 = window0Iterable.stream().collect(Collectors.toList());
660661
Assertions.assertEquals(0, list0.size());
661662

@@ -694,6 +695,8 @@ public void shouldContinueToNextWindowWhenStreamingWindowTimeout() {
694695
upstream.next(1);
695696
upstream.next(2);
696697

698+
// When time out triggers, it will close the window0Iterable stream, which will end the blocking collect() call
699+
// and return the list, list0, with events received so far (which is less than demanded).
697700
final List<Integer> list0 = window0Iterable.stream().collect(Collectors.toList());
698701
Assertions.assertEquals(Arrays.asList(1, 2), list0);
699702

0 commit comments

Comments
 (0)