Hazelcast-Jet与Kafka源和汇管道的稳定性如何？

Question

我正在运行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的数据也不一致。

Answer 1

这很可能是配置错误。通过阅读有关此问题的多个报告，可能是broker 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.