es.davy.ai

Preguntas y respuestas de programación confiables

¿Tienes una pregunta?

Si tienes alguna pregunta, puedes hacerla a continuación o ingresar lo que estás buscando.

Aplicación Kinesis – Excepción de tiempo de espera en Flink 1.11

29 Jul, 2023 Programación

Estoy trabajando con la aplicación Kinesis usando Flink 1.11, pero obtengo el siguiente error al iniciar mi aplicación:

java.util.concurrent.TimeoutException: El heartbeat del TaskManager con id 421563c271e57acb4592f9d447d45b42 ha excedido el tiempo de espera.
    at org.apache.flink.runtime.resourcemanager.ResourceManager$TaskManagerHeartbeatListener.notifyHeartbeatTimeout(ResourceManager.java:1202)
    at org.apache.flink.runtime.heartbeat.HeartbeatMonitorImpl.run(HeartbeatMonitorImpl.java:109)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRunAsync(AkkaRpcActor.java:402)
    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:195)
    at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:74)
    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:152)
    at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26)
    at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21)
    at scala.PartialFunction.applyOrElse(PartialFunction.scala:123)
    at scala.PartialFunction.applyOrElse$(PartialFunction.scala:122)
    at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21)
    at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
    at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
    at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
    at akka.actor.Actor.aroundReceive(Actor.scala:517)
    at akka.actor.Actor.aroundReceive$(Actor.scala:515)
    at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225)
    at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592)
    at akka.actor.ActorCell.invoke(ActorCell.scala:561)
    at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258)
    at akka.dispatch.Mailbox.run(Mailbox.scala:225)
    at akka.dispatch.Mailbox.exec(Mailbox.scala:235)
    at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)

Estoy usando la configuración predeterminada en Kinesis y 4 KPUs.

Tags:  , , ,

Answer

  1. Avatar for davy.ai
    davy.ai

    El mensaje de error sugiere que uno de los TaskManagers en el clúster Flink no pudo enviar una señal de latido al ResourceManager dentro de un marco de tiempo específico, probablemente debido a un problema de red o hardware que causó que el TaskManager se detuviera o dejara de responder. Esto podría ser un problema transitorio o un problema más persistente, dependiendo de la causa raíz.

    Algunas posibles soluciones para este error son:

    • Aumentar la configuración de tiempo de espera del latido en Flink para permitir retrasos más largos en la comunicación. Sin embargo, esto solo podría ocultar el problema subyacente y llevar a problemas de rendimiento más graves en el futuro.
    • Verificar la configuración de red y hardware del TaskManager afectado y del clúster Flink en su conjunto para asegurarse de que estén configurados correctamente y funcionando como se espera. Esto podría incluir verificar la congestión de la red, la falta de recursos u otros problemas que puedan afectar la estabilidad y confiabilidad del clúster.
    • Revisar la configuración y los registros de Kinesis para determinar si hay algún problema con el flujo de datos o los pipelines de procesamiento que puedan estar causando que el TaskManager se quede rezagado o no pueda mantenerse al día con la carga de trabajo. Esto podría implicar optimizar el código de la aplicación, modificar la configuración de Kinesis o ajustar la arquitectura de implementación para adaptarse mejor a los requisitos de la carga de trabajo.

Comments are closed.