啟動(dòng)Flink的sql-client.sh，創(chuàng)建Kafka的source端表，然后查詢Kafka的數(shù)據(jù)時(shí)報(bào)錯(cuò)。

2024-06-18 16:10:12

org.apache.flink.util.FlinkException: Global failure triggered by OperatorCoordinator for 'Source: kafka_rmc_cust_analog_u[1]' (operator bc764cd8ddf7a0cff126f51c16239658).

    at org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder$LazyInitializedCoordinatorContext.failJob(OperatorCoordinatorHolder.java:600)

    at org.apache.flink.runtime.operators.coordination.RecreateOnResetOperatorCoordinator$QuiesceableContext.failJob(RecreateOnResetOperatorCoordinator.java:237)

    at org.apache.flink.runtime.source.coordinator.SourceCoordinatorContext.failJob(SourceCoordinatorContext.java:374)

    at org.apache.flink.runtime.source.coordinator.SourceCoordinatorContext.handleUncaughtExceptionFromAsyncCall(SourceCoordinatorContext.java:387)

    at org.apache.flink.util.ThrowableCatchingRunnable.run(ThrowableCatchingRunnable.java:42)

    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)

    at java.util.concurrent.FutureTask.run(FutureTask.java:266)

    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)

    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)

    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: org.apache.flink.util.FlinkRuntimeException: Failed to list subscribed topic partitions due to

    at org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumerator.checkPartitionChanges(KafkaSourceEnumerator.java:234)

    at org.apache.flink.runtime.source.coordinator.ExecutorNotifier.lambda$null$1(ExecutorNotifier.java:83)

    at org.apache.flink.util.ThrowableCatchingRunnable.run(ThrowableCatchingRunnable.java:40)

    ... 7 more

Caused by: java.lang.NoSuchMethodError: org.apache.kafka.clients.admin.DescribeTopicsResult.allTopicNames()Lorg/apache/kafka/common/KafkaFuture;

    at org.apache.flink.connector.kafka.source.enumerator.subscriber.KafkaSubscriberUtils.getTopicMetadata(KafkaSubscriberUtils.java:44)

    at org.apache.flink.connector.kafka.source.subscriber.TopicListSubscriber.getSubscribedTopicPartitions(TopicListSubscriber.java:52)

    at org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumerator.getSubscribedTopicPartitions(KafkaSourceEnumerator.java:219)

    at org.apache.flink.runtime.source.coordinator.ExecutorNotifier.lambda$notifyReadyAsync$2(ExecutorNotifier.java:80)

    ... 7 more

 

由于Flink在與Kafka集成時(shí)，調(diào)用了org.apache.kafka.clients.admin.DescribeTopicsResult.allTopicNames()，但是目前的Kafka的client的Jar包中沒(méi)有該方法導(dǎo)致。也就是Flink版本與Kafka版本不兼容。

可以去Kafka官網(wǎng)下載源代碼包，查看./org/apache/kafka/clients/admin這個(gè)文件下的類：DescribeTopicsResult中是否含有allTopicNames()這個(gè)方法。

（1）、下載并安裝Kafka3.5.0以上版本即可。

（2）、如果Kafka已經(jīng)安裝，可以嘗試將Flink版本降低一點(diǎn)。