Kafka Connect无法消费Topic原Partition 0数据问题求助

我遇到一个异常问题：Kafka Connect无法消费存储日志数据的某Topic中的数据。该Topic初始仅1个分区，总大小超250MB，我将其扩容至5个分区后，发现其余分区均可正常消费，但原Partition 0始终无法消费。

Partition 0的偏移量曾卡在1，我多次手动增加消费者偏移量，但情况并未改善。我可以通过kafka-ui查看所有数据，也能使用kafkacat导出全部数据，想请教为何仅单个Kafka分区出现消费问题？

消费者组状态信息 #

GROUP                        TOPIC                 PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                                                    HOST            CLIENT-ID
connect-request-log-elk-sink request-log           0          6               281294          281288          connector-consumer-request-log-elk-sink-0-ed4062a0-8712-46f5-939a-f0b783c3352c /XXX  connector-consumer-request-log-elk-sink-0
connect-request-log-elk-sink request-log           1          9882            16840           6958            connector-consumer-request-log-elk-sink-1-8f4a0ddf-7c91-4cde-93ab-05c5bf1deb2a /XXX  connector-consumer-request-log-elk-sink-1
connect-request-log-elk-sink request-log           3          9994            16918           6924            connector-consumer-request-log-elk-sink-3-83d0863e-dbc5-47ec-b240-b0e8eacbcb94 /XXX   connector-consumer-request-log-elk-sink-3
connect-request-log-elk-sink request-log           4          9744            16628           6884            connector-consumer-request-log-elk-sink-4-2c190184-5787-4b6d-91f4-16dd71fc6ea5 /XXX   connector-consumer-request-log-elk-sink-4
connect-request-log-elk-sink request-log           2          10885           16687           5802            connector-consumer-request-log-elk-sink-2-5e2c46c5-4e43-4a4c-af13-07d133e39038 /XXX   connector-consumer-request-log-elk-sink-2

Kafka Connect配置 #

{
    "connection.url": "http://xxx:9200",
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "type.name": "request-log-elk-sink",
    "behavior.on.null.values": "delete",
    "auto.create.indices.at.start": "false",
    "read.timout.ms": "10000",
    "tasks.max": "5",
    "key.ignore": "true",
    "retry.backoff.ms": "5000",
    "transforms": "InsertTimestamp",
    "transforms.InsertTimestamp.type":"org.apache.kafka.connect.transforms.InsertField$Value",
    "transforms.InsertTimestamp.timestamp.field": "eventTime",
    "errors.deadletterqueue.context.headers.enable": "true",
    "max.buffered.records": "1000",
    "errors.deadletterqueue.topic.replication.factor": "1",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "errors.log.enable": "true",
    "topics": "request-log,request-exception-log",
    "batch.size": "500",
    "max.in.flight.requests": "2",
    "schema.ignore": "true",
    "behavior.on.malformed.documents": "IGNORE",
    "flush.timeout.ms": "600000",
    "errors.deadletterqueue.topic.name": "request-log-dead-letter",
    "name": "request-log-elk-sink",
    "value.converter.schemas.enable": "false",
    "linger.ms": "1000"
}

1. 原分区0存在损坏或格式异常的消息 #

虽然kafkacat能导出数据，但Kafka Connect的转换器或Sink逻辑对消息格式更敏感。原分区0作为初始分区，可能存在：

• 不符合JSON规范的消息（比如语法错误、缺失引号）
• 消息大小超过Connect默认的max.request.size（1MB）
• 特殊编码或不可见字符导致解析失败

解决步骤：

• 用kafkacat逐步导出分区0的消息排查：

  kafkacat -b <kafka-broker地址> -t request-log -p 0 -o 6 -c 20 -J
  

• 查看Connect任务的日志，搜索JsonParseException或MalformedDocument相关报错
• 找到异常消息后，用kafka-consumer-groups.sh手动跳过该偏移量

2. Elasticsearch写入失败导致任务卡住 #

原分区0数据量远大于其他分区，批量写入ES时可能触发：

• 文档ID重复导致写入失败，behavior.on.malformed.documents=IGNORE会掩盖错误但任务会持续重试
• ES索引磁盘配额不足、分片限流，仅针对大流量的分区0触发

解决步骤：

• 查看ES集群日志，检查是否有该Sink索引的写入报错
• 修改Connect配置，临时开启errors.log.include.messages=true，打印具体错误内容
• 检查ES索引的健康状态和磁盘使用情况

3. 分区扩容后的元数据同步问题 #

分区扩容后，Connect任务可能未刷新元数据，导致消费逻辑异常：

• 任务启动时加载旧的分区信息，扩容后未重新获取
• 消费者组分区分配策略异常，导致原分区0的任务停滞

解决步骤：

• 重启该Sink任务（无需重启整个Connect集群）
• 重新确认消费者组的分区分配状态：

  kafka-consumer-groups.sh --describe --group connect-request-log-elk-sink --bootstrap-server <kafka-broker地址>
  

4. 任务内存/线程阻塞 #

原分区0数据量巨大，可能导致：

• 批量处理时内存不足，GC频繁阻塞任务线程
• InsertTimestamp转换逻辑处理大量消息时出现性能瓶颈

解决步骤：

• 降低batch.size（比如从500调到100），减少单次处理的消息数
• 查看Connect Worker的JVM监控指标（GC频率、线程状态）
• 临时增加Worker的JVM内存分配（修改KAFKA_HEAP_OPTS参数）

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