public void schedule(final ConnectorTaskId id, final WorkerSourceTask workerTask) {
        long commitIntervalMs = config.getLong(WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_CONFIG);
        ScheduledFuture<?> commitFuture = commitExecutorService.scheduleWithFixedDelay(() -> {
            try (LoggingContext loggingContext = LoggingContext.forOffsets(id)) {
                commit(workerTask);
            }
        }, commitIntervalMs, commitIntervalMs, TimeUnit.MILLISECONDS);
        committers.put(id, commitFuture);
    }
@SuppressWarnings("unchecked")
    @Test
    public void testSchedule() {
        ArgumentCaptor<Runnable> taskWrapper = ArgumentCaptor.forClass(Runnable.class);

        when(executor.scheduleWithFixedDelay(
                taskWrapper.capture(), eq(DEFAULT_OFFSET_COMMIT_INTERVAL_MS),
                eq(DEFAULT_OFFSET_COMMIT_INTERVAL_MS), eq(TimeUnit.MILLISECONDS))
        ).thenReturn((ScheduledFuture) commitFuture);

        committer.schedule(taskId, task);
        assertNotNull(taskWrapper.getValue());
        assertEquals(singletonMap(taskId, commitFuture), committers);
    }