当检查点被禁用时，Kafka010JsonTableSource将不会自动提交kafka偏移量

Question

我已经设置了一个本地的Kafka0.10+Flink1.4环境。

我使用下面的代码来消费来自Kafka主题的数据：

val tableSource:KafkaTableSource = Kafka010JsonTableSource.builder()
.forTopic(kafkaConfig.topic)
.withKafkaProperties(props)
.withSchema(dynamicJsonSchema)
.withRowtimeAttribute(enventTimeFieldName,new ExistingField(enventTimeFieldName), new BoundedOutOfOrderTimestamps(30000L))
.build() 

tableEnv.registerTableSource(tableName, tableSource)

val tableResult:Table = tableEnv.sqlQuery(sql)

在我执行这段代码后，总是会发现警告消息：

自动提交组TaxidataGroup的偏移量{taxidata-0=offsetandMetadata{offset=728461,Metadata=“}}失败：无法完成提交，因为该组已重新平衡并将分区分配给其他成员。这意味着对poll（）的后续调用之间的时间比配置的max.poll.interval.ms长，这通常意味着poll循环花费了太多时间处理消息。您可以通过增加会话超时，或者通过使用max.poll.records减小poll（）中返回的批的最大大小来解决这一问题。

无论我在Kafka中设置了什么属性，它总是显示上面的警告消息。

  {
    "propertyKey": "enable.auto.commit",
    "propertyValue": "true"
  },
  {
    "propertyKey": "session.timeout.ms",
    "propertyValue": "250000"
  },
  {
    "propertyKey": "request.timeout.ms",
    "propertyValue": "305000"
  },
  {
    "propertyKey": "auto.commit.interval.ms",
    "propertyValue": "800000"
  },
  {
    "propertyKey": "max.poll.records",
    "propertyValue": "300"
  },
  {
    "propertyKey": "max.poll.interval.ms",
    "propertyValue": "300000"
  }

我不确定如果Kafka010JsonTableSource flink1.4会自动提交偏移量。但测试结果表明它不会自动提交偏移量。有谁能帮忙证实一下这个问题吗？或者你能在我的代码中看到任何其他问题吗？

Answer 1

是否尝试设置低于代理的group.max.session.timeout.ms值的session.timeout.ms值？根据https://github.com/dpkp/kafka-python/issues/746，这似乎是问题所在。