会话超时时Kafka侦听器回滚事务

Question

我正在使用spring-kafka版本1.1.3来使用来自主题的消息。在使用者配置中，自动提交设置为true，而max.poll.records设置为10。session.timeout.ms与服务器协商为10秒。

在收到消息后，我将它的一部分保存到数据库中。我的数据库有时会非常慢,这会导致kafka侦听器会话超时：

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

由于我无法增加服务器上的会话超时，并且max.poll.records已降至10，因此我希望能够将数据库调用包装在事务中，该事务将在kafka会话超时的情况下回滚。

这可能吗？我怎样才能做到这一点？

不幸的是，我在文件中找不到解决办法。

Answer 1

您必须考虑升级到Spring Kafka1.2和Kafka0.10.x。旧的阿帕奇Kafaka有一个缺陷与心跳。因此，使用autocommit和slow listener，您最终会出现意外的重新平衡，这样的问题就由您自己解决了。你用的SpringKafaka的版本有这样一个逻辑：

// if the container is set to auto-commit, then execute in the
// same thread
// otherwise send to the buffering queue
if (this.autoCommit) {
    invokeListener(records);
}
else {
    if (sendToListener(records)) {
        if (this.assignedPartitions != null) {
            // avoid group management rebalance due to a slow
            // consumer
            this.consumer.pause(this.assignedPartitions);
            this.paused = true;
            this.unsent = records;
        }
    }
}

因此，您可以考虑关闭autocommit，并依赖默认打开的内置pause特性。