kafka routine load kafka consume failed Broker: Offset out of range

[xonze]

使用kafka routine load数据过程中的Offset out of range问题

可能由于大批量导入历史数据的原因，中间出现过个别机器有commit失败的情况，时间久了就会出现各个BE所分配的partition消费offset不均衡差异过大，最终累计时间久了kafka中旧的数据会过期自动清理，这个时候就会出现个别BE对应partition消费的当前的offset已经没有数据了（就是BE消费太慢，kafka已经过期的数据的offset它才在消费），这个时候就会报以下错误，并且会暂停routine load

kafka consume failed: b2489f7e1dc83663-8a71e3ae834b3f9a, msg: Broker: Offset out of range

这里有两个问题：

1. routine load过程中没有动态的考虑offset均衡问题，每个BE始终分配的那几个固定的partition，
2. 消费kafka中已经过期的offset对应的数据的时候应该跳过并继续往后消费，而不是报Offset out of range并暂停routine load

[morningman]

1. routine load过程中没有动态的考虑offset均衡问题，每个BE始终分配的那几个固定的partition

We will find a way to balance the partition

2. 消费kafka中已经过期的offset对应的数据的时候应该跳过并继续往后消费，而不是报Offset out of range并暂停routine load

I think it is not safe to just skip and continue. We add a new feature that kafka client will commit the consumed offset back to kafka, so that the kafka can delete the messages before the committed offset. You can see this PR: #1734

[ZengRL]

可以增加routine load配置："property.auto.offset.reset" = "earliest"