Azure eventthub Kafka org.apache.kafka.common.errors.TimeoutException 的一些记录

Question

有一个包含 80 到 100 条记录的 ArrayList，尝试将每个单独的记录（POJO，不是整个列表）流式传输并发送到 Kafka 主题（事件中心）。每隔一小时安排一次 cron 作业，将这些记录 (POJO) 发送到事件中心。

能够看到正在发送到 eventthub 的消息，但是在成功运行 3 到 4 次后出现以下异常（其中包括正在发送的几条消息和几条失败并出现以下异常）

    Expiring 14 record(s) for eventhubname: 30125  ms has passed since batch creation plus linger time

以下是使用的 Producer 的配置，

    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.ACKS_CONFIG, "1");
    props.put(ProducerConfig.RETRIES_CONFIG, "3");

消息保留期 - 7 分区 - 6 使用 spring Kafka(2.2.3) 发送事件 方法标记为@Async，其中写有kafka send

    @Async
    protected void send() {
       kafkatemplate.send(record);
    }

预期 - 不会从 kafka 抛出异常 实际 - org.apache.kafka.common.errors.TimeoutException 被抛出

Answer 1

Prakash - 我们已经看到了一些问题，其中尖峰的生产者模式会出现批处理超时。

这里的问题是生产者有两个 TCP 连接可以闲置超过 4 分钟 - 此时，Azure 负载均衡器会关闭空闲连接。 Kafka 客户端不知道连接已关闭，因此它尝试在死连接上发送批处理，该连接超时，此时重试开始。

• 将 connections.max.idle.ms 设置为
• 将 metadata.max.age.ms 设置为

请随时与 Github 上的 EH 产品团队联系，我们非常擅长回应问题 - https://github.com/Azure/azure-event-hubs-for-kafka

Answer 2

此异常表明您正在以比发送记录更快的速度排队记录。将记录添加到批次后，发送该批次有时间限制，以确保它已在指定的持续时间内发送。这由 Producer 配置参数 request.timeout.ms 控制。如果批处理的排队时间超过了超时限制，则会抛出异常。该批次中的记录将从发送队列中删除。

请检查以下类似问题，这可能会有所帮助。

Kafka producer TimeoutException: Expiring 1 record(s)

您也可以查看此链接

when-does-the-apache-kafka-client-throw-a-batch-expired-exception/34794261#34794261 for reason more details about batch expired exception.

同时实施适当的重试策略。

请注意，这不考虑扫描仪端的任何网络问题。由于网络问题，您将无法发送到任一集线器。

希望对你有帮助。