@Override
    public CompletableFuture<Void> quiesce() {
        if (!quiesced) {
            quiesced = true;

            if (numRunningTimers.get() == 0) {
                quiesceCompletedFuture.complete(null);
            }
        }

        return quiesceCompletedFuture;
    }
@Test
    void testQuiesce() throws Exception {
        ProcessingTimeServiceImpl processingTimeService =
                new ProcessingTimeServiceImpl(timerService, v -> v);

        final CompletableFuture<?> timerRunFuture = new CompletableFuture();
        final OneShotLatch timerWaitLatch = new OneShotLatch();

        ScheduledFuture<?> timer =
                processingTimeService.registerTimer(
                        0,
                        timestamp -> {
                            timerRunFuture.complete(null);
                            timerWaitLatch.await();
                        });

        // wait for the timer to run, then quiesce the time service
        timerRunFuture.get(TESTING_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
        CompletableFuture<?> quiesceCompletedFuture = processingTimeService.quiesce();

        // after the timer server is quiesced, tests #registerTimer() and #scheduleAtFixedRate()
        assertThat((Future<?>) processingTimeService.registerTimer(0, timestamp -> {}))
                .isInstanceOf(NeverCompleteFuture.class);
        assertThat(
                        (Future<?>)
                                processingTimeService.scheduleAtFixedRate(
                                        timestamp -> {}, 0, Long.MAX_VALUE))
                .isInstanceOf(NeverCompleteFuture.class);

        // when the timer is finished, the quiesce-completed future should be completed
        assertThat(quiesceCompletedFuture).isNotDone();
        timerWaitLatch.trigger();
        timer.get(TESTING_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
        assertThat(quiesceCompletedFuture).isDone();
    }