Flink Connector Exception: Failed to list subscribed topic partitions due to

[BohanZhang0222]

Env

Pulsar Version: 2.9.2
Flink Version: 1.14

##Source Conf

{"topicNames":["persistent://autopilot/x01/autopilot-filter-iot-signal-canlog-x01-testtwo-partition-0"],"topicsPattern":null,"subscriptionName":"ontest-lyt","subscriptionType":"Key_Shared","subscriptionMode":"Durable","receiverQueueSize":1000,"acknowledgementsGroupTimeMicros":100000,"negativeAckRedeliveryDelayMicros":60000000,"maxTotalReceiverQueueSizeAcrossPartitions":50000,"consumerName":null,"ackTimeoutMillis":0,"tickDurationMillis":1000,"priorityLevel":0,"maxPendingChunkedMessage":10,"autoAckOldestChunkedMessageOnQueueFull":false,"expireTimeOfIncompleteChunkedMessageMillis":60000,"cryptoFailureAction":"FA
IL","properties":{},"readCompacted":false,"subscriptionInitialPosition":"Latest","patternAutoDiscoveryPeriod":60,"regexSubscriptionMode":"PersistentOnly","deadLetterPolicy":null,"retryEnable":false,"autoUpdatePartitions":true,"autoUpdatePartitionsIntervalSeconds":60,"replicateSubscriptionState":false,"resetIncludeHead":false,"keySharedPolicy":{"keySharedMode":"STICKY","allowOutOfOrderDelivery":false,"ranges":[{"start":0,"end":65535}],"hashRangeTotal":65536},"batchIndexAckEnabled":false,"ackReceiptEnabled":false,"poolMessages":false,"maxPendingChuckedMessage":10}
2022-08-21 16:56:45,407 INFO  org.apache.pulsar.client.impl.ConsumerStatsRecorderImpl      [] - Pulsar client config: {"serviceUrl":"pulsar://bigdata-pulsar-ontest.dip.chj.cloud:6650","authPluginClassName":null,"authParams":null,"authParamMap":null,"operationTimeoutMs":30000,"lookupTimeoutMs":30000,"statsIntervalSeconds":60,"numIoThreads":1,"numListenerThreads":1,"connectionsPerBroker":1,"useTcpNoDelay":true,"useTls":false,"tlsTrustCertsFilePath":"","
tlsAllowInsecureConnection":false,"tlsHostnameVerificationEnable":false,"concurrentLookupRequest":5000,"maxLookupRequest":50000,"maxLookupRedirects":20,"maxNumberOfRejectedRequestPerConnection":50,"keepAliveIntervalSeconds":30,"connectionTimeoutMs":10000,"requestTimeoutMs":60000,"initialBackoffIntervalNanos":100000000,"maxBackoffIntervalNanos":60000000000,"enableBusyWait":false,"listenerName":null,"useKeyStoreTls":false,"sslProvider":null,"tlsTrustStoreType":"JKS","tlsTrustStorePath":null,"tlsTrustStorePassword":null,"tlsCiphers":[],"tlsProtocols":[],"memoryLimitBytes":0,"proxyServiceUrl":null,"proxyProtocol":null,"enableTransaction":false,"socks5ProxyAddress":null,"socks5ProxyUsername":null,"socks5ProxyPassword":null}

Exception

2022-08-21 17:11:14,147 ERROR org.apache.flink.runtime.source.coordinator.SourceCoordinatorContext [] - Exception while handling result from async call in SourceCoordinator-Source: Pulsar Source. Triggering job failover.
org.apache.flink.util.FlinkRuntimeException: Failed to list subscribed topic partitions due to 
	at org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumerator.checkPartitionChanges(PulsarSourceEnumerator.java:232) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.flink.runtime.source.coordinator.ExecutorNotifier.lambda$null$4(ExecutorNotifier.java:133) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.flink.util.ThrowableCatchingRunnable.run(ThrowableCatchingRunnable.java:40) [flink-dist_2.11-1.14.3.jar:1.14.3]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_162]
	at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_162]
	at java.util.concurrent.ScheduledThr
eadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) [?:1.8.0_162]
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) [?:1.8.0_162]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_162]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_162]
	at java.lang.Thread.run(Thread.java:748) [?:1.8.0_162]
Caused by: java.lang.IllegalStateException: org.apache.pulsar.client.admin.PulsarAdminException$ServerSideErrorException: HTTP 502 Bad Gateway
	at org.apache.flink.connector.pulsar.source.enumerator.subscriber.impl.BasePulsarSubscriber.queryTopicMetadata(BasePulsarSubscriber.java:53) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.flink.connector.pulsar.source.enumerator.subscriber.impl.TopicListSubscriber.lambda$getSubscribedTopicPartitions$0(TopicListSubscriber.java:48) ~[mylogpack-jar-with-dependencies.jar:?]
	at java.util.stre
am.ReferencePipeline$3$1.accept(ReferencePipeline.java:193) ~[?:1.8.0_162]
	at java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948) ~[?:1.8.0_162]
	at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481) ~[?:1.8.0_162]
	at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471) ~[?:1.8.0_162]
	at java.util.stream.ReduceOps$ReduceTask.doLeaf(ReduceOps.java:747) ~[?:1.8.0_162]
	at java.util.stream.ReduceOps$ReduceTask.doLeaf(ReduceOps.java:721) ~[?:1.8.0_162]
	at java.util.stream.AbstractTask.compute(AbstractTask.java:316) ~[?:1.8.0_162]
	at java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:731) ~[?:1.8.0_162]
	at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289) ~[?:1.8.0_162]
	at java.util.concurrent.ForkJoinTask.doInvoke(ForkJoinTask.java:401) ~[?:1.8.0_162]
	at java.util.concurrent.ForkJoinTask.invoke(ForkJoinTask.java:734) ~[?:1.8.0_162]
	at java.util.stream.ReduceOps$ReduceOp.evaluateParallel(ReduceOps.java:714) 
~[?:1.8.0_162]
	at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:233) ~[?:1.8.0_162]
	at java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:499) ~[?:1.8.0_162]
	at org.apache.flink.connector.pulsar.source.enumerator.subscriber.impl.TopicListSubscriber.getSubscribedTopicPartitions(TopicListSubscriber.java:55) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumerator.getSubscribedTopicPartitions(PulsarSourceEnumerator.java:172) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.flink.runtime.source.coordinator.ExecutorNotifier.lambda$notifyReadyAsync$5(ExecutorNotifier.java:130) ~[mylogpack-jar-with-dependencies.jar:?]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[?:1.8.0_162]
	at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308) ~[?:1.8.0_162]
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180) ~
[?:1.8.0_162]
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294) ~[?:1.8.0_162]
	... 3 more
Caused by: org.apache.pulsar.client.admin.PulsarAdminException$ServerSideErrorException: HTTP 502 Bad Gateway
	at org.apache.pulsar.client.admin.internal.BaseResource.getApiException(BaseResource.java:219) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.client.admin.internal.TopicsImpl$7.failed(TopicsImpl.java:529) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.org.glassfish.jersey.client.JerseyInvocation$1.failed(JerseyInvocation.java:882) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.org.glassfish.jersey.client.JerseyInvocation$1.completed(JerseyInvocation.java:863) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.org.glassfish.jersey.client.ClientRuntime.processResponse(ClientRuntime.java:229) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.org.glassfish
.jersey.client.ClientRuntime.access$200(ClientRuntime.java:62) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.org.glassfish.jersey.client.ClientRuntime$2.lambda$response$0(ClientRuntime.java:173) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.org.glassfish.jersey.internal.Errors$1.call(Errors.java:248) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.org.glassfish.jersey.internal.Errors$1.call(Errors.java:244) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.org.glassfish.jersey.internal.Errors.process(Errors.java:292) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.org.glassfish.jersey.internal.Errors.process(Errors.java:274) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.org.glassfish.jersey.internal.Errors.process(Errors.java:244) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.org.glassfish.jersey.process.internal.RequestScope.runInScope(RequestScope.java:288)
 ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.org.glassfish.jersey.client.ClientRuntime$2.response(ClientRuntime.java:173) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.client.admin.internal.http.AsyncHttpConnector.lambda$apply$1(AsyncHttpConnector.java:228) ~[mylogpack-jar-with-dependencies.jar:?]
	at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:760) ~[?:1.8.0_162]
	at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:736) ~[?:1.8.0_162]
	at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474) ~[?:1.8.0_162]
	at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1962) ~[?:1.8.0_162]
	at org.apache.pulsar.client.admin.internal.http.AsyncHttpConnector.lambda$retryOperation$4(AsyncHttpConnector.java:270) ~[mylogpack-jar-with-dependencies.jar:?]
	at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:760) ~[?:1.8.0_162]
	at java.uti
l.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:736) ~[?:1.8.0_162]
	at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474) ~[?:1.8.0_162]
	at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1962) ~[?:1.8.0_162]
	at org.apache.pulsar.shade.org.asynchttpclient.netty.NettyResponseFuture.loadContent(NettyResponseFuture.java:222) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.org.asynchttpclient.netty.NettyResponseFuture.done(NettyResponseFuture.java:257) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.org.asynchttpclient.netty.handler.AsyncHttpClientHandler.finishUpdate(AsyncHttpClientHandler.java:241) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.org.asynchttpclient.netty.handler.HttpHandler.handleChunk(HttpHandler.java:114) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.org.asynchttpclient.netty.handler.HttpHandler.handleRead(HttpHandler.java:1
43) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.org.asynchttpclient.netty.handler.AsyncHttpClientHandler.channelRead(AsyncHttpClientHandler.java:78) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java
:379) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:327) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:299) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandle
r.java:251) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHa
ndlerContext.java:365) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:722) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:658) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:584) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:496) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.util.concurrent.SingleTh
readEventExecutor$4.run(SingleThreadEventExecutor.java:986) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[mylogpack-jar-with-dependencies.jar:?]
	at org.apache.pulsar.shade.io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[mylogpack-jar-with-dependencies.jar:?]
	... 1 more

Other references

yarn log
shadow_model_run_mile_pulsar_to_mysql.log

maven pom
pom(9).xml.zip

[tisonkun]

@BohanZhang0222 code rendered informally. Could you quote code in a fenced block?

cc @syhily @imaffe

[syhily]

It's tracked here. streamnative/flink#162

[syhily]

@BohanZhang0222 I think the root cause should be some performance issues on Pulsar side. Since flink connector requests a lot of metadata informations in an extremely short period, the 5xx error could happen easily. We will add rate limit and retry logic on connector side. But the core fix should be accomplished on the Pulsar side.

You can track the fix progress on the connector here. streamnative/flink#162

cc @codelipenghui