自定义Kafka Sink Connector单分区场景下tasks.max配置困惑

你对tasks.max的理解是正确的：Sink连接器的tasks.max上限确实等于所有输入主题的总分区数（17个单分区主题的话就是17）。但任务没有按预期分配，问题几乎肯定出在你的自定义Sink Connector实现上，核心原因是连接器没有正确将分区拆分到多个任务中。

关键检查点：自定义Connector的taskConfigs方法实现 #

Kafka Connect框架依赖SinkConnector类中的taskConfigs(int maxTasks)方法来决定如何分配任务和分区。如果这个方法没有正确生成差异化的任务配置，框架就只会启动一个任务——因为重复的任务配置会被合并。

你需要确保：

• 在taskConfigs中，将所有17个主题分区均匀分配到不同的任务配置里。比如当maxTasks=17时，每个任务的配置只包含一个单独的主题分区（如topicA-0、topicB-0等）。
• 每个任务配置必须带有唯一标识的分区集合，这样Connect框架才会识别为不同的任务，进而启动多个实例。

举个简化的实现逻辑：

@Override
public List<Map<String, String>> taskConfigs(int maxTasks) {
    List<Map<String, String>> taskConfigs = new ArrayList<>();
    // 获取所有输入主题的分区信息
    List<TopicPartition> allPartitions = getInputTopicPartitions();
    
    // 计算实际要启动的任务数（不超过总分区数和maxTasks的最小值）
    int tasksToCreate = Math.min(maxTasks, allPartitions.size());
    for (int i = 0; i < tasksToCreate; i++) {
        Map<String, String> config = new HashMap<>(this.configProperties);
        // 将当前任务负责的分区写入配置
        TopicPartition partition = allPartitions.get(i);
        config.put("assigned.partition", partition.topic() + "-" + partition.partition());
        taskConfigs.add(config);
    }
    return taskConfigs;
}

配套SinkTask的实现 #

你的SinkTask类需要在start(Map<String, String> config)方法中读取分配给它的分区，然后只消费这些分区，而非硬编码消费所有主题。比如：

@Override
public void start(Map<String, String> config) {
    String assignedPartitionStr = config.get("assigned.partition");
    String[] parts = assignedPartitionStr.split("-");
    TopicPartition assignedPartition = new TopicPartition(parts[0], Integer.parseInt(parts[1]));
    // 初始化消费者时仅订阅该分区
}

其他排查点 #

• 确认Connect Worker的worker.max.tasks配置没有限制任务数量（默认无限制，一般无需调整）。
• 查看Connect的worker日志，检查是否存在任务启动失败的报错信息，这也可能导致任务未正常分配。

内容的提问来源于stack exchange，提问作者Pawan