Hazelcast-Jet 与 Kafka 源和接收器管道的稳定性如何？答案

【问题标题】：how stable is Hazelcast-Jet with Kafka source and sink pipelines?Hazelcast-Jet 与 Kafka 源和接收器管道的稳定性如何？
【发布时间】：2019-09-30 10:57:30
【问题描述】：

我正在运行 Hazelcast-Jet 的 3 节点嵌入式集群，控制台中经常出现以下错误。可能的原因是什么？

 [jet] [3.0] Execution of job '15ba-4fbe-1b73-9ed1', execution 61d7-46eb-5875-8799 failed after 60,112 ms
    com.hazelcast.jet.JetException: Exception in ProcessorTasklet{streamKafka#1}: org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata
            at com.hazelcast.jet.impl.execution.TaskletExecutionService$BlockingWorker.run(TaskletExecutionService.java:250)
            at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
            at java.util.concurrent.FutureTask.run(FutureTask.java:266)
            at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
            at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
            at java.lang.Thread.run(Thread.java:748)
            at ------ submitted from ------.(Unknown Source)
            at com.hazelcast.spi.impl.operationservice.impl.InvocationFuture.resolve(InvocationFuture.java:126)
            at com.hazelcast.spi.impl.AbstractInvocationFuture$1.run(AbstractInvocationFuture.java:251)
            at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
            at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
            at java.lang.Thread.run(Thread.java:748)
            at com.hazelcast.util.executor.HazelcastManagedThread.executeRun(HazelcastManagedThread.java:64)
            at com.hazelcast.util.executor.HazelcastManagedThread.run(HazelcastManagedThread.java:80)
            at ------ submitted from ------.(Unknown Source)
            at com.hazelcast.spi.impl.operationservice.impl.InvocationFuture.resolve(InvocationFuture.java:126)
            at com.hazelcast.spi.impl.AbstractInvocationFuture$1.run(AbstractInvocationFuture.java:251)
            at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
            at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
            at java.lang.Thread.run(Thread.java:748)
            at com.hazelcast.util.executor.HazelcastManagedThread.executeRun(HazelcastManagedThread.java:64)
            at com.hazelcast.util.executor.HazelcastManagedThread.run(HazelcastManagedThread.java:80)
    Caused by: org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata

May 14, 2019 5:36:13 AM com.hazelcast.jet.impl.MasterJobContext
SEVERE: [127.0.0.1]:5701 [jet] [3.0] Execution of job '4940-dffe-4fd6-2f43', execution 2b9a-1f3d-4ecc-e116 failed after 60,209 ms
com.hazelcast.jet.JetException: Exception in ProcessorTasklet{streamKafka#1}: org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata
        at com.hazelcast.jet.impl.execution.TaskletExecutionService$BlockingWorker.run(TaskletExecutionService.java:250)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
        at ------ submitted from ------.(Unknown Source)
        at com.hazelcast.spi.impl.operationservice.impl.InvocationFuture.resolve(InvocationFuture.java:126)
        at com.hazelcast.spi.impl.AbstractInvocationFuture$1.run(AbstractInvocationFuture.java:251)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
        at com.hazelcast.util.executor.HazelcastManagedThread.executeRun(HazelcastManagedThread.java:64)
        at com.hazelcast.util.executor.HazelcastManagedThread.run(HazelcastManagedThread.java:80)
Caused by: org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata

谁能帮我理解一下？

Kafka-source 和 sink 的数据也不一致。

【问题讨论】：

看起来KafkaConsumer 无法获取元数据，原因的堆栈跟踪丢失。问题可能与 Jet 无关。这可能是一个答案：stackoverflow.com/a/55277943/952135，看起来这个问题经常出现在许多使用 kafka 的系统中
@Oliv 虽然信息不多，但用完整的堆栈跟踪更新了问题。
原因的堆栈跟踪仍然丢失。我的意思是Caused by: org.apache.kafka.common.errors.TimeoutException 之后的行
@Oliv 这就是它在控制台中显示的内容。
查看了其他报告，在所有报告中，此异常都缺少堆栈跟踪。这很奇怪。

标签： java hazelcast-jet

【解决方案1】：

这很可能是配置错误。从阅读该问题的多个报告来看，它可能是错误的代理 URL、错误的 SSH 配置、网络故障或类似情况。 Kafka 客户端不会立即报告连接错误，而是会重试连接，直到最终超时。

您应该有来自 Kafka 客户端的其他日志，这些日志应该重复显示原因，确保您启用了 Kafka 日志记录。

[kafka-producer-network-thread | producer-1] WARN org.apache.kafka.clients.NetworkClient - [Producer clientId=producer-1] Connection to node -1 (/127.0.0.1:55561) could not be established. Broker may not be available.
[kafka-producer-network-thread | producer-1] WARN org.apache.kafka.clients.NetworkClient - [Producer clientId=producer-1] Connection to node -1 (/127.0.0.1:55561) could not be established. Broker may not be available.
[kafka-producer-network-thread | producer-1] WARN org.apache.kafka.clients.NetworkClient - [Producer clientId=producer-1] Connection to node -1 (/127.0.0.1:55561) could not be established. Broker may not be available.
[kafka-producer-network-thread | producer-1] WARN org.apache.kafka.clients.NetworkClient - [Producer clientId=producer-1] Connection to node -1 (/127.0.0.1:55561) could not be established. Broker may not be available.
[kafka-producer-network-thread | producer-1] WARN org.apache.kafka.clients.NetworkClient - [Producer clientId=producer-1] Connection to node -1 (/127.0.0.1:55561) could not be established. Broker may not be available.
[kafka-producer-network-thread | producer-1] WARN org.apache.kafka.clients.NetworkClient - [Producer clientId=producer-1] Connection to node -1 (/127.0.0.1:55561) could not be established. Broker may not be available.
[kafka-producer-network-thread | producer-1] WARN org.apache.kafka.clients.NetworkClient - [Producer clientId=producer-1] Connection to node -1 (/127.0.0.1:55561) could not be established. Broker may not be available.
[kafka-producer-network-thread | producer-1] WARN org.apache.kafka.clients.NetworkClient - [Producer clientId=producer-1] Connection to node -1 (/127.0.0.1:55561) could not be established. Broker may not be available.
[kafka-producer-network-thread | producer-1] WARN org.apache.kafka.clients.NetworkClient - [Producer clientId=producer-1] Connection to node -1 (/127.0.0.1:55561) could not be established. Broker may not be available.
[kafka-producer-network-thread | producer-1] WARN org.apache.kafka.clients.NetworkClient - [Producer clientId=producer-1] Connection to node -1 (/127.0.0.1:55561) could not be established. Broker may not be available.
... lot more of these

10:40:40,681 DEBUG || - [JobExecutionService] hz._hzInstance_1_jet.jet.blocking.thread-2 - [127.0.0.1]:5701 [jet] [3.1-SNAPSHOT] Execution of job '336c-68c3-7b9d-0c2f', execution 5c43-a795-8172-bc57 completed with failure
java.util.concurrent.CompletionException: com.hazelcast.jet.JetException: Exception in ProcessorTasklet{writeKafka(7f147d66-3952-4e86-980a-226cc8e6ac9b)#1}: org.apache.kafka.common.errors.TimeoutException: Topic 7f147d66-3952-4e86-980a-226cc8e6ac9b not present in metadata after 60000 ms.
    at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:292)
    at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:308)
    at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:593)
    at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
    at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474)
    at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:1977)
    at com.hazelcast.jet.impl.util.NonCompletableFuture.internalCompleteExceptionally(NonCompletableFuture.java:59)
    at com.hazelcast.jet.impl.execution.TaskletExecutionService$ExecutionTracker.taskletDone(TaskletExecutionService.java:398)
    at com.hazelcast.jet.impl.execution.TaskletExecutionService$BlockingWorker.run(TaskletExecutionService.java:255)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:266)
    at java.util.concurrent.FutureTask.run(FutureTask.java)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
Caused by: com.hazelcast.jet.JetException: Exception in ProcessorTasklet{writeKafka(7f147d66-3952-4e86-980a-226cc8e6ac9b)#1}: org.apache.kafka.common.errors.TimeoutException: Topic 7f147d66-3952-4e86-980a-226cc8e6ac9b not present in metadata after 60000 ms.
    at com.hazelcast.jet.impl.execution.TaskletExecutionService$BlockingWorker.run(TaskletExecutionService.java:250)
    ... 6 more
Caused by: org.apache.kafka.common.errors.TimeoutException: Topic 7f147d66-3952-4e86-980a-226cc8e6ac9b not present in metadata after 60000 ms.

【讨论】：