Certificates are not generating due to collection-certificate-generator flink job not stable. #65

bhargav0702 · 2023-01-23T13:16:10Z

bhargav0702
Jan 23, 2023

Hi Team,Recently we observed our collection-certificate-generator flink job is continuously failing with below error.

As per the below logs it is showing InvalidEventException, we are unable to identify the root cause of the issue.After clearing the lag and redeployment the jobs is getting started successful. Now currently the job is failing with same error and the lag is 32, we have captured all 32 events from the input kafka topic of this job. Below are the 32 events.

The collection-certificate-generator job is deployed with release-4.4.0_RC3 tag.

error logs

org.sunbird.job.exception.InvalidEventException: Could not forward element to next operator at org.sunbird.job.certgen.functions.CertificateGeneratorFunction.processElement(CertificateGeneratorFunction.scala:74) ~[collection-certificate-generator-1.0.0.jar:?] at org.sunbird.job.certgen.functions.CertificateGeneratorFunction.processElement(CertificateGeneratorFunction.scala:31) ~[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:85) ~[flink-dist_2.12-1.12.0.jar:1.12.0] at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:193) ~[flink-dist_2.12-1.12.0.jar:1.12.0] at org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.processElement(StreamTaskNetworkInput.java:179) ~[flink-dist_2.12-1.12.0.jar:1.12.0] at org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.emitNext(StreamTaskNetworkInput.java:152) ~[flink-dist_2.12-1.12.0.jar:1.12.0] at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:67) ~[flink-dist_2.12-1.12.0.jar:1.12.0] at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:372) ~[flink-dist_2.12-1.12.0.jar:1.12.0] at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:186) ~[flink-dist_2.12-1.12.0.jar:1.12.0] at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:575) ~[flink-dist_2.12-1.12.0.jar:1.12.0] at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:539) ~[flink-dist_2.12-1.12.0.jar:1.12.0] at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:722) ~[flink-dist_2.12-1.12.0.jar:1.12.0] at org.apache.flink.runtime.taskmanager.Task.run(Task.java:547) ~[flink-dist_2.12-1.12.0.jar:1.12.0] at java.lang.Thread.run(Unknown Source) ~[?:?] 2023-01-23 06:20:31,820 INFO org.apache.flink.runtime.executiongraph.failover.flip1.RestartPipelinedRegionFailoverStrategy [] - Calculating tasks to restart to recover the failed task d020ec05d4319b79a5df438d01634aac_0. 2023-01-23 06:20:31,821 INFO org.apache.flink.runtime.executiongraph.failover.flip1.RestartPipelinedRegionFailoverStrategy [] - 2 tasks should be restarted to recover the failed task d020ec05d4319b79a5df438d01634aac_0. 2023-01-23 06:20:31,822 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Job collection-certificate-generator (2c0fd4dc9703cd89b65f4be69fba358e) switched from state RUNNING to RESTARTING. 2023-01-23 06:20:31,823 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Source: certificate (1/1) (2da22ae1afb29529983c27334e64bfdd) switched from RUNNING to CANCELING. 2023-01-23 06:20:31,850 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Source: certificate (1/1) (2da22ae1afb29529983c27334e64bfdd) switched from CANCELING to CANCELED. 2023-01-23 06:21:01,831 INFO org.apache.flink.runtime.resourcemanager.StandaloneResourceManager [] - Registering TaskManager with ResourceID 10.10.1.229:6122-3aec54 (akka.tcp://[email protected]:6122/user/rpc/taskmanager_0) at ResourceManager 2023-01-23 06:22:04,662 INFO org.apache.flink.runtime.jobmaster.slotpool.SlotPoolImpl [] - Releasing idle slot [866e91451eb659ee41cf6fe5f29bdb44]

Kafka topic events text

gandham-santhosh · 2023-01-23T13:28:05Z

gandham-santhosh
Jan 23, 2023

@reshmi-nair can you please check this?

8 replies

coolbung Jan 24, 2023

Some additional context.

I see the below error in the collection-certificate-generator-taskmanager logs

2023-01-24 12:30:06,096 ERROR org.sunbird.job.certgen.functions.CreateUserFeedFunction     [] - Error response from createUserFeed API for request :: {"request":{"userId":"b81e00e0-5ec0-4c00-bd12-6871 │
│ 589439e6","category":"Notification","priority":1,"data":{"type":1,"actionData":{"actionType":"certificateUpdate","title":"Introduction to NUDM Platforms, Program Governance and Implementation","descri │
│ ption":"You have earned a certificate! Download it from your profile page.","identifier":"do_1136598919702609921643"}}}} :: response is :: 500 ::  {"id":".private.user.feed.v1.create","ver":"private", │
│ "ts":"2023-01-24 12:30:06:090+0000","params":{"resmsgid":null,"msgid":"112ebfcf-b1a0-498d-b48e-52e235008417","err":null,"status":"SERVER_ERROR","errmsg":null},"responseCode":"SERVER_ERROR","result":{} │
│ }

And see the below logs in the learner service

 {"timestamp":"2023-01-24T12:38:46.406Z","msg":"{\"eid\":\"LOG\",\"ets\":1674563926406,\"ver\":\"3.0\",\"mid\":\"0192e116b7b7f457df4a00f6fdddfce4\",\"actor\":{\"id\":\"e4b95f99-fb2c-490f-af91-58392e1b1 │
│ 9a4\",\"type\":\"User\"},\"context\":{\"channel\":\"0130152125009674241\",\"pdata\":{\"id\":\"dev.nulp.portal\",\"pid\":\"learner-service\",\"ver\":\"4.5.0\"},\"env\":\"User\",\"did\":\"ff9702948bc4af │
│ 01edc37ab9c59fa2d8\",\"cdata\":[{\"id\":\"0192e116b7b7f457df4a00f6fdddfce4\",\"type\":\"Request\"}],\"rollup\":{}},\"edata\":{\"level\":\"info\",\"type\":\"Api_access\",\"message\":\"Response from : G │
│ ET call \\n<!DOCTYPE html>\\n<html lang=\\\"en\\\">\\n    <head>\\n        <title>Error</title>\\n        \\n<style  type=\\\"text/css\\\">\\n            html, body, pre {\\n                margin: 0; │
│ \\n                padding: 0;\\n                font-family: Monaco, 'Lucida Console', monospace;\\n                background: #ECECEC;\\n            }\\n            h1 {\\n                margin: 0 │
│ ;\\n                background: #A31012;\\n                padding: 20px 45px;\\n                color: #fff;\\n                text-shadow: 1px 1px 1px rgba(0,0,0,.3);\\n                border-bottom │
│ : 1px solid #690000;\\n                font-size: 28px;\\n            }\\n            p#detail {\\n                margin: 0;\\n                padding: 15px 45px;\\n                background: #F5A0A │
│ 0;\\n                border-top: 4px solid #D36D6D;\\n                color: #730000;\\n                text-shadow: 1px 1px 1px rgba(255,255,255,.3);\\n                font-size: 14px;\\n             │
│     border-bottom: 1px solid #BA7A7A;\\n            }\\n        </style>\\n    </head>\\n    <body>\\n        <h1>Oops, an error occurred</h1>\\n\\n        <p id=\\\"detail\\\">\\n            This exc │
│ eption has been logged with id <strong>80ck62nfp</strong>.\\n        </p>\\n\\n    </body>\\n</html> status 500 - Internal Server Error\",\"params\":[]}}","lname":"TelemetryEventLogger","tname":"appli │
│ cation-akka.actor.most-used-two-dispatcher-7212","level":"INFO","HOSTNAME":"learner-64fb8c4d6d-fqstd","application.home":"/home/sunbird/learner/learning-service-1.0-SNAPSHOT"}                          │
│ {"timestamp":"2023-01-24T12:38:46.407Z","msg":"{\"eid\":\"ERROR\",\"ets\":1674563926407,\"ver\":\"3.0\",\"mid\":\"0192e116b7b7f457df4a00f6fdddfce4\",\"actor\":{\"id\":\"e4b95f99-fb2c-490f-af91-58392e1 │
│ b19a4\",\"type\":\"User\"},\"context\":{\"channel\":\"0130152125009674241\",\"pdata\":{\"id\":\"dev.nulp.portal\",\"pid\":\"learner-service\",\"ver\":\"4.5.0\"},\"env\":\"User\",\"did\":\"ff9702948bc4 │
│ af01edc37ab9c59fa2d8\",\"cdata\":[{\"id\":\"0192e116b7b7f457df4a00f6fdddfce4\",\"type\":\"Request\"}],\"rollup\":{}},\"edata\":{\"err\":\"INTERNAL_ERROR\",\"stacktrace\":\"FeedServiceImpl:readV1Notifi │
│ cation Exception occurred while mapping. com.fasterxml.jackson.databind.exc.MismatchedInputException.from(MismatchedInputException.java:59)com.fasterxml.jackson.databind.ObjectMapper._initForReading(O │
│ bjectMapper.java:4344)com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:4189)com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3205)com.fasterxml.jac │
│ kson.databind.ObjectMapper.readValue(ObjectMapper.java:3173)org.sunbird.client.NotificationServiceClient.readV1Notification(NotificationServiceClient.java:105)org.sunbird.service.feed.impl.FeedService │
│ Impl.getFeedsByProperties(FeedServiceImpl.java:49)org.sunbird.actor.feed.UserFeedActor.getUserFeed(UserFeedActor.java:52)org.sunbird.actor.feed.UserFeedActor.onReceive(UserFeedActor.java:33)org.sunbir │
│ d.actor.core.BaseActor.onReceive(BaseActor.java:20)akka.actor.UntypedAbstractActor$$anonfun$receive$1.applyOrElse(AbstractActor.scala:291)akka.actor.Actor$class.aroundReceive(Actor.scala:517)akka.acto │
│ r.AbstractActor.aroundReceive(AbstractActor.scala:180)akka.actor.ActorCell.receiveMessage(ActorCell.scala:588)akka.actor.ActorCell.invoke(ActorCell.scala:557)akka.dispatch.Mailbox.processMailbox(Mailb │
│ ox.scala:258)akka.dispatch.Mailbox.run(Mailbox.scala:225)akka.dispatch.Mailbox.exec(Mailbox.scala:235)akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)akka.dispatch.forkjoin.ForkJoinPo │
│ ol$WorkQueue.runTask(ForkJoinPool.java:1339)akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)\",\"errt │
│ ype\":\"api_access\",\"requestid\":\"0192e116b7b7f457df4a00f6fdddfce4\"}}","lname":"TelemetryEventLogger","tname":"application-akka.actor.most-used-two-dispatcher-7212","level":"INFO","HOSTNAME":"lear │
│ ner-64fb8c4d6d-fqstd","application.home":"/home/sunbird/learner/learning-service-1.0-SNAPSHOT"}                                                                                                          │
│ {"timestamp":"2023-01-24T12:38:46.407Z","msg":"Telemetry Object Creation Error for event : ERROR  missing required fields :err","lname":"org.sunbird.telemetry.validator.TelemetryObjectValidatorV3","tn │
│ ame":"application-akka.actor.most-used-two-dispatcher-7212","level":"INFO","HOSTNAME":"learner-64fb8c4d6d-fqstd","application.home":"/home/sunbird/learner/learning-service-1.0-SNAPSHOT"}               │
│ {"timestamp":"2023-01-24T12:38:46.407Z","msg":"Telemetry Object Creation Error for event : ERROR  missing required fields :errtype","lname":"org.sunbird.telemetry.validator.TelemetryObjectValidatorV3" │
│ ,"tname":"Thread-76807","level":"INFO","HOSTNAME":"learner-64fb8c4d6d-fqstd","application.home":"/home/sunbird/learner/learning-service-1.0-SNAPSHOT"}                                                   │

coolbung Jan 24, 2023

Observed Cassandra connection errors in the notification service due to connection consistency set to QUORUM. Updating it to ONE seems to have resolved the notification service error. In turn, the error from the collection-certificate-generator job also seem to have gone.

@komalm @bhargav0702 please test out the various cases and confirm the certificates are processed without any consumer lag showing up.

gandham-santhosh Jan 24, 2023

Few more observations and actions. Setup cassandra as multi-node as data size is increasing (3 nodes to start with) and update the replication factor of the keyspaces, read/write consistency accordingly. Run compactions and setup cassandra reaper to schedule repairs. This will make sure data is consisted and updated in all the replicas in the cluster. http://cassandra-reaper.io/

SanthoshVasabhaktula Jan 25, 2023

@coolbung - Can you share the Cassandra cluster specs? Consistency should always be set to LOCAL_QUORUM but not ONE. When set to ONE any failure in any node leads to permanent data loss.

In addition, the Flink job going down frequently seems to be an infra issue. If any dependencies to the Flink job are not responding it will shutdown and stop processing the request queue.

@rhwarrier

coolbung Jan 25, 2023

I think this is a minimal installation, so there is no Cassandra cluster, but a single node. The content service cassandra connection is also set to ONE. @bhargav0702 can you share these details and action the cluster node scaling per these recommendations.

pramodkvarma · 2023-01-24T12:47:41Z

pramodkvarma
Jan 24, 2023
Maintainer

@madhucr @SanthoshVasabhaktula please look into this. Is this telemetry event generation issue?

2 replies

reshmi-nair Jan 25, 2023

@bhargav0702 Looking into this. Will provide an update with the findings.

SanthoshVasabhaktula Jan 25, 2023

@pramodkvarma - It is not a telemetry event generation issue, but seems that the certificate generation flink job is going down. Seems like an infra issue and @reshmi-nair is looking into it.

coolbung · 2023-01-26T10:32:33Z

coolbung
Jan 26, 2023

Summarising the issues in this case -

Symptom was certificates weren't getting issued
The collection-certificate-generator job was restarting continuously due to errors. We were seeing high consumer lags for the consumer/topic associated to this job.
These errors were pointing to a 500 error from the createNotification API in the learner service.
The learner service in turn calls the notification service to send notification. This API was also showing errors in the notification service logs
The notification service logs indicated a cassandra connection failure due to the connection consistency of QUORUM. This installation has a single cassandra node so QUORUM didnt work (event the content API config has ONE instead of QUORUM).
Changing it to ONE fixed the notification service error, which resulted into the job also not throwing errors anymore. Certificates are getting generated normally now.

cc @pramodkvarma @madhucr @gandham-santhosh

0 replies

coolbung · 2023-01-27T04:51:30Z

coolbung
Jan 27, 2023

On a related thread.

One of the courses that was affected has an assessment, and we are trying to re-issue certificates to the learners who have completed the course.

As part of troubleshooting / fixing this problem, in the past couple of weeks, the team updated the consumer lag multiple times. This has resulted into

No certificates being issued to learners in-spite of completing courses
The sunbird_courses.assessment_aggregator table does not have records for users who have completed. Due to this, re-issue isn't working.
We are currently requesting the affected learners to redo the assessment to get a certificate

Is there a way to replay any messages to get the sunbird_courses.assessment_aggregator table populated so that we dont need to ask the learners to re-do the assessment to get a certificate.

@SanthoshVasabhaktula @reshmi-nair

3 replies

maheshkumargangula Jan 27, 2023
Maintainer

@coolbung - If the Kafka topic has the data for the impacted period (past couple of weeks), please stop the assessment-agg job and change the consumer group value to oldest and start the job. It will replay the events and populate the sunbird_courses.assessment_aggregator.

If the data is not present in Kafka, We need to use the backup of the same topic (hope SECOR setup is done for the topic) to replay.
Cc: @SanthoshVasabhaktula

SanthoshVasabhaktula Jan 27, 2023

@coolbung - I couldn't understand what you meant by " the team updated the consumer lag multiple times". How is someone able to update lag? Did you mean offset?
Also why does assessment_aggregator table has no data? Is the assessment aggregator job not running?

coolbung Jan 27, 2023

Thanks @maheshkumargangula

@SanthoshVasabhaktula Yes I meant offset, not lag.

The assessment aggregator job is running, My guess is that when the team updated the offset to the latest message, the skipped messages were not picked up by the job, and so the score wasn't inserted into sunbird_courses.assessment_aggregator.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Sunbird Lern

Certificates are not generating due to collection-certificate-generator flink job not stable. #65

{{title}}

Replies: 4 comments 13 replies

{{title}}

{{title}}

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{title}}

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

Select a reply

Certificates are not generating due to collection-certificate-generator flink job not stable. #65

Replies: 4 comments · 13 replies

pramodkvarma Jan 24, 2023 Maintainer

maheshkumargangula Jan 27, 2023 Maintainer

Replies: 4 comments 13 replies

pramodkvarma
Jan 24, 2023
Maintainer

maheshkumargangula Jan 27, 2023
Maintainer