@Override
    public void run() {
        if (processedEvents.get() > 0) {
            LOG.debug("checkpointing offset after reaching timeout, with a batch of {}", processedEvents.get());
            eventContext.updateCheckpointAsync()
                    .subscribe(unused -> LOG.debug("Processed one event..."),
                            error -> LOG.debug("Error when updating Checkpoint: {}", error.getMessage()),
                            () -> {
                                LOG.debug("Checkpoint updated.");
                                processedEvents.set(0);
                            });
        } else {
            LOG.debug("skip checkpointing offset even if timeout is reached. No events processed");
        }
    }
@Test
    void testProcessedEventsResetWhenCheckpointUpdated() {
        var processedEvents = new AtomicInteger(1);
        var eventContext = Mockito.mock(EventContext.class);

        Mockito.when(eventContext.updateCheckpointAsync())
                .thenReturn(Mono.just("").then());

        var timerTask = new EventHubsCheckpointUpdaterTimerTask(eventContext, processedEvents);

        timerTask.run();

        assertEquals(0, processedEvents.get());
    }