Kafka·Flink：Kafka制作人错误取决于Flink作业

Question

当我与Kafka·Flink一起进行概念验证时，我发现了以下几点：似乎Kafka制作人的错误可能会由于Flink方面的工作量而发生？！

以下是更多细节：

我有样本文件，如样本？？。EDR由约700000行组成，具有“实体”、“值”、“时间戳”等值

我使用以下命令创建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制作人都在使用它时，你的网络似乎已经饱和了，因此你得到了超时异常。