Merge pull request #1987 from Sunbird-Knowlg/transaction-event-processor

transaction-event-processor config added
Sunbird-Knowlg · May 22, 2024 · a69d16a · a69d16a
2 parents f6e5c3c + f5b3a05
commit a69d16a
Show file tree

Hide file tree

Showing 2 changed files with 56 additions and 0 deletions.
diff --git a/kubernetes/ansible/roles/flink-jobs-deploy/defaults/main.yml b/kubernetes/ansible/roles/flink-jobs-deploy/defaults/main.yml
@@ -126,6 +126,14 @@ collection_certificate_generator_rc_badcharlist: "{{ rc_bad_char_list | default(
 registry_sunbird_keyspace: "sunbird"
 cert_registry_table: "cert_registry"
 
+### transaction event processor related vars
+transaction_event_processor_consumer_parallelism: 1
+transaction_event_processor_parallelism: 1
+transaction_event_processor_producer_parallelism: 1
+transaction_event_processor_default_channel: "{{ default_channel | default('org.sunbird') }}"
+enable_audit_event_generator: true
+enable_audit_history_indexer: true
+enable_obsrv_metadata_generator: false
 
 ### to be removed
 job_classname: ""
@@ -293,6 +301,13 @@ flink_job_names:
     taskmanager_memory: 1024m
     taskslots: 1
     cpu_requests: 0.3
+  transaction-event-processor:
+    job_class_name: 'org.sunbird.job.transaction.task.TransactionEventProcessorStreamTask'
+    replica: 1
+    jobmanager_memory: 1024m
+    taskmanager_memory: 1024m
+    taskslots: 1
+    cpu_requests: 0.3
 
 ### Global vars
 middleware_course_keyspace: "sunbird_courses"

diff --git a/kubernetes/helm_charts/datapipeline_jobs/values.j2 b/kubernetes/helm_charts/datapipeline_jobs/values.j2
@@ -1321,3 +1321,44 @@ cassandra-data-migration:
     parallelism.default: 1
     jobmanager.execution.failover-strategy: region
     taskmanager.memory.network.fraction: 0.1
+
+transaction-event-processor:
+  transaction-event-processor: |+
+    include file("/data/flink/conf/base-config.conf")
+    job {
+      env = "{{ env_name }}"
+    }
+
+    kafka {
+      input.topic = "{{ env_name }}.learning.graph.events"
+      output.audit.topic = "{{ env_name }}.telemetry.raw"
+      output.obsrv.topic = "{{ env_name }}.transaction.meta"
+      groupId = "{{ env_name }}-transaction-event-processor-group"
+    }
+
+    task {
+      consumer.parallelism = {{ transaction_event_processor_consumer_parallelism }}
+      parallelism = {{ transaction_event_processor_parallelism }}
+      producer.parallelism = {{ transaction_event_processor_producer_parallelism }}
+      window.time = 60
+    }
+
+    schema {
+      basePath = "{kp_schema_base_path}"
+    }
+
+    channel.default = "{{ transaction_event_processor_default_channel }}"
+
+    job {
+        audit-event-generator =  {{ enable_audit_event_generator }}
+        audit-history-indexer = {{ enable_audit_history_indexer }}
+        obsrv-metadata-generator = {{ enable_obsrv_metadata_generator }}
+    }
+
+  flink-conf: |+
+    jobmanager.memory.flink.size: {{ flink_job_names['transaction-event-processor'].jobmanager_memory }}
+    taskmanager.memory.flink.size: {{ flink_job_names['transaction-event-processor'].taskmanager_memory }}
+    taskmanager.numberOfTaskSlots: {{ flink_job_names['transaction-event-processor'].taskslots }}
+    parallelism.default: 1
+    jobmanager.execution.failover-strategy: region
+    taskmanager.memory.network.fraction: 0.1