collection-Certificate-generator-jobmanager is unstable

[Nithin-ns24]

Hii team,

we have deployed KP-flink-job of collection-Certificate-generator with release-4.10.0_RC1.
From past 3 days, we noticed that collection-Certificate-generator-jobmanager is going down and collection-Certificate-generator-taskmanager is restarting multiple times because of unstable job manager

Below are the logs from collection-Certificate-generator-jobmanager
Starting standalonejob as a console application on host collection-certificate-generator-jobmanager-jclll.
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/opt/flink/lib/log4j-slf4j-impl-2.16.0.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/flink/lib/qrcode-image-generator-1.0.0.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory]
WARNING: sun.reflect.Reflection.getCallerClass is not supported. This will impact performance.
2023-07-21 06:26:06,195 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - --------------------------------------------------------------------------------
2023-07-21 06:26:06,198 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - Preconfiguration:
2023-07-21 06:26:06,198 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] -

RESOURCE_PARAMS extraction logs:
jvm_params: -Xmx939524096 -Xms939524096 -XX:MaxMetaspaceSize=268435456
dynamic_configs: -D jobmanager.memory.off-heap.size=134217728b -D jobmanager.memory.jvm-overhead.min=201326592b -D jobmanager.memory.jvm-metaspace.size=268435456b -D jobmanager.memory.heap.size=939524096b -D jobmanager.memory.jvm-overhead.max=201326592b
logs: WARNING: sun.reflect.Reflection.getCallerClass is not supported. This will impact performance.
INFO [] - Loading configuration property: jobmanager.memory.flink.size, 1024m
INFO [] - Loading configuration property: taskmanager.memory.flink.size, 1024m
INFO [] - Loading configuration property: taskmanager.numberOfTaskSlots, 1
INFO [] - Loading configuration property: parallelism.default, 1
INFO [] - Loading configuration property: jobmanager.execution.failover-strategy, region
INFO [] - Loading configuration property: taskmanager.memory.network.fraction, 0.1
INFO [] - The derived from fraction jvm overhead memory (142.222mb (149130811 bytes)) is less than its min value 192.000mb (201326592 bytes), min value will be used instead
INFO [] - Final Master Memory configuration:
INFO [] - Total Process Memory: 1.438gb (1543503872 bytes)
INFO [] - Total Flink Memory: 1024.000mb (1073741824 bytes)
INFO [] - JVM Heap: 896.000mb (939524096 bytes)
INFO [] - Off-heap: 128.000mb (134217728 bytes)
INFO [] - JVM Metaspace: 256.000mb (268435456 bytes)
INFO [] - JVM Overhead: 192.000mb (201326592 bytes)

2023-07-21 06:26:06,198 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - --------------------------------------------------------------------------------
2023-07-21 06:26:06,199 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - Starting StandaloneApplicationClusterEntryPoint (Version: 1.13.5, Scala: 2.12, Rev:0ff28a7, Date:2021-12-14T23:26:04+01:00)
2023-07-21 06:26:06,199 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - OS current user: flink
log4j:WARN No appenders could be found for logger (org.apache.hadoop.metrics2.lib.MutableMetricsFactory).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by org.apache.hadoop.security.authentication.util.KerberosUtil (file:/opt/flink/lib/activity-aggregate-updater-1.0.0.jar) to method sun.security.krb5.Config.getInstance()
WARNING: Please consider reporting this to the maintainers of org.apache.hadoop.security.authentication.util.KerberosUtil
WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations
WARNING: All illegal access operations will be denied in a future release
2023-07-21 06:26:06,494 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - Current Hadoop/Kerberos user: flink
2023-07-21 06:26:06,495 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - JVM: OpenJDK 64-Bit Server VM - Oracle Corporation - 11/11.0.13+8
2023-07-21 06:26:06,495 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - Maximum heap size: 866 MiBytes
2023-07-21 06:26:06,495 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - JAVA_HOME: /usr/local/openjdk-11
2023-07-21 06:26:06,497 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - Hadoop version: 2.7.3
2023-07-21 06:26:06,497 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - JVM Options:
2023-07-21 06:26:06,497 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - -Xmx939524096
2023-07-21 06:26:06,497 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - -Xms939524096
2023-07-21 06:26:06,497 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - -XX:MaxMetaspaceSize=268435456
2023-07-21 06:26:06,497 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - -Dlog.file=/opt/flink/log/flink--standalonejob-0-collection-certificate-generator-jobmanager-jclll.log
2023-07-21 06:26:06,497 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - -Dlog4j.configuration=file:/opt/flink/conf/log4j-console.properties
2023-07-21 06:26:06,497 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - -Dlog4j.configurationFile=file:/opt/flink/conf/log4j-console.properties
2023-07-21 06:26:06,497 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - -Dlogback.configurationFile=file:/opt/flink/conf/logback-console.xml
2023-07-21 06:26:06,497 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - Program Arguments:
2023-07-21 06:26:06,498 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - --configDir
2023-07-21 06:26:06,498 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - /opt/flink/conf
2023-07-21 06:26:06,498 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - -D
2023-07-21 06:26:06,498 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - jobmanager.memory.off-heap.size=134217728b
2023-07-21 06:26:06,498 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - -D
2023-07-21 06:26:06,499 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - jobmanager.memory.jvm-overhead.min=201326592b
2023-07-21 06:26:06,499 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - -D
2023-07-21 06:26:06,499 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - jobmanager.memory.jvm-metaspace.size=268435456b
2023-07-21 06:26:06,499 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - -D
2023-07-21 06:26:06,499 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - jobmanager.memory.heap.size=939524096b
2023-07-21 06:26:06,499 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - -D
2023-07-21 06:26:06,499 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - jobmanager.memory.jvm-overhead.max=201326592b
2023-07-21 06:26:06,499 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - --job-classname=org.sunbird.job.certgen.task.CertificateGeneratorStreamTask
2023-07-21 06:26:06,499 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - ****** (sensitive information)
2023-07-21 06:26:06,499 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - -Dweb.submit.enable=false
2023-07-21 06:26:06,499 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - -Dmetrics.reporter.prom.class=org.apache.flink.metrics.prometheus.PrometheusReporter
2023-07-21 06:26:06,500 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - -Dmetrics.reporter.prom.port=9250
2023-07-21 06:26:06,500 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - -Djobmanager.rpc.address=collection-certificate-generator-jobmanager
2023-07-21 06:26:06,500 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - -Djobmanager.rpc.port=6123
2023-07-21 06:26:06,500 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - -Dparallelism.default=1
2023-07-21 06:26:06,500 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - -Dblob.server.port=6124
2023-07-21 06:26:06,500 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - -Dqueryable-state.server.ports=6125
2023-07-21 06:26:06,500 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - --config.file.path
2023-07-21 06:26:06,500 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - /data/flink/conf/collection-certificate-generator.conf
2023-07-21 06:26:06,500 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - Classpath: /opt/flink/lib/activity-aggregate-updater-1.0.0.jar:/opt/flink/lib/asset-enrichment-1.0.0.jar:/opt/flink/lib/audit-event-generator-1.0.0.jar:/opt/flink/lib/audit-history-indexer-1.0.0.jar:/opt/flink/lib/auto-creator-v2-1.0.0.jar:/opt/flink/lib/collection-cert-pre-processor-1.0.0.jar:/opt/flink/lib/collection-certificate-generator-1.0.0.jar:/opt/flink/lib/content-auto-creator-1.0.0.jar:/opt/flink/lib/content-publish-1.0.0.jar:/opt/flink/lib/enrolment-reconciliation-1.0.0.jar:/opt/flink/lib/flink-azure-fs-hadoop-1.13.5.jar:/opt/flink/lib/flink-csv-1.13.5.jar:/opt/flink/lib/flink-json-1.13.5.jar:/opt/flink/lib/flink-metrics-prometheus_2.12-1.13.5.jar:/opt/flink/lib/flink-s3-fs-hadoop-1.13.5.jar:/opt/flink/lib/flink-shaded-hadoop-2-uber-2.8.3-10.0.jar:/opt/flink/lib/flink-shaded-zookeeper-3.4.14.jar:/opt/flink/lib/flink-streaming-scala_2.12-1.13.6.jar:/opt/flink/lib/flink-table-blink_2.12-1.13.5.jar:/opt/flink/lib/flink-table_2.12-1.13.5.jar:/opt/flink/lib/log4j-1.2-api-2.16.0.jar:/opt/flink/lib/log4j-api-2.16.0.jar:/opt/flink/lib/log4j-core-2.16.0.jar:/opt/flink/lib/log4j-slf4j-impl-2.16.0.jar:/opt/flink/lib/metrics-data-transformer-1.0.0.jar:/opt/flink/lib/post-publish-processor-1.0.0.jar:/opt/flink/lib/qrcode-image-generator-1.0.0.jar:/opt/flink/lib/questionset-publish-1.0.0.jar:/opt/flink/lib/relation-cache-updater-1.0.0.jar:/opt/flink/lib/scala-library-2.12.11.jar:/opt/flink/lib/search-indexer-1.0.0.jar:/opt/flink/lib/video-stream-generator-1.0.0.jar:/opt/flink/lib/flink-dist_2.12-1.13.5.jar:::
2023-07-21 06:26:06,500 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - --------------------------------------------------------------------------------
2023-07-21 06:26:06,501 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - Registered UNIX signal handlers for [TERM, HUP, INT]
2023-07-21 06:26:06,517 INFO org.apache.flink.configuration.GlobalConfiguration [] - Loading configuration property: jobmanager.memory.flink.size, 1024m
2023-07-21 06:26:06,517 INFO org.apache.flink.configuration.GlobalConfiguration [] - Loading configuration property: taskmanager.memory.flink.size, 1024m
2023-07-21 06:26:06,518 INFO org.apache.flink.configuration.GlobalConfiguration [] - Loading configuration property: taskmanager.numberOfTaskSlots, 1
2023-07-21 06:26:06,518 INFO org.apache.flink.configuration.GlobalConfiguration [] - Loading configuration property: parallelism.default, 1
2023-07-21 06:26:06,518 INFO org.apache.flink.configuration.GlobalConfiguration [] - Loading configuration property: jobmanager.execution.failover-strategy, region
2023-07-21 06:26:06,518 INFO org.apache.flink.configuration.GlobalConfiguration [] - Loading configuration property: taskmanager.memory.network.fraction, 0.1
2023-07-21 06:26:06,570 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - Starting StandaloneApplicationClusterEntryPoint.
2023-07-21 06:26:06,589 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - Install default filesystem.
2023-07-21 06:26:06,636 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - Install security context.
2023-07-21 06:26:06,716 WARN org.apache.flink.runtime.util.HadoopUtils [] - Could not find Hadoop configuration via any of the supported methods (Flink configuration, environment variables).
2023-07-21 06:26:06,767 INFO org.apache.flink.runtime.security.modules.HadoopModule [] - Hadoop user set to flink (auth:SIMPLE)
2023-07-21 06:26:06,770 INFO org.apache.flink.runtime.security.modules.JaasModule [] - Jaas file will be created as /tmp/jaas-13042896602265024060.conf.
2023-07-21 06:26:06,775 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - Initializing cluster services.
2023-07-21 06:26:06,791 INFO org.apache.flink.runtime.rpc.akka.AkkaRpcServiceUtils [] - Trying to start actor system, external address collection-certificate-generator-jobmanager:6123, bind address 0.0.0.0:6123.
2023-07-21 06:26:07,616 INFO org.apache.flink.runtime.rpc.akka.AkkaRpcServiceUtils [] - Actor system started at akka.tcp://flink@collection-certificate-generator-jobmanager:6123
2023-07-21 06:26:07,634 INFO org.apache.flink.configuration.Configuration [] - Config uses fallback configuration key 'jobmanager.rpc.address' instead of key 'rest.address'
2023-07-21 06:26:07,639 INFO org.apache.flink.runtime.blob.BlobServer [] - Created BLOB server storage directory /tmp/blobStore-170371ca-8040-4333-bead-20ba63a735ca
2023-07-21 06:26:07,643 INFO org.apache.flink.runtime.blob.BlobServer [] - Started BLOB server at 0.0.0.0:6124 - max concurrent requests: 50 - max backlog: 1000
2023-07-21 06:26:07,660 WARN org.apache.flink.runtime.metrics.ReporterSetup [] - Multiple implementations of the same reporter were found in 'lib' and/or 'plugins' directories for org.apache.flink.metrics.prometheus.PrometheusReporterFactory. It is recommended to remove redundant reporter JARs to resolve used versions' ambiguity.
2023-07-21 06:26:07,660 WARN org.apache.flink.runtime.metrics.ReporterSetup [] - Multiple implementations of the same reporter were found in 'lib' and/or 'plugins' directories for org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterFactory. It is recommended to remove redundant reporter JARs to resolve used versions' ambiguity.
2023-07-21 06:26:07,686 INFO org.apache.flink.metrics.prometheus.PrometheusReporter [] - Started PrometheusReporter HTTP server on port 9250.
2023-07-21 06:26:07,687 INFO org.apache.flink.runtime.metrics.MetricRegistryImpl [] - Reporting metrics for reporter prom of type org.apache.flink.metrics.prometheus.PrometheusReporter.
2023-07-21 06:26:07,690 INFO org.apache.flink.runtime.rpc.akka.AkkaRpcServiceUtils [] - Trying to start actor system, external address collection-certificate-generator-jobmanager:0, bind address 0.0.0.0:0.
2023-07-21 06:26:07,729 INFO org.apache.flink.runtime.rpc.akka.AkkaRpcServiceUtils [] - Actor system started at akka.tcp://flink-metrics@collection-certificate-generator-jobmanager:40961
2023-07-21 06:26:07,740 INFO org.apache.flink.runtime.rpc.akka.AkkaRpcService [] - Starting RPC endpoint for org.apache.flink.runtime.metrics.dump.MetricQueryService at akka://flink-metrics/user/rpc/MetricQueryService .
2023-07-21 06:26:07,778 INFO org.apache.flink.configuration.Configuration [] - Config uses fallback configuration key 'jobmanager.rpc.address' instead of key 'rest.address'
2023-07-21 06:26:07,779 INFO org.apache.flink.runtime.jobmaster.MiniDispatcherRestEndpoint [] - Upload directory /tmp/flink-web-8ff48e74-5941-4cb2-bce5-3835dd3f79b8/flink-web-upload does not exist.
2023-07-21 06:26:07,780 INFO org.apache.flink.runtime.jobmaster.MiniDispatcherRestEndpoint [] - Created directory /tmp/flink-web-8ff48e74-5941-4cb2-bce5-3835dd3f79b8/flink-web-upload for file uploads.
2023-07-21 06:26:07,791 INFO org.apache.flink.runtime.jobmaster.MiniDispatcherRestEndpoint [] - Starting rest endpoint.
2023-07-21 06:26:07,998 INFO org.apache.flink.runtime.webmonitor.WebMonitorUtils [] - Determined location of main cluster component log file: /opt/flink/log/flink--standalonejob-0-collection-certificate-generator-jobmanager-jclll.log
2023-07-21 06:26:07,998 INFO org.apache.flink.runtime.webmonitor.WebMonitorUtils [] - Determined location of main cluster component stdout file: /opt/flink/log/flink--standalonejob-0-collection-certificate-generator-jobmanager-jclll.out
2023-07-21 06:26:08,133 INFO org.apache.flink.runtime.jobmaster.MiniDispatcherRestEndpoint [] - Rest endpoint listening at collection-certificate-generator-jobmanager:8081
2023-07-21 06:26:08,134 INFO org.apache.flink.runtime.jobmaster.MiniDispatcherRestEndpoint [] - http://collection-certificate-generator-jobmanager:8081 was granted leadership with leaderSessionID=00000000-0000-0000-0000-000000000000
2023-07-21 06:26:08,136 INFO org.apache.flink.runtime.jobmaster.MiniDispatcherRestEndpoint [] - Web frontend listening at http://collection-certificate-generator-jobmanager:8081.
2023-07-21 06:26:08,158 INFO org.apache.flink.runtime.rpc.akka.AkkaRpcService [] - Starting RPC endpoint for org.apache.flink.runtime.resourcemanager.StandaloneResourceManager at akka://flink/user/rpc/resourcemanager_0 .
2023-07-21 06:26:08,173 INFO org.apache.flink.runtime.dispatcher.runner.DefaultDispatcherRunner [] - DefaultDispatcherRunner was granted leadership with leader id 00000000-0000-0000-0000-000000000000. Creating new DispatcherLeaderProcess.
2023-07-21 06:26:08,179 INFO org.apache.flink.runtime.dispatcher.runner.SessionDispatcherLeaderProcess [] - Start SessionDispatcherLeaderProcess.
2023-07-21 06:26:08,181 INFO org.apache.flink.runtime.resourcemanager.StandaloneResourceManager [] - Starting the resource manager.
2023-07-21 06:26:08,184 INFO org.apache.flink.runtime.resourcemanager.StandaloneResourceManager [] - ResourceManager akka.tcp://flink@collection-certificate-generator-jobmanager:6123/user/rpc/resourcemanager_0 was granted leadership with fencing token 00000000000000000000000000000000
2023-07-21 06:26:08,180 INFO org.apache.flink.runtime.dispatcher.runner.SessionDispatcherLeaderProcess [] - Recover all persisted job graphs.
2023-07-21 06:26:08,185 INFO org.apache.flink.runtime.dispatcher.runner.SessionDispatcherLeaderProcess [] - Successfully recovered 0 persisted job graphs.
2023-07-21 06:26:08,214 INFO org.apache.flink.runtime.rpc.akka.AkkaRpcService [] - Starting RPC endpoint for org.apache.flink.runtime.dispatcher.StandaloneDispatcher at akka://flink/user/rpc/dispatcher_1 .
2023-07-21 06:26:08,225 INFO org.apache.flink.client.ClientUtils [] - Starting program (detached: true)
2023-07-21 06:26:08,269 INFO org.apache.flink.api.java.typeutils.TypeExtractor [] - class org.sunbird.job.certgen.functions.NotificationMetaData does not contain a setter for field userId
2023-07-21 06:26:08,269 INFO org.apache.flink.api.java.typeutils.TypeExtractor [] - Class class org.sunbird.job.certgen.functions.NotificationMetaData cannot be used as a POJO type because not all fields are valid POJO fields, and must be processed as GenericType. Please read the Flink documentation on "Data Types & Serialization" for details of the effect on performance.
2023-07-21 06:26:08,270 INFO org.apache.flink.api.java.typeutils.TypeExtractor [] - class org.sunbird.job.certgen.functions.UserFeedMetaData does not contain a setter for field userId
2023-07-21 06:26:08,270 INFO org.apache.flink.api.java.typeutils.TypeExtractor [] - Class class org.sunbird.job.certgen.functions.UserFeedMetaData cannot be used as a POJO type because not all fields are valid POJO fields, and must be processed as GenericType. Please read the Flink documentation on "Data Types & Serialization" for details of the effect on performance.
2023-07-21 06:26:08,299 INFO org.apache.flink.api.java.typeutils.TypeExtractor [] - class org.sunbird.job.certgen.domain.Event does not contain a getter for field jobName
2023-07-21 06:26:08,299 INFO org.apache.flink.api.java.typeutils.TypeExtractor [] - class org.sunbird.job.certgen.domain.Event does not contain a setter for field jobName
2023-07-21 06:26:08,299 INFO org.apache.flink.api.java.typeutils.TypeExtractor [] - Class class org.sunbird.job.certgen.domain.Event cannot be used as a POJO type because not all fields are valid POJO fields, and must be processed as GenericType. Please read the Flink documentation on "Data Types & Serialization" for details of the effect on performance.
2023-07-21 06:26:08,300 INFO org.apache.flink.api.java.typeutils.TypeExtractor [] - class org.sunbird.job.certgen.functions.NotificationMetaData does not contain a setter for field userId
2023-07-21 06:26:08,300 INFO org.apache.flink.api.java.typeutils.TypeExtractor [] - Class class org.sunbird.job.certgen.functions.NotificationMetaData cannot be used as a POJO type because not all fields are valid POJO fields, and must be processed as GenericType. Please read the Flink documentation on "Data Types & Serialization" for details of the effect on performance.
2023-07-21 06:26:08,300 INFO org.apache.flink.api.java.typeutils.TypeExtractor [] - class org.sunbird.job.certgen.functions.UserFeedMetaData does not contain a setter for field userId
2023-07-21 06:26:08,300 INFO org.apache.flink.api.java.typeutils.TypeExtractor [] - Class class org.sunbird.job.certgen.functions.UserFeedMetaData cannot be used as a POJO type because not all fields are valid POJO fields, and must be processed as GenericType. Please read the Flink documentation on "Data Types & Serialization" for details of the effect on performance.
2023-07-21 06:26:08,475 WARN org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer [] - Property [transaction.timeout.ms] not specified. Setting it to 3600000 ms
2023-07-21 06:26:08,635 INFO org.apache.flink.client.deployment.application.executors.EmbeddedExecutor [] - Job 02f628c175e6e7eb123817aab9934c08 is submitted.
2023-07-21 06:26:08,635 INFO org.apache.flink.client.deployment.application.executors.EmbeddedExecutor [] - Submitting Job with JobId=02f628c175e6e7eb123817aab9934c08.
2023-07-21 06:26:08,641 INFO org.apache.flink.runtime.dispatcher.StandaloneDispatcher [] - Received JobGraph submission 02f628c175e6e7eb123817aab9934c08 (collection-certificate-generator).
2023-07-21 06:26:08,641 INFO org.apache.flink.runtime.dispatcher.StandaloneDispatcher [] - Submitting job 02f628c175e6e7eb123817aab9934c08 (collection-certificate-generator).
2023-07-21 06:26:08,664 INFO org.apache.flink.runtime.rpc.akka.AkkaRpcService [] - Starting RPC endpoint for org.apache.flink.runtime.jobmaster.JobMaster at akka://flink/user/rpc/jobmanager_2 .
2023-07-21 06:26:08,671 INFO org.apache.flink.runtime.jobmaster.JobMaster [] - Initializing job collection-certificate-generator (02f628c175e6e7eb123817aab9934c08).
2023-07-21 06:26:08,689 INFO org.apache.flink.runtime.jobmaster.JobMaster [] - Using restart back off time strategy FixedDelayRestartBackoffTimeStrategy(maxNumberRestartAttempts=3, backoffTimeMS=240000) for collection-certificate-generator (02f628c175e6e7eb123817aab9934c08).
2023-07-21 06:26:08,719 INFO org.apache.flink.runtime.jobmaster.JobMaster [] - Running initialization on master for job collection-certificate-generator (02f628c175e6e7eb123817aab9934c08).
2023-07-21 06:26:08,719 INFO org.apache.flink.runtime.jobmaster.JobMaster [] - Successfully ran initialization on master in 0 ms.
2023-07-21 06:26:08,733 INFO org.apache.flink.runtime.scheduler.adapter.DefaultExecutionTopology [] - Built 1 pipelined regions in 1 ms
2023-07-21 06:26:08,740 INFO org.apache.flink.runtime.jobmaster.JobMaster [] - Using job/cluster config to configure application-defined state backend: File State Backend (checkpoints: 'wasbs://flink-check-points-store@aksdevstorag.blob.core.windows.net/checkpoint/collection-certificate-generator', savepoints: 'null, fileStateThreshold: -1)
2023-07-21 06:26:08,741 INFO org.apache.flink.runtime.jobmaster.JobMaster [] - Using application-defined state backend: File State Backend (checkpoints: 'wasbs://flink-check-points-store@aksdevstorag.blob.core.windows.net/checkpoint/collection-certificate-generator', savepoints: 'null, fileStateThreshold: 20480)
2023-07-21 06:26:08,742 INFO org.apache.flink.runtime.jobmaster.JobMaster [] - Using legacy state backend File State Backend (checkpoints: 'wasbs://flink-check-points-store@aksdevstorag.blob.core.windows.net/checkpoint/collection-certificate-generator', savepoints: 'null, fileStateThreshold: 20480) as Job checkpoint storage
2023-07-21 06:26:08,750 INFO org.apache.flink.fs.azurefs.AzureFSFactory [] - Trying to load and instantiate Azure File System
2023-07-21 06:26:08,843 INFO org.apache.flink.fs.azure.shaded.org.eclipse.jetty.util.log [] - Logging initialized @3568ms
2023-07-21 06:26:08,947 WARN org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.metrics2.impl.MetricsConfig [] - Cannot locate configuration: tried hadoop-metrics2-azure-file-system.properties,hadoop-metrics2.properties
2023-07-21 06:26:08,993 INFO org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.metrics2.impl.MetricsSystemImpl [] - Scheduled Metric snapshot period at 10 second(s).
2023-07-21 06:26:08,994 INFO org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.metrics2.impl.MetricsSystemImpl [] - azure-file-system metrics system started
2023-07-21 06:26:10,022 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator [] - No checkpoint found during restore.
2023-07-21 06:26:10,027 INFO org.apache.flink.runtime.jobmaster.JobMaster [] - Using failover strategy org.apache.flink.runtime.executiongraph.failover.flip1.RestartPipelinedRegionFailoverStrategy@44e9b36a for collection-certificate-generator (02f628c175e6e7eb123817aab9934c08).
2023-07-21 06:26:10,038 INFO org.apache.flink.runtime.jobmaster.JobMaster [] - Starting execution of job collection-certificate-generator (02f628c175e6e7eb123817aab9934c08) under job master id 00000000000000000000000000000000.
2023-07-21 06:26:10,040 INFO org.apache.flink.runtime.jobmaster.JobMaster [] - Starting scheduling with scheduling strategy [org.apache.flink.runtime.scheduler.strategy.PipelinedRegionSchedulingStrategy]
2023-07-21 06:26:10,040 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Job collection-certificate-generator (02f628c175e6e7eb123817aab9934c08) switched from state CREATED to RUNNING.
2023-07-21 06:26:10,046 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Source: certificate (1/1) (296d5c61a0b965dd6677c2e0e44703a4) switched from CREATED to SCHEDULED.
2023-07-21 06:26:10,046 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - collection-certificate-generator -> (Sink: collection-certificate-generator-audit-events-sink, notifier, user-feed) (1/1) (eaeaa2f1e89561942b8b88a31e7b0546) switched from CREATED to SCHEDULED.
2023-07-21 06:26:10,059 INFO org.apache.flink.runtime.jobmaster.JobMaster [] - Connecting to ResourceManager akka.tcp://flink@collection-certificate-generator-jobmanager:6123/user/rpc/resourcemanager_*(00000000000000000000000000000000)
2023-07-21 06:26:10,064 INFO org.apache.flink.runtime.jobmaster.JobMaster [] - Resolved ResourceManager address, beginning registration
2023-07-21 06:26:10,065 INFO org.apache.flink.runtime.resourcemanager.StandaloneResourceManager [] - Registering job manager 00000000000000000000000000000000@akka.tcp://flink@collection-certificate-generator-jobmanager:6123/user/rpc/jobmanager_2 for job 02f628c175e6e7eb123817aab9934c08.
2023-07-21 06:26:10,071 INFO org.apache.flink.runtime.resourcemanager.StandaloneResourceManager [] - Registered job manager 00000000000000000000000000000000@akka.tcp://flink@collection-certificate-generator-jobmanager:6123/user/rpc/jobmanager_2 for job 02f628c175e6e7eb123817aab9934c08.
2023-07-21 06:26:10,073 INFO org.apache.flink.runtime.jobmaster.JobMaster [] - JobManager successfully registered at ResourceManager, leader id: 00000000000000000000000000000000.
2023-07-21 06:26:10,074 INFO org.apache.flink.runtime.resourcemanager.slotmanager.DeclarativeSlotManager [] - Received resource requirements from job 02f628c175e6e7eb123817aab9934c08: [ResourceRequirement{resourceProfile=ResourceProfile{UNKNOWN}, numberOfRequiredSlots=1}]
2023-07-21 06:26:16,729 INFO org.apache.flink.runtime.resourcemanager.StandaloneResourceManager [] - Registering TaskManager with ResourceID 10.51.0.253:6122-3bce41 (akka.tcp://flink@10.51.0.253:6122/user/rpc/taskmanager_0) at ResourceManager
2023-07-21 06:26:16,821 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Source: certificate (1/1) (296d5c61a0b965dd6677c2e0e44703a4) switched from SCHEDULED to DEPLOYING.
2023-07-21 06:26:16,821 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Deploying Source: certificate (1/1) (attempt #0) with attempt id 296d5c61a0b965dd6677c2e0e44703a4 to 10.51.0.253:6122-3bce41 @ 10-51-0-253.collection-certificate-generator-taskmanager-prometheus.flink-kp-dev.svc.cluster.local (dataPort=33061) with allocation id fcd9fdfd2eda7ec49738e851b4c5f28f
2023-07-21 06:26:16,828 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - collection-certificate-generator -> (Sink: collection-certificate-generator-audit-events-sink, notifier, user-feed) (1/1) (eaeaa2f1e89561942b8b88a31e7b0546) switched from SCHEDULED to DEPLOYING.
2023-07-21 06:26:16,828 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Deploying collection-certificate-generator -> (Sink: collection-certificate-generator-audit-events-sink, notifier, user-feed) (1/1) (attempt #0) with attempt id eaeaa2f1e89561942b8b88a31e7b0546 to 10.51.0.253:6122-3bce41 @ 10-51-0-253.collection-certificate-generator-taskmanager-prometheus.flink-kp-dev.svc.cluster.local (dataPort=33061) with allocation id fcd9fdfd2eda7ec49738e851b4c5f28f
2023-07-21 06:26:17,278 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Source: certificate (1/1) (296d5c61a0b965dd6677c2e0e44703a4) switched from DEPLOYING to INITIALIZING.
2023-07-21 06:26:17,292 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - collection-certificate-generator -> (Sink: collection-certificate-generator-audit-events-sink, notifier, user-feed) (1/1) (eaeaa2f1e89561942b8b88a31e7b0546) switched from DEPLOYING to INITIALIZING.
2023-07-21 06:26:17,803 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Source: certificate (1/1) (296d5c61a0b965dd6677c2e0e44703a4) switched from INITIALIZING to RUNNING.
2023-07-21 06:26:18,767 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - collection-certificate-generator -> (Sink: collection-certificate-generator-audit-events-sink, notifier, user-feed) (1/1) (eaeaa2f1e89561942b8b88a31e7b0546) switched from INITIALIZING to RUNNING.
2023-07-21 06:26:19,020 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - collection-certificate-generator -> (Sink: collection-certificate-generator-audit-events-sink, notifier, user-feed) (1/1) (eaeaa2f1e89561942b8b88a31e7b0546) switched from RUNNING to FAILED on 10.51.0.253:6122-3bce41 @ 10-51-0-253.collection-certificate-generator-taskmanager-prometheus.flink-kp-dev.svc.cluster.local (dataPort=33061).
org.sunbird.job.exception.InvalidEventException: head of empty list
at org.sunbird.job.certgen.functions.CertificateGeneratorFunction.processElement(CertificateGeneratorFunction.scala:83) ~[collection-certificate-generator-1.0.0.jar:?]
at org.sunbird.job.certgen.functions.CertificateGeneratorFunction.processElement(CertificateGeneratorFunction.scala:33) ~[collection-certificate-generator-1.0.0.jar:?]
at org.sunbird.job.BaseProcessKeyedFunction.processElement(BaseProcessFunction.scala:113) ~[activity-aggregate-updater-1.0.0.jar:?]
at org.apache.flink.streaming.api.operators.KeyedProcessOperator.processElement(KeyedProcessOperator.java:83) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:205) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:134) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:105) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:66) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:423) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:204) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:684) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at org.apache.flink.streaming.runtime.tasks.StreamTask.executeInvoke(StreamTask.java:639) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:650) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:623) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:779) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at java.lang.Thread.run(Unknown Source) ~[?:?]
2023-07-21 06:26:19,037 INFO org.apache.flink.runtime.executiongraph.failover.flip1.RestartPipelinedRegionFailoverStrategy [] - Calculating tasks to restart to recover the failed task d020ec05d4319b79a5df438d01634aac_0.
2023-07-21 06:26:19,038 INFO org.apache.flink.runtime.executiongraph.failover.flip1.RestartPipelinedRegionFailoverStrategy [] - 2 tasks should be restarted to recover the failed task d020ec05d4319b79a5df438d01634aac_0.
2023-07-21 06:26:19,043 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Job collection-certificate-generator (02f628c175e6e7eb123817aab9934c08) switched from state RUNNING to RESTARTING.
2023-07-21 06:26:19,044 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Source: certificate (1/1) (296d5c61a0b965dd6677c2e0e44703a4) switched from RUNNING to CANCELING.
2023-07-21 06:26:19,068 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Source: certificate (1/1) (296d5c61a0b965dd6677c2e0e44703a4) switched from CANCELING to CANCELED.
2023-07-21 06:26:19,072 INFO org.apache.flink.runtime.resourcemanager.slotmanager.DeclarativeSlotManager [] - Clearing resource requirements of job 02f628c175e6e7eb123817aab9934c08
2023-07-21 06:27:50,079 INFO org.apache.flink.runtime.jobmaster.slotpool.DefaultDeclarativeSlotPool [] - Releasing slot [fcd9fdfd2eda7ec49738e851b4c5f28f].

Recently we have deployed RC jobs also, can someone please let us know if this issue is related to the RC jobs setup. If not, kindly help us resolve the issue. Thank you
CC @gandham-santhosh

[amitpriyadarshi]

@Nithin-ns24 as per the log analysis it seems in the kafka topic one of the current offset event is invalid.
2023-07-21 06:26:19,020 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - collection-certificate-generator -> (Sink: collection-certificate-generator-audit-events-sink, notifier, user-feed) (1/1) (eaeaa2f1e89561942b8b88a31e7b0546) switched from RUNNING to FAILED on 10.51.0.253:6122-3bce41 @ 10-51-0-253.collection-certificate-generator-taskmanager-prometheus.flink-kp-dev.svc.cluster.local (dataPort=33061). org.sunbird.job.exception.InvalidEventException: head of empty list at org.sunbird.job.certgen.functions.CertificateGeneratorFunction.processElement(CertificateGeneratorFunction.scala:83) ~[collection-certificate-generator-1.0.0.jar:?]

Please check the task manager log as well and also get the current processing event and discard it.
cc: @vinukumar-vs @Krishnaj20 @rajeevsathish @gandham-santhosh