Re: Re:Re: The wrong Options of Kafka Connector, will make the cluster can not run any job
oh, I am wrong again, the last it is in flink_1.12.2 not flink_1.11.2 -- Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/
Re:Re: The wrong Options of Kafka Connector, will make the cluster can not run any job
I had tested flink job in flink_1.11.2 and flink_1.12.2. The error log I post before is in flink_1.11.2 cluster. Now I run job in flink_1.11.2. 1. The wrong Options of Kafka Connector Ip is right, port is wrong, ``` CREATE TABLE KafkaTable ( message STRING ) WITH ( 'connector' = 'kafka', 'topic' = 'filebeat_json_install_log', 'properties.bootstrap.servers' = '192.168.0.77:9093', 'properties.group.id' = 'testGroup', 'scan.startup.mode' = 'latest-offset', 'format' = 'json' ); ``` 2. Job details In flink web UI Log in Root Exception Tabs, as below: ``` 2021-04-27 15:59:11 org.apache.kafka.common.errors.TimeoutException: Timeout of 6ms expired before the position for partition filebeat_json_install_log-3 could be determined ``` 3. Logs in Job Manager Job Manager print logs continuously as below: ``` org.apache.kafka.common.errors.TimeoutException: Timeout of 6ms expired before the position for partition filebeat_json_install_log-3 could be determined 2021-04-27 08:03:16,162 INFO org.apache.flink.runtime.executiongraph.failover.flip1.RestartPipelinedRegionFailoverStrategy [] - Calculating tasks to restart to recover the failed task cbc357ccb763df2852fee8c4fc7d55f2_0. 2021-04-27 08:03:16,163 INFO org.apache.flink.runtime.executiongraph.failover.flip1.RestartPipelinedRegionFailoverStrategy [] - 1 tasks should be restarted to recover the failed task cbc357ccb763df2852fee8c4fc7d55f2_0. 2021-04-27 08:03:16,163 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Job v2_ods_device_action_log (876dbcddcec696d42ed887512dacdf22) switched from state RUNNING to RESTARTING. 2021-04-27 08:03:17,163 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Job v2_ods_device_action_log (876dbcddcec696d42ed887512dacdf22) switched from state RESTARTING to RUNNING. 2021-04-27 08:03:17,164 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator[] - Restoring job 876dbcddcec696d42ed887512dacdf22 from Checkpoint 6 @ 1619510548493 for 876dbcddcec696d42ed887512dacdf22 located at oss://tanwan-datahub/test/flinksql/checkpoints/876dbcddcec696d42ed887512dacdf22/chk-6. 2021-04-27 08:03:17,165 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator[] - No master state to restore 2021-04-27 08:03:17,165 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Source: time, pk_id, key_id, idfv, media_site_id]) (1/1) (278dd023107c2fd3f2b42383e0c01794) switched from CREATED to SCHEDULED. 2021-04-27 08:03:17,165 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Source: ...) (1/1) (278dd023107c2fd3f2b42383e0c01794) switched from SCHEDULED to DEPLOYING. 2021-04-27 08:03:17,166 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Deploying Source: TableSourceScan(tac2fd3f2b42383e0c01794 to 192.168.3.64:6122-55a668 @ 192.168.3.64 (dataPort=34077) with allocation id 091b8c459bd00a2deaea398a41c831ab 2021-04-27 08:03:17,176 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Source: TableSourceScan(table=[[d...3e0c01794) switched from DEPLOYING to RUNNING. ``` 3. Cancel job When I cancel the job ,Job Manager print logs as below: ``` 2021-04-27 08:11:18,190 INFO org.apache.flink.runtime.dispatcher.StandaloneDispatcher [] - Job 876dbcddcec696d42ed887512dacdf22 reached globally terminal state CANCELED. 2021-04-27 08:11:18,196 INFO org.apache.flink.runtime.jobmaster.JobMaster [] - Stopping the JobMaster for job v2_ods_device_action_log(876dbcddcec696d42ed887512dacdf22). 2021-04-27 08:11:18,197 INFO org.apache.flink.runtime.jobmaster.slotpool.SlotPoolImpl [] - Suspending SlotPool. 2021-04-27 08:11:18,197 INFO org.apache.flink.runtime.jobmaster.JobMaster [] - Close ResourceManager connection 65303b0e98faaa00ada09ad7271be558: Stopping JobMaster for job v2_ods_device_action_log(876dbcddcec696d42ed887512dacdf22).. 2021-04-27 08:11:18,197 INFO org.apache.flink.runtime.jobmaster.slotpool.SlotPoolImpl [] - Stopping SlotPool. 2021-04-27 08:11:18,197 INFO org.apache.flink.runtime.resourcemanager.StandaloneResourceManager [] - Disconnect job manager 0...@akka.tcp://flink@flink-jobmanager:6123/user/rpc/jobmanager_3 for job 876dbcddcec696d42ed887512dacdf22 from the resource manager. 2021-04-27 08:11:18,216 WARN org.apache.flink.runtime.taskmanager.TaskManagerLocation [] - No hostname could be resolved for the IP address 192.168.3.64, using IP address as host name. Local input split assignment (such as for HDFS files) may be impacted. 2021-04-27 08:11:18,271 INFO org.apache.flink.fs.osshadoop.shaded.com.aliyun.oss [] - [Server]Unable to execute HTTP request: Not Found [ErrorCode]: NoSuchKey [RequestId]: 6087C726766D47343487BE32 [HostId]: null 2021-04-27 08:11:18,275 INFO
Re: The wrong Options of Kafka Connector, will make the cluster can not run any job
Thanks a lot for your message. This could be a bug in Flink. It seems that the archival of the execution graph is failing because some classes are unloaded. What I observe from your stack traces is that some classes are loaded from flink-dist_2.11-1.11.2.jar, while other classes are loaded from template-common-jar-0.0.1. Maybe Flink is closing the usercode classloader, and this is causing the exception during the archival of the execution graph. Can you make sure that the core Flink classes are only in your classpath once (in flink-dist), and the template-common-jar-0.0.1 doesn't contain the runtime Flink classes? (for example by setting the Flink dependencies to provided when using the maven-shade-plugin). For the issue while submitting the job, I can not provide you any further help, because you haven't posted the exception that occurred in the REST handler. Could you post this exception here as well? Best wishes, Robert On Sun, Apr 25, 2021 at 2:44 PM chenxuying wrote: > environment: > > flinksql 1.12.2 > > k8s session mode > > description: > > I got follow error log when my kafka connector port was wrong > > > > > 2021-04-25 16:49:50 > > org.apache.kafka.common.errors.TimeoutException: Timeout of 6ms > expired before the position for partition filebeat_json_install_log-3 could > be determined > > > > > > I got follow error log when my kafka connector ip was wrong > > > > > 2021-04-25 20:12:53 > > org.apache.flink.runtime.JobException: Recovery is suppressed by > NoRestartBackoffTimeStrategy > > at > org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:118) > > at > org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:80) > > at > org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:233) > > at > org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:224) > > at > org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:215) > > at > org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:669) > > at > org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:89) > > at > org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:447) > > at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) > > at > sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) > > at > sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) > > at java.lang.reflect.Method.invoke(Method.java:498) > > at > org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:305) > > at > org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:212) > > at > org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:77) > > at > org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:158) > > at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26) > > at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21) > > at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123) > > at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21) > > at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170) > > at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) > > at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) > > at akka.actor.Actor$class.aroundReceive(Actor.scala:517) > > 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) > > Caused by: org.apache.kafka.common.errors.TimeoutException: Timeout > expired while fetching topic metadata > > > > > > When the job was cancelled,there was follow error log: > > > > > 2021-04-25 08:53:41,115 INFO > org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Job > v2_ods_device_action_log (fcc451b8a521398b10e5b86153141fbf) switched from > state CANCELLING to CANCELED. > > 2021-04-25 08:53:41,115 INFO > org.apache.flink.runtime.checkpoint.CheckpointCoordinator[] - Stopping > checkpoint coordinator for job
The wrong Options of Kafka Connector, will make the cluster can not run any job
environment: flinksql 1.12.2 k8s session mode description: I got follow error log when my kafka connector port was wrong > 2021-04-25 16:49:50 org.apache.kafka.common.errors.TimeoutException: Timeout of 6ms expired before the position for partition filebeat_json_install_log-3 could be determined > I got follow error log when my kafka connector ip was wrong > 2021-04-25 20:12:53 org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:118) at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:80) at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:233) at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:224) at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:215) at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:669) at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:89) at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:447) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:498) at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:305) at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:212) at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:77) at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:158) at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26) at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21) at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123) at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) at akka.actor.Actor$class.aroundReceive(Actor.scala:517) 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) Caused by: org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata > When the job was cancelled,there was follow error log: > 2021-04-25 08:53:41,115 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Job v2_ods_device_action_log (fcc451b8a521398b10e5b86153141fbf) switched from state CANCELLING to CANCELED. 2021-04-25 08:53:41,115 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator[] - Stopping checkpoint coordinator for job fcc451b8a521398b10e5b86153141fbf. 2021-04-25 08:53:41,115 INFO org.apache.flink.runtime.checkpoint.StandaloneCompletedCheckpointStore [] - Shutting down 2021-04-25 08:53:41,115 INFO org.apache.flink.runtime.checkpoint.CompletedCheckpoint [] - Checkpoint with ID 1 at 'oss://tanwan-datahub/test/flinksql/checkpoints/fcc451b8a521398b10e5b86153141fbf/chk-1' not discarded. 2021-04-25 08:53:41,115 INFO org.apache.flink.runtime.checkpoint.CompletedCheckpoint [] - Checkpoint with ID 2 at 'oss://tanwan-datahub/test/flinksql/checkpoints/fcc451b8a521398b10e5b86153141fbf/chk-2' not discarded. 2021-04-25 08:53:41,116 INFO org.apache.flink.runtime.checkpoint.CompletedCheckpoint [] - Checkpoint with ID 3 at 'oss://tanwan-datahub/test/flinksql/checkpoints/fcc451b8a521398b10e5b86153141fbf/chk-3' not discarded. 2021-04-25 08:53:41,137 INFO org.apache.flink.runtime.dispatcher.StandaloneDispatcher [] - Job fcc451b8a521398b10e5b86153141fbf reached globally terminal state CANCELED. 2021-04-25 08:53:41,148 INFO org.apache.flink.runtime.jobmaster.JobMaster [] - Stopping the JobMaster for job
