kafka-flink：Kafka 生产者错误取决于 flink 作业

Question

当我使用 kafka-flink 进行概念验证时，我发现了以下内容：似乎 kafka producer 错误可能是由于 flink 端完成的工作量而发生的？！

这里有更多细节：

我有像 sample??.EDR 这样的示例文件，由大约 700'000 行组成，其值如“entity”、“value”、“timestamp”

我使用以下命令创建kafka主题：

~/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --partitions 1 --replication-factor 1 --topic gprs

我使用以下命令加载主题的示例文件：

[13:00] kafka@ubu19: ~/fms
% /home/kafka/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic gprs < ~/sample/sample01.EDR

我在 flink 方面的工作是聚合每个实体的价值，滑动窗口为 6 小时和 72 小时（aggregationeachsix，aggregationeachsentytwo）。

我做了三个场景：

1. 在没有任何作业运行的情况下加载主题中的文件
2. 在运行aggregationeachsix 作业的主题中加载文件
3. 在运行 aggregationeachsix 和 aggregationeachsentytwo 作业的主题中加载文件

结果是前两个场景都在工作，但对于第三种场景，我在加载文件时在 kafka 生产者端出现以下错误（并不总是在同一个文件中，它可以是第一个、第二个、第三个甚至以后的文件）：

    [plenty of lines before this part]
    [2017-08-09 12:56:53,409] ERROR Error when sending message to topic gprs with key: null, value: 35 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
    org.apache.kafka.common.errors.TimeoutException: Expiring 233 record(s) for gprs-0: 1560 ms has passed since last append
    [2017-08-09 12:56:53,409] ERROR Error when sending message to topic gprs with key: null, value: 37 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
    org.apache.kafka.common.errors.TimeoutException: Expiring 233 record(s) for gprs-0: 1560 ms has passed since last append
    [2017-08-09 12:56:53,409] ERROR Error when sending message to topic gprs with key: null, value: 37 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
    org.apache.kafka.common.errors.TimeoutException: Expiring 233 record(s) for gprs-0: 1560 ms has passed since last append
    [2017-08-09 12:56:53,412] WARN Got error produce response with correlation id 1627 on topic-partition gprs-0, retrying (2 attempts left). Error: NETWORK_EXCEPTION (org.apache.kafka.clients.producer.internals.Sender)
    [2017-08-09 12:56:53,412] WARN Got error produce response with correlation id 1626 on topic-partition gprs-0, retrying (2 attempts left). Error: NETWORK_EXCEPTION (org.apache.kafka.clients.producer.internals.Sender)
    [2017-08-09 12:56:53,412] WARN Got error produce response with correlation id 1625 on topic-partition gprs-0, retrying (2 attempts left). Error: NETWORK_EXCEPTION (org.apache.kafka.clients.producer.internals.Sender)
    [2017-08-09 12:56:53,412] WARN Got error produce response with correlation id 1624 on topic-partition gprs-0, retrying (2 attempts left). Error: NETWORK_EXCEPTION (org.apache.kafka.clients.producer.internals.Sender)
    [2017-08-09 12:56:53,515] ERROR Error when sending message to topic gprs with key: null, value: 35 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
    org.apache.kafka.common.errors.TimeoutException: Expiring 8 record(s) for gprs-0: 27850 ms has passed since batch creation plus linger time
    [2017-08-09 12:56:53,515] ERROR Error when sending message to topic gprs with key: null, value: 37 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
    org.apache.kafka.common.errors.TimeoutException: Expiring 8 record(s) for gprs-0: 27850 ms has passed since batch creation plus linger time
    [plenty of lines after this part]

我的问题是为什么 flink 会对 kafka 生产者产生影响，然后，我需要改变什么来避免这个错误？

Answer 1

当 flink 和 kafka-producer 都在使用它时，您的网络似乎正在饱和，因此您会得到 TimeoutExceptions。