@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 testMirrorCheckpointConnectorEnabled() {
        // enable the checkpoint emission
        MirrorCheckpointConfig config = new MirrorCheckpointConfig(
                makeProps("emit.checkpoints.enabled", "true"));

        Set<String> knownConsumerGroups = new HashSet<>();
        knownConsumerGroups.add(CONSUMER_GROUP);
        // MirrorCheckpointConnector as minimum to run taskConfig()
        MirrorCheckpointConnector connector = new MirrorCheckpointConnector(knownConsumerGroups,
                config);
        List<Map<String, String>> output = connector.taskConfigs(1);
        // expect 1 task will be created
        assertEquals(1, output.size(),
                "MirrorCheckpointConnectorEnabled for " + CONSUMER_GROUP + " has incorrect size");
        assertEquals(CONSUMER_GROUP, output.get(0).get(MirrorCheckpointConfig.TASK_CONSUMER_GROUPS),
                "MirrorCheckpointConnectorEnabled for " + CONSUMER_GROUP + " failed");
    }