@Override
public List<Map<String, String>> taskConfigs(int maxTasks) {
    return ConnectorUtils.groupPartitions(new ArrayList<>(streamShards.keySet()), maxTasks).stream().map(taskShards -> {
        final Map<String, String> taskConfig = new HashMap<>();
        taskConfig.put(TaskConfig.Keys.REGION, config.region.getName());
        taskConfig.put(TaskConfig.Keys.TOPIC_FORMAT, config.topicFormat);
        taskConfig.put(TaskConfig.Keys.SHARDS, taskShards.stream().map(Shard::getShardId).collect(Collectors.joining(",")));
        taskShards.forEach(shard -> {
            final TableDescription tableDesc = streamShards.get(shard);
            taskConfig.put(shard.getShardId() + "." + TaskConfig.Keys.TABLE, tableDesc.getTableName());
            taskConfig.put(shard.getShardId() + "." + TaskConfig.Keys.STREAM_ARN, tableDesc.getLatestStreamArn());
        });
        return taskConfig;
    }).collect(Collectors.toList());
}