Thanks for the update. > Could the backpressure timeout and heartbeat timeout be because of Heap Usage close to Max configured?
Could be. This is one of the things I had in mind under overloaded in: > might be related to one another via some different deeper problem (broken network environment, something being overloaded) You can easily diagnose it. Just attach a memory profiler or check gc logs, just as you would normally do when debugging a non-Flink standalone Java application. It can also be a symptom of a failing network environment. I would first check for GC pauses/stops/gaps in the logs that would indicate stalled JVM caused those RPC timeouts. If that doesn't bring you closer to a solution I would then check for the network environment in your cluster/cloud. Both of those might be a reason behind your Kafka issues. Hard to tell. Definitely you shouldn't have heartbeat timeouts in your cluster, so something IS wrong with your setup. Best, Piotrek czw., 15 lip 2021 o 17:17 Rahul Patwari <rahulpatwari8...@gmail.com> napisał(a): > Thanks for the feedback Piotrek. > > We have observed the issue again today. As we are using Flink 1.11.1, I > tried to check the backpressure of Kafka source tasks from the > Jobmanager UI. > The backpressure request was canceled due to Timeout and "No Data" was > displayed in UI. Here are the respective logs: > > java.util.concurrent.TimeoutException: Invocation of public abstract > java.util.concurrent.CompletableFuture > org.apache.flink.runtime.taskexecutor.TaskExecutorGateway.requestTaskBackPressure(org.apache.flink.runtime.executiongraph.ExecutionAttemptID,int,org.apache.flink.api.common.time.Time) > timed out. > at > org.apache.flink.runtime.jobmaster.RpcTaskManagerGateway.requestTaskBackPressure(RpcTaskManagerGateway.java:67) > ............................. > Caused by: akka.pattern.AskTimeoutException: Ask timed out on > [Actor[akka.tcp://flink@xX.X.X.X:X/user/rpc/taskmanager_0#-1457664622]] > after [15000 ms]. Message of type > [org.apache.flink.runtime.rpc.messages.RemoteRpcInvocation]. A typical > reason for `AskTimeoutException` is that the recipient actor didn't send a > reply. > at > akka.pattern.PromiseActorRef$.$anonfun$defaultOnTimeout$1(AskSupport.scala:635) > ~[flink-dist_2.12-1.11.1.jar:1.11.1] > ............................. > > During this time, the heartbeat of one of the Taskmanager to the > Jobmanager timed out. Here are the respective logs: > > java.util.concurrent.TimeoutException: Heartbeat of TaskManager with id > bead57c15b447eac08531693ec91edc4 timed out. at > org.apache.flink.runtime.jobmaster.JobMaster$TaskManagerHeartbeatListener.notifyHeartbeatTimeout(JobMaster.java:1193) > .................. > > Because of heartbeat timeout, there was an internal restart of Flink and > the Kafka consumption rate recovered after the restart. > > Could the backpressure timeout and heartbeat timeout be because of Heap > Usage close to Max configured? > > On Wed, Jul 14, 2021 at 6:29 PM Piotr Nowojski <pnowoj...@apache.org> > wrote: > >> Hi Rahul, >> >> I would highly doubt that you are hitting the network bottleneck case. It >> would require either a broken environment/network or throughputs in orders >> of GB/second. More likely you are seeing empty input pool and you haven't >> checked the documentation [1]: >> >> > inPoolUsage - An estimate of the input buffers usage. (ignores >> LocalInputChannels) >> >> If local channels are backpressured, inPoolUsage will be 0. You can check >> downstream task's inputQueueLength or isBackPressured metrics. Besides >> that, I would highly recommend upgrading to Flink 1.13.x if you are >> investigating backpressure problems as described in the blog post. >> >> > 1. Can the backpressure Cause "DisconnectException", "Error Sending >> Fetch Request to node ..." and other Kafka Consumer logs mentioned above? >> >> No, I don't think it's possible. Those two might be related to one >> another via some different deeper problem (broken network environment, >> something being overloaded), but I don't see a way how one could cause the >> other. >> >> Piotrek >> >> [1] >> https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/ops/metrics/#default-shuffle-service >> >> śr., 14 lip 2021 o 14:18 Rahul Patwari <rahulpatwari8...@gmail.com> >> napisał(a): >> >>> Thanks, Piotrek. >>> >>> We have two Kafka sources. We are facing this issue for both of them. >>> The downstream tasks with the sources form two independent directed acyclic >>> graphs, running within the same Streaming Job. >>> >>> For Example: >>> source1 -> task1 -> sink1 >>> source2 -> task2 -> sink2 >>> >>> There is backpressure in both sources. Verified using the >>> "isBackPressured" metric. >>> For one of the sources, "outPoolUsage" is high whereas "inPoolUsage" of >>> immediate downstream task is ~ 0. I think we are observing the rare case >>> mentioned at the end in [1]. >>> >>> I have a couple of questions: >>> >>> 1. Can the backpressure Cause "DisconnectException", "Error Sending >>> Fetch Request to node ..." and other Kafka Consumer logs mentioned above? >>> 2. What could be the next steps in resolving the backpressure issue >>> - the rare case >>> >>> [1] https://flink.apache.org/2021/07/07/backpressure.html >>> >>> When the stream is running as expected, these are the thread dump of >>> Kafka Source tasks. Comparing the thread dumps - The "Kafka Fetcher" >>> thread, which polls records is blocked by "Legacy Source" Thread(main >>> Thread) - probably because of backpressure. >>> >>> { >>> "threadName": "Kafka Fetcher for Source: SourceEventSignature (8/12)", >>> "stringifiedThreadInfo": "\"Kafka Fetcher for Source: >>> SourceEventSignature (8/12)\" Id=521 RUNNABLE\n\tat >>> sun.nio.ch.EPollArrayWrapper.epollWait(Native Method)\n\tat >>> sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)\n\tat >>> sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)\n\tat >>> sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)\n\t- locked >>> sun.nio.ch.Util$3@1aaa5678\n\t- locked >>> java.util.Collections$UnmodifiableSet@6c629341\n\t- locked >>> sun.nio.ch.EPollSelectorImpl@3e783067\n\tat >>> sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)\n\tat >>> org.apache.kafka.common.network.Selector.select(Selector.java:794)\n\tat >>> org.apache.kafka.common.network.Selector.poll(Selector.java:467)\n\tat >>> org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:547)\n\t...\n\n\tNumber >>> of locked synchronizers = 1\n\t- >>> java.util.concurrent.locks.ReentrantLock$FairSync@462572c9\n\n" >>> }, { >>> "threadName": "Kafka Fetcher for Source: SourceEventTransition (4/12)", >>> "stringifiedThreadInfo": "\"Kafka Fetcher for Source: >>> SourceEventTransition (4/12)\" Id=520 RUNNABLE\n\tat >>> sun.nio.ch.EPollArrayWrapper.epollWait(Native Method)\n\tat >>> sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)\n\tat >>> sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)\n\tat >>> sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)\n\t- locked >>> sun.nio.ch.Util$3@ef4e5e3\n\t- locked >>> java.util.Collections$UnmodifiableSet@767487e7\n\t- locked >>> sun.nio.ch.EPollSelectorImpl@9707a46\n\tat >>> sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)\n\tat >>> org.apache.kafka.common.network.Selector.select(Selector.java:794)\n\tat >>> org.apache.kafka.common.network.Selector.poll(Selector.java:467)\n\tat >>> org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:547)\n\t...\n\n\tNumber >>> of locked synchronizers = 1\n\t- >>> java.util.concurrent.locks.ReentrantLock$FairSync@642ee29b\n\n" >>> }, { >>> "threadName": "Legacy Source Thread - Source: SourceEventSignature >>> (8/12)", >>> "stringifiedThreadInfo": "\"Legacy Source Thread - Source: >>> SourceEventSignature (8/12)\" Id=515 WAITING on >>> java.lang.Object@4d5cc800\n\tat >>> java.lang.Object.wait(Native Method)\n\t- waiting on >>> java.lang.Object@4d5cc800\n\tat >>> java.lang.Object.wait(Object.java:502)\n\tat >>> org.apache.flink.streaming.connectors.kafka.internal.Handover.pollNext(Handover.java:74)\n\tat >>> org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.runFetchLoop(KafkaFetcher.java:133)\n\tat >>> org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:755)\n\tat >>> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100)\n\tat >>> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63)\n\tat >>> org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:201)\n\n" >>> }, { >>> "threadName": "Legacy Source Thread - Source: SourceEventTransition >>> (4/12)", >>> "stringifiedThreadInfo": "\"Legacy Source Thread - Source: >>> SourceEventTransition (4/12)\" Id=514 WAITING on >>> java.lang.Object@1fc525f3\n\tat >>> java.lang.Object.wait(Native Method)\n\t- waiting on >>> java.lang.Object@1fc525f3\n\tat >>> java.lang.Object.wait(Object.java:502)\n\tat >>> org.apache.flink.streaming.connectors.kafka.internal.Handover.pollNext(Handover.java:74)\n\tat >>> org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.runFetchLoop(KafkaFetcher.java:133)\n\tat >>> org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:755)\n\tat >>> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100)\n\tat >>> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63)\n\tat >>> org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:201)\n\n" >>> } >>> >>> On Wed, Jul 14, 2021 at 2:39 PM Piotr Nowojski <pnowoj...@apache.org> >>> wrote: >>> >>>> Hi, >>>> >>>> Waiting for memory from LocalBufferPool is a perfectly normal symptom >>>> of a backpressure [1][2]. >>>> >>>> Best, >>>> Piotrek >>>> >>>> [1] https://flink.apache.org/2021/07/07/backpressure.html >>>> [2] https://www.ververica.com/blog/how-flink-handles-backpressure >>>> >>>> śr., 14 lip 2021 o 06:05 Rahul Patwari <rahulpatwari8...@gmail.com> >>>> napisał(a): >>>> >>>>> Thanks, David, Piotr for your reply. >>>>> >>>>> I managed to capture the Thread dump from Jobmanaager UI for few task >>>>> managers. >>>>> Here is the thread dump for Kafka Source tasks in one task manager. I >>>>> could see the same stack trace in other task managers as well. It seems >>>>> like Kafka Source tasks are waiting on Memory. Any Pointers? >>>>> >>>>> { >>>>> "threadName": "Kafka Fetcher for Source: SourceEventTransition (6/12)", >>>>> "stringifiedThreadInfo": "\"Kafka Fetcher for Source: >>>>> SourceEventTransition (6/12)\" Id=581 WAITING on >>>>> java.lang.Object@444c0edc\n\tat >>>>> java.lang.Object.wait(Native Method)\n\t- waiting on >>>>> java.lang.Object@444c0edc\n\tat >>>>> java.lang.Object.wait(Object.java:502)\n\tat >>>>> org.apache.flink.streaming.connectors.kafka.internal.Handover.produce(Handover.java:117)\n\tat >>>>> org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread.run(KafkaConsumerThread.java:261)\n\n" >>>>> }, { >>>>> "threadName": "Kafka Fetcher for Source: SourceEventSignature (7/12)", >>>>> "stringifiedThreadInfo": "\"Kafka Fetcher for Source: >>>>> SourceEventSignature (7/12)\" Id=580 WAITING on >>>>> java.lang.Object@7d3843a9\n\tat >>>>> java.lang.Object.wait(Native Method)\n\t- waiting on >>>>> java.lang.Object@7d3843a9\n\tat >>>>> java.lang.Object.wait(Object.java:502)\n\tat >>>>> org.apache.flink.streaming.connectors.kafka.internal.Handover.produce(Handover.java:117)\n\tat >>>>> org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread.run(KafkaConsumerThread.java:261)\n\n" >>>>> }, { >>>>> "threadName": "Legacy Source Thread - Source: SourceEventSignature >>>>> (7/12)", >>>>> "stringifiedThreadInfo": "\"Legacy Source Thread - Source: >>>>> SourceEventSignature (7/12)\" Id=408 WAITING on >>>>> java.util.concurrent.CompletableFuture$Signaller@4c613ed7\n\tat >>>>> sun.misc.Unsafe.park(Native Method)\n\t- waiting on >>>>> java.util.concurrent.CompletableFuture$Signaller@4c613ed7\n\tat >>>>> java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)\n\tat >>>>> java.util.concurrent.CompletableFuture$Signaller.block(CompletableFuture.java:1707)\n\tat >>>>> java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:3323)\n\tat >>>>> java.util.concurrent.CompletableFuture.waitingGet(CompletableFuture.java:1742)\n\tat >>>>> java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1908)\n\tat >>>>> org.apache.flink.runtime.io.network.buffer.LocalBufferPool.requestMemorySegmentBlocking(LocalBufferPool.java:293)\n\tat >>>>> org.apache.flink.runtime.io.network.buffer.LocalBufferPool.requestBufferBuilderBlocking(LocalBufferPool.java:266)\n\t...\n\n" >>>>> }, { >>>>> "threadName": "Legacy Source Thread - Source: SourceEventTransition >>>>> (6/12)", >>>>> "stringifiedThreadInfo": "\"Legacy Source Thread - Source: >>>>> SourceEventTransition (6/12)\" Id=409 WAITING on >>>>> java.util.concurrent.CompletableFuture$Signaller@5765d0d4\n\tat >>>>> sun.misc.Unsafe.park(Native Method)\n\t- waiting on >>>>> java.util.concurrent.CompletableFuture$Signaller@5765d0d4\n\tat >>>>> java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)\n\tat >>>>> java.util.concurrent.CompletableFuture$Signaller.block(CompletableFuture.java:1707)\n\tat >>>>> java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:3323)\n\tat >>>>> java.util.concurrent.CompletableFuture.waitingGet(CompletableFuture.java:1742)\n\tat >>>>> java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1908)\n\tat >>>>> org.apache.flink.runtime.io.network.buffer.LocalBufferPool.requestMemorySegmentBlocking(LocalBufferPool.java:293)\n\tat >>>>> org.apache.flink.runtime.io.network.buffer.LocalBufferPool.requestBufferBuilderBlocking(LocalBufferPool.java:266)\n\t...\n\n" >>>>> } >>>>> >>>>> On Tue, Jul 13, 2021 at 7:07 PM Piotr Nowojski <pnowoj...@apache.org> >>>>> wrote: >>>>> >>>>>> Hi, >>>>>> >>>>>> I'm not sure, maybe someone will be able to help you, but it sounds >>>>>> like it would be better for you to: >>>>>> - google search something like "Kafka Error sending fetch request >>>>>> TimeoutException" (I see there are quite a lot of results, some of them >>>>>> might be related) >>>>>> - ask this question on the Kafka mailing list >>>>>> - ask this question on stackoverflow as a Kafka question >>>>>> >>>>>> In short, FlinkKafkaConsumer is a very thin wrapper around the >>>>>> KafkaConsumer class, so the thing you are observing has most likely very >>>>>> little to do with the Flink itself. In other words, if you are observing >>>>>> such a problem you most likely would be possible to reproduce it without >>>>>> Flink. >>>>>> >>>>>> Best, >>>>>> Piotrek >>>>>> >>>>>> pt., 9 lip 2021 o 12:30 Rahul Patwari <rahulpatwari8...@gmail.com> >>>>>> napisał(a): >>>>>> >>>>>>> Hi, >>>>>>> >>>>>>> We have a Flink 1.11.1 Version streaming pipeline in production >>>>>>> which reads from Kafka. >>>>>>> Kafka Server version is 2.5.0 - confluent 5.5.0 >>>>>>> Kafka Client Version is 2.4.1 - >>>>>>> {"component":"org.apache.kafka.common.utils.AppInfoParser$AppInfo","message":"Kafka >>>>>>> version: 2.4.1","method":"<init>"} >>>>>>> >>>>>>> Occasionally(every 6 to 12 hours), we have observed that the Kafka >>>>>>> consumption rate went down(NOT 0) and the following logs were observed: >>>>>>> Generally, the consumption rate across all consumers is 4k records/sec. >>>>>>> When this issue occurred, the consumption rate dropped to < 50 >>>>>>> records/sec >>>>>>> >>>>>>> org.apache.kafka.common.errors.DisconnectException: null >>>>>>> >>>>>>> {"time":"2021-07-07T22:13:37,385","severity":"INFO","component":"org.apache.kafka.clients.FetchSessionHandler","message":"[Consumer >>>>>>> clientId=consumer-MFTDataProcessorEventSignatureConsumerGroupV1R1-3, >>>>>>> groupId=MFTDataProcessorEventSignatureConsumerGroupV1R1] Error sending >>>>>>> fetch request (sessionId=405798138, epoch=5808) to node 8: >>>>>>> {}.","method":"handleError"} >>>>>>> >>>>>>> org.apache.kafka.common.errors.TimeoutException: Failed >>>>>>> >>>>>>> {"time":"2021-07-07T22:26:41,379","severity":"INFO","component":"org.apache.kafka.clients.consumer.internals.AbstractCoordinator","message":"[Consumer >>>>>>> clientId=consumer-MFTDataProcessorEventSignatureConsumerGroupV1R1-3, >>>>>>> groupId=MFTDataProcessorEventSignatureConsumerGroupV1R1] Group >>>>>>> coordinator >>>>>>> 100.98.40.16:9092 (id: 2147483623 rack: null) is unavailable or >>>>>>> invalid, will attempt rediscovery","method":"markCoordinatorUnknown"} >>>>>>> >>>>>>> {"time":"2021-07-07T22:27:10,465","severity":"INFO","component":"org.apache.kafka.clients.consumer.internals.AbstractCoordinator$FindCoordinatorResponseHandler","message":"[Consumer >>>>>>> clientId=consumer-MFTDataProcessorEventSignatureConsumerGroupV1R1-3, >>>>>>> groupId=MFTDataProcessorEventSignatureConsumerGroupV1R1] Discovered >>>>>>> group >>>>>>> coordinator 100.98.40.16:9092 (id: 2147483623 rack: >>>>>>> null)","method":"onSuccess"} >>>>>>> >>>>>>> The consumers retried for more than an hour but the above logs are >>>>>>> observed again. >>>>>>> The consumers started pulling data after a manual restart. >>>>>>> >>>>>>> No WARN or ERROR logs were observed in Kafka or Zookeeper during >>>>>>> this period. >>>>>>> >>>>>>> Our observation from this incident is that Kafka Consumer retries >>>>>>> could not resolve the issue but a manual restart (or) Flink internal >>>>>>> restart(Failure rate restart policy) does. >>>>>>> >>>>>>> Has anyone faced this issue before? Any pointers are appreciated. >>>>>>> >>>>>>> Regards, >>>>>>> Rahul >>>>>>> >>>>>>
