Ok, thanks for the update. Great that you managed to resolve this issue :) Best, Piotrek
pon., 19 lip 2021 o 17:13 Rahul Patwari <rahulpatwari8...@gmail.com> napisał(a): > Hi Piotrek, > > I was just about to update. > You are right. The issue is because of a stalled task manager due to High > Heap Usage. And the High Heap Usage is because of a Memory Leak in a > library we are using. > > Thanks for your help. > > On Mon, Jul 19, 2021 at 8:31 PM Piotr Nowojski <pnowoj...@apache.org> > wrote: > >> 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 >>>>>>>>> >>>>>>>>
