CommitFailedException：无法完成提交，因为组已重新平衡

Question

环境:Hadoop2.75.+Flink1.4+Kafka0.10

我已经建立了一个实时数据处理项目。我使用Flink表源API(Kafka010JsonTableSource）作为表源。从kafka获取数据，然后执行一个SQL，最后输出到一个kafka主题。这是一个清晰的流程，但是我在Flink集群上执行时遇到了异常，下面是我的主要代码：

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) 
env.getConfig.setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000))
env.enableCheckpointing(5000) 
val tableEnv = TableEnvironment.getTableEnvironment(env)
val tableSource:KafkaTableSource = Kafka010JsonTableSource.builder()
    .forTopic(kafkaConfig.topic)
    .withKafkaProperties(props)
    .withSchema(dynamicJsonSchema)
    .withRowtimeAttribute(
         enventTimeFieldName,  
         new ExistingField(enventTimeFieldName),  
         new MyBoundedOutOfOrderTimestamps(100L)) 
    .build()
tableEnv.registerTableSource(tableName, tableSource)
val tableResult:Table = tableEnv.sqlQuery(sql)
tableResult.writeToSink(new Kafka010JsonTableSink(kafkaOutput.topic, props))

我已经启用了检查点。第一次在flink上执行时，我只是遵循Consumer的默认配置。在Flink任务运行之后，我通过kafka shell命令（kafka-consumer-groups.sh）检查了偏移量，发现了一个奇怪的情况。根据shell命令的输出和Flink任务管理器的日志，我发现偏移量在几秒钟开始时提交成功，但后来我继续遇到许多异常，如下所示：

于是我根据上面的错误信息搜索解决方案，有人告诉我应该增加session.timeout.ms，然后我就照做了，但还是失败了。之后我尝试了以下多种组合配置来测试，kafka偏移量总是在开始的时候提交成功，但是后来提交的will失败了。我真的不知道解决它，你能帮我修好它吗？非常感谢！！！！！！

kafka使用者配置组合如下:{“PropertyKey”：“Session.Timeout.ms”,“PropertyValue”：“300000”},{“PropertyKey”：“Request.Timeout.ms”,“PropertyValue”：“505000”},{“PropertyKey”：“Auto.Commit.Interval.ms”,“PropertyValue”：“10000”},{“PropertyKey”：“Max.Poll.Records”,“PropertyValue”：“50”},{“PropertyKey”：“我希望您能帮助修复以上错误，非常感谢！！！




Answer 1

我找到根本原因了。重新平衡错误总是发生的原因是两个消费者（一个是消费者输入数据，另一个是消费者输出数据）的组名是相同的。我怀疑仅仅一个协调器还不足以处理两个消费者的偏移量提交操作。当我更改了一个消费者的组名后，世界突然变得安静了。错误从未发生过。