Kafka Connect自定义Sink连接器手动管理offset提交不生效如何处理

必改配置项 #

• 连接器配置中除了设置"consumer.enable.auto.commit": "false"外，还需新增配置"offset.commit.policy": "MANUAL_PER_PARTITION"，明确告知Worker使用手动提交策略，避免Worker使用默认的定期自动提交逻辑覆盖你的配置。
• 若开启了恰好一次语义，需确认"exactly.once.support"配置未设置为required，该模式下会通过Kafka事务自动提交offset，不受preCommit返回值控制。
• 检查消费者组的auto.offset.reset配置，需设置为earliest避免首次消费或offset过期时直接从最新位置拉取，导致看起来消息没有被重新消费的错觉。

代码调整逻辑 #

• 不要固定返回空Map给preCommit方法，自行维护全局的Map<TopicPartition, OffsetAndMetadata>缓存，仅将处理成功的消息对应的offset更新到缓存中，preCommit方法直接返回该缓存即可。如果固定返回空Map，Worker会认为没有offset需要提交，下次启动时仍会读取上一次全局提交的offset位置，不会回退到当前未处理的消息位置。
• 消息处理失败时，不要静默吞掉异常，必须抛出org.apache.kafka.connect.errors.RetriableException，触发Worker的重试机制，此时Worker不会推进拉取位点，下次poll会重新拉取该批次的所有消息。
• 不需要主动调用空参数的flush方法，该方法会触发Worker的默认提交逻辑，仅当你需要主动提交offset时，调用SinkTaskContext.offsetCommit(offsets)传入你要提交的offset集合即可。
• 运行过程中未重启进程时，消费者会在内存中维护拉取位点，即使未提交到外部offset存储，也会继续拉取新消息。如果需要运行时就触发未处理消息的重消费，需调用SinkTaskContext.offset(TopicPartition)获取当前位点后，调用consumer.seek()回退到目标位置，或者抛出RetriableException触发批次重试。

内容的提问来源于stack exchange，提问作者omer bar lev