@Override
    public List<Map<String, String>> taskConfigs(int maxTasks) {
        if (knownConsumerGroups == null) {
            // If knownConsumerGroup is null, it means the initial loading has not finished.
            // An exception should be thrown to trigger the retry behavior in the framework.
            log.debug("Initial consumer loading has not yet completed");
            throw new RetriableException("Timeout while loading consumer groups.");
        }

        // if the replication is disabled, known consumer group is empty, or checkpoint emission is
        // disabled by setting 'emit.checkpoints.enabled' to false, the interval of checkpoint emission
        // will be negative and no 'MirrorCheckpointTask' will be created
        if (!config.enabled() || knownConsumerGroups.isEmpty()
                || config.emitCheckpointsInterval().isNegative()) {
            return Collections.emptyList();
        }
        int numTasks = Math.min(maxTasks, knownConsumerGroups.size());
        List<List<String>> groupsPartitioned = ConnectorUtils.groupPartitions(new ArrayList<>(knownConsumerGroups), numTasks);
        return IntStream.range(0, numTasks)
                .mapToObj(i -> config.taskConfigForConsumerGroups(groupsPartitioned.get(i), i))
                .collect(Collectors.toList());
    }
@Test
    public void testNoConsumerGroup() {
        MirrorCheckpointConfig config = new MirrorCheckpointConfig(makeProps());
        MirrorCheckpointConnector connector = new MirrorCheckpointConnector(new HashSet<>(), config);
        List<Map<String, String>> output = connector.taskConfigs(1);
        // expect no task will be created
        assertEquals(0, output.size(), "ConsumerGroup shouldn't exist");
    }