Enrolment Reconciliation

'enrolment-reconciliation' job syncs the progress of the enrolled user. This job perform same tasks as 'activity-aggregator-updater' flink job. This is triggered by the user if the progress is not updated during course consumption.

Configuration variables:

Variable	Default Value	Purpose
kafka.input.topic	{{env}}.batch.enrolment.sync.request	Kafka topic from which messages/events are read to be processed.
kafka.output.failed.topic	{{env}}.enrolment.reconciliation.failed	Kafka topic to which message is written when an exception occurs while processing an event.
kafka.audit.topic	{{env}}.telemetry.raw	Kakfa topic to which and audit message is written to.
kafka.certissue.topic	{{env}}.issue.certificate.request	Kafka topic used to trigger certificate issue pre-processor job
kafka.groupId	{{env}}-enrolment-reconciliation-group	Kafka input topic group Id
lms-cassandra.keyspace	sunbird_courses	Cassandra keyspace name
lms-cassandra.content_consumption.table	user_content_consumption	Cassandra table used to store content wise data for a collection of a batch by a user. Content progress, status etc
lms-cassandra.user_enrolments.table	user_enrolments	Cassandra table used to store user enrolment data in a collection of a particular batch. This also holds the consumption progress, enrolment status and issued certificate details
lms-cassandra.user_activity_agg.table	user_activity_agg	Cassandra table used to store user consumption aggregate details of a collection in a batch. Aggregates like the consumption completed content count
redis.database.relationCache.id	10	Redis index from which computed data like leafnodes and optionalnodes is read.
threshold.batch.write.size	10	Property used to specify batch size of the database update queries while updating a specific cassandra table in batch format
activity.module.aggs.enabled	true	Used to configure if the consumption aggregation calculation is to be enabled on course leaf nodes
activity.filter.processed.enrolments	true	Used to configure if the activity aggregation process is to be skipped for user enrolments with status 2 (completed courses)
activity.collection.status.cache.expiry	3600 (in seconds)	Expiry time of TTL cache set to read or store latest collection 'status' information. If latest TTL cache doesnt have collection 'status' information, then the same is read from Search service configured below and TTL cache will be updated.
service.search.basePath	IP of the search service	IP of the search service

Sample event:

{
  "eid": "BE_JOB_REQUEST",
  "ets": 1683556693275,
  "mid": "LP.1683556693275.a18caaec-50ac-4c19-bcfe-84e2ad9aa5eb",
  "actor": {
    "type": "System",
    "id": "Course Batch Updater"
  },
  "context": {
    "pdata": {
      "ver": "1.0",
      "id": "org.sunbird.platform"
    }
  },
  "object": {
    "type": "CourseBatchEnrolment",
    "id": "0131029098223779842_fbe926ac-a395-40e4-a65b-9b4f711d7642"
  },
  "edata": {
    "action": "user-enrolment-sync",
    "iteration": 1,
    "batchId": "0131029098223779842",
    "userId": "fbe926ac-a395-40e4-a65b-9b4f711d7642",
    "courseId": "do_21310287978188800014457"
  }
}

Source code:

data-pipeline/lms-jobs/enrolment-reconciliation at release-5.3.0 · Sunbird-Lern/data-pipelineGitHub