@Override
    public List<SourceRecord> poll() throws InterruptedException {
        try {
            long deadline = System.currentTimeMillis() + interval.toMillis();
            while (!stopping && System.currentTimeMillis() < deadline) {
                Thread.sleep(pollTimeout.toMillis());
            }
            if (stopping || !checkpointStore.isInitialized()) {
                // we are stopping, or not fully initialized, return early.
                return null;
            }
            List<SourceRecord> records = new ArrayList<>();
            for (String group : consumerGroups) {
                records.addAll(sourceRecordsForGroup(group));
            }
            if (records.isEmpty()) {
                // WorkerSourceTask expects non-zero batches or null
                return null;
            } else {
                return records;
            }
        } catch (Throwable e) {
            log.warn("Failure polling consumer state for checkpoints.", e);
            return null;
        }
    }
@Test
    public void testCheckpointStoreInitialized() throws InterruptedException {
        CheckpointStore checkpointStore = mock(CheckpointStore.class);

        MirrorCheckpointTask task = new MirrorCheckpointTask("source1", "target2",
                new DefaultReplicationPolicy(),
                new OffsetSyncStoreTest.FakeOffsetSyncStore(),
                Collections.singleton("group"),
                Collections.emptyMap(),
                checkpointStore) {

            @Override
            List<SourceRecord> sourceRecordsForGroup(String group) {
                SourceRecord sr = new SourceRecord(Collections.emptyMap(), Collections.emptyMap(), "", 0, null, null);
                return Collections.singletonList(sr);
            }
        };

        assertNull(task.poll());

        when(checkpointStore.isInitialized()).thenReturn(true);
        List<SourceRecord> polled = task.poll();
        assertEquals(1, polled.size());
    }