values.j2

namespace: {{ flink_namespace }}
imagepullsecrets: {{ imagepullsecrets }}
dockerhub: {{ dockerhub }}
repository: {{flink_repository|default('data-pipeline')}}
image_tag: {{ image_tag }}
cloud_service_provider: {{ cloud_service_provider }}
cloud_storage_key: {{ cloud_storage_key }}
cloud_storage_secret: {{ cloud_storage_secret }}
cloud_storage_endpoint: {{ cloud_private_storage_endpoint }}
cloud_storage_path_style_access: {{ cloud_storage_pathstyle_access }}
cloud_storage_project_id: {{ cloud_private_storage_project }}
serviceMonitor:
  enabled: {{ service_monitor_enabled | lower}}

replicaCount: {{taskmana_replicacount|default(1)}}

scale_properties:
  assessment-aggregator:
    enabled: {{ flink_job_names['assessment-aggregator'].scale_enabled | lower}}
    scale_target_value: {{ flink_job_names['assessment-aggregator'].scale_target_value | default(0) }}
    min_replica: {{ flink_job_names['assessment-aggregator'].min_replica | default(1) }}
    max_replica: {{ flink_job_names['assessment-aggregator'].max_replica | default(2) }}
  program-user-info:
    enabled: {{ flink_job_names['program-user-info'].scale_enabled | lower}}
    scale_target_value: {{ flink_job_names['program-user-info'].scale_target_value | default(0) }}
    min_replica: {{ flink_job_names['program-user-info'].min_replica | default(1) }}
    max_replica: {{ flink_job_names['program-user-info'].max_replica | default(2) }}

jobmanager:
  rpc_port: {{ jobmanager_rpc_port }}
  blob_port: {{ jobmanager_blob_port }}
  query_port: {{ jobmanager_query_port }}
  ui_port: {{ jobmanager_ui_port }}
  prom_port: {{ jobmanager_prom_port }}
  heap_memory: {{ jobmanager_heap_memory }}

service: {{ jobmanager_ui_service|to_json }}

rest_port: {{ jobmanager_ui_rest_port }}
resttcp_port: {{ jobmanager_ui_tcp_port }}

taskmanager:
  prom_port: {{ taskmanager_prom_port }}
  rpc_port: {{ taskmanager_rpc_port }}
  heap_memory: {{ taskmanager_heap_memory }}
  replicas: {{taskmanager_replicacount|default(1)}}

job_classname: {{ job_classname }}
{{ taskmanager_liveness | to_nice_yaml }}

log4j_console_properties: |
  # This affects logging for both user code and Flink
  rootLogger.level = {{ flink_jobs_console_log_level | default(INFO) }}
  rootLogger.appenderRef.console.ref = ConsoleAppender

  # Uncomment this if you want to _only_ change Flink's logging
  #logger.flink.name = org.apache.flink
  #logger.flink.level = {{ flink_jobs_console_log_level | default(INFO) }}

  # The following lines keep the log level of common libraries/connectors on
  # log level INFO. The root logger does not override this. You have to manually
  # change the log levels here.
  logger.akka.name = akka
  logger.akka.level = {{ flink_libraries_log_level | default(INFO) }}
  logger.kafka.name= org.apache.kafka
  logger.kafka.level = {{ flink_libraries_log_level | default(INFO) }}
  logger.hadoop.name = org.apache.hadoop
  logger.hadoop.level = {{ flink_libraries_log_level | default(INFO) }}
  logger.zookeeper.name = org.apache.zookeeper
  logger.zookeeper.level = {{ flink_libraries_log_level | default(INFO) }}

  # Log all infos to the console
  appender.console.name = ConsoleAppender
  appender.console.type = CONSOLE
  appender.console.layout.type = PatternLayout
  appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n

  # Suppress the irrelevant (wrong) warnings from the Netty channel handler
  logger.netty.name = org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline
  logger.netty.level = OFF

base_config: |
  kafka {
      broker-servers = "{{ kafka_brokers }}"
      producer.broker-servers = "{{ kafka_brokers }}"
      consumer.broker-servers = "{{ kafka_brokers }}"
      zookeeper = "{{ zookeepers }}"
      producer {
        max-request-size = {{ producer_max_request_size }}
        batch.size = {{ producer_batch_size }}
        linger.ms = {{ producer_linger_ms }}
      }
    }
    job {
      env = "{{ env_name }}"
      enable.distributed.checkpointing = true
      statebackend {
        blob {
          storage {
            account = "{% if cloud_service_provider == "azure" %}{{ cloud_storage_key }}.blob.core.windows.net{% elif cloud_service_provider in ["aws", "oci"] %}{{ cloud_storage_flink_bucketname }}{% endif %}"
            container = "{{ cloud_storage_flink_bucketname }}"
            checkpointing.dir = "checkpoint"
          }
        }
{% if cloud_service_provider == "azure" %}
        base.url = "wasbs://"${job.statebackend.blob.storage.container}"@"${job.statebackend.blob.storage.account}"/"${job.statebackend.blob.storage.checkpointing.dir}
{% elif cloud_service_provider in ["aws", "oci"] %}
        base.url = "s3://"${job.statebackend.blob.storage.container}"/"${job.statebackend.blob.storage.checkpointing.dir}
{% endif %}
      }
    }
    task {
      parallelism = 1
      consumer.parallelism = 1
      checkpointing.compressed = {{ checkpoint_compression_enabled|lower }}
      checkpointing.interval = {{ checkpoint_interval }}
      checkpointing.pause.between.seconds = {{ checkpoint_pause_between_seconds }}
      restart-strategy.attempts = {{ restart_attempts }}
      restart-strategy.delay = {{ restart_delay }} # in milli-seconds
    }
    redisdb.connection.timeout = {{ redis_timeout }}
    redis {
      host = {{ dp_redis_host }}
      port = 6379
    }
    redis-meta {
      {% if metadata2_redis_host is defined %}
      host = {{ metadata2_redis_host }}
      {% else %}
      host = {{ redis_host }}
      {% endif %}
      port = 6379
    }
    lms-cassandra {
      host = "{{ core_cassandra_connection_ip }}"
      port = "9042"
      isMultiDCEnabled = {{ cassandra_isMultiDCEnabled | lower }}
    }
    neo4j {
      routePath = "{{ neo4j_route_path }}"
      graph = "domain"
    }
    es {
        basePath = "{{ search_es_host }}"
    }
    schema {
      basePath = "{{ kp_schema_base_path }}"
      supportedVersion = {
        itemset = "2.0"
      }
    }
    ml-cassandra {
      host = "{{ core_cassandra_connection_ip }}"
      port = "9042"
      isMultiDCEnabled = {{ cassandra_isMultiDCEnabled | lower }}
    }
    ml-mongo {
        host = "{{ mongo_connection_ip }}"
        port = "27017"
        database = "{{ ml_database }}"
    }

    sunbird_instance_name = "{{ sunbird_installation | default('Sunbird')}}"

activity-aggregate-updater:
  activity-aggregate-updater: |+
    include file("/data/flink/conf/base-config.conf")
    kafka {
      input.topic = {{ kafka_topic_course_batch_job_request }}
      output.audit.topic = {{ kafka_topic_telemetry_raw }}
      output.failed.topic = {{ kafka_topic_activity_agg_failed }}
      output.certissue.topic = {{ kafka_topic_certificate_request }}
      groupId = {{ kafka_group_activity_agg }}
    }
    task {
      window.shards = {{ activity_agg_window_shards }}
      checkpointing.interval = {{ activity_agg_checkpointing_interval }}
      checkpointing.pause.between.seconds = {{ activity_agg_checkpointing_pause_interval }}
      restart-strategy.attempts = {{ restart_attempts }} # max 3 restart attempts
      restart-strategy.delay = 240000 # in milli-seconds # on max restarts (3) within 4 min the job will fail.
      consumer.parallelism = {{ activity_agg_consumer_parallelism }}
      dedup.parallelism = {{ activity_agg_dedup_parallelism }}
      activity.agg.parallelism = {{ activity_agg_parallelism }}
      enrolment.complete.parallelism = {{ enrolment_complete_parallelism }}
    }
    lms-cassandra {
      keyspace = "{{ middleware_course_keyspace }}"
      consumption.table = "{{ middleware_consumption_table }}"
      user_activity_agg.table = "{{ middleware_user_activity_agg_table }}"
      user_enrolments.table = "user_enrolments"
    }
    redis {
      database {
        relationCache.id = {{ redis_database_relation_cache_index | default(10) }}
      }
    }
    dedup-redis {
      host = {{ dedup_redis_host }}
      port = 6379
      database.index = {{ activity_agg_dedup_index }}
      database.expiry = {{ activity_agg_dedup_expiry }}
    }
    threshold.batch.read.interval = {{ activity_agg_batch_interval }}
    threshold.batch.read.size = {{ activity_agg_batch_read_size }}
    threshold.batch.write.size = {{ activity_agg_batch_write_size }}
    activity {
      module.aggs.enabled = true
      input.dedup.enabled = true
      filter.processed.enrolments = {{ activity_agg_enrolment_filter_processe_enabled | lower }}
      collection.status.cache.expiry = {{ activity_agg_collection_status_cache_expiry_time }}
    }
    service {
      search.basePath = "{{ kp_search_service_base_url }}"
    }


  flink-conf: |+
    jobmanager.memory.flink.size: {{ flink_job_names['activity-aggregate-updater'].jobmanager_memory }}
    taskmanager.memory.flink.size: {{ flink_job_names['activity-aggregate-updater'].taskmanager_memory }}
    taskmanager.numberOfTaskSlots: {{ flink_job_names['activity-aggregate-updater'].taskslots }}
    parallelism.default: 1
    jobmanager.execution.failover-strategy: region
    taskmanager.memory.network.fraction: 0.1

relation-cache-updater:
  relation-cache-updater: |+
    include file("/data/flink/conf/base-config.conf")
    kafka {
      input.topic = {{ kafka_topic_content_publish_request }}
      groupId = {{ kafka_group_relation_cache_updater }}
    }
    task {
      consumer.parallelism = {{ relation_cache_updater_consumer_parallelism }}
      parallelism = {{ relation_cache_updater_parallelism }}
    }
    lms-cassandra {
          keyspace = "{{ middleware_hierarchy_keyspace }}"
          table = "{{ middleware_content_hierarchy_table }}"
    }
    redis {
      database.index = {{ redis_database_index | default(10) }}
    }
    dp-redis {
      host = {{ dp_redis_host }}
      port = 6379
      database.index = {{ redis_dp_redis_database_index | default(5) }}
    }

  flink-conf: |+
    jobmanager.memory.flink.size: {{ flink_job_names['relation-cache-updater'].jobmanager_memory }}
    taskmanager.memory.flink.size: {{ flink_job_names['relation-cache-updater'].taskmanager_memory }}
    taskmanager.numberOfTaskSlots: {{ flink_job_names['relation-cache-updater'].taskslots }}
    parallelism.default: 1
    jobmanager.execution.failover-strategy: region
    taskmanager.memory.network.fraction: 0.1

enrolment-reconciliation:
  enrolment-reconciliation: |+
    include file("/data/flink/conf/base-config.conf")
    kafka {
      input.topic = {{ kafka_topic_enrolment_sync_request }}
      output.audit.topic = {{ kafka_topic_telemetry_raw }}
      output.failed.topic = {{ kafka_topic_activity_agg_failed }}
      output.certissue.topic = {{ kafka_topic_certificate_request }}
      groupId = {{ kafka_group_enrolment_reconciliation }}
    }
    task {
      restart-strategy.attempts = {{ restart_attempts }} # max 3 restart attempts
      restart-strategy.delay = 240000 # in milli-seconds # on max restarts (3) within 4 min the job will fail.
      consumer.parallelism = {{ enrolment_reconciliation_consumer_parallelism }}
      enrolment.reconciliation.parallelism = {{ enrolment_reconciliation_parallelism }}
      enrolment.complete.parallelism = {{ enrolment_complete_parallelism }}
    }
    lms-cassandra {
      keyspace = "{{ middleware_course_keyspace }}"
      consumption.table = "{{ middleware_consumption_table }}"
      user_activity_agg.table = "{{ middleware_user_activity_agg_table }}"
      user_enrolments.table = "user_enrolments"
    }
    redis {
      database {
        relationCache.id = {{ redis_database_relation_cache_index | default(10) }}
      }
    }
    threshold.batch.write.size = {{ enrolment_reconciliation_batch_write_size }}
    activity {
      module.aggs.enabled = true
      collection.status.cache.expiry = {{ enrolment_reconciliation_collection_status_cache_expiry_time }}
    }
    service {
      search.basePath = "{{ kp_search_service_base_url }}"
    }


  flink-conf: |+
    jobmanager.memory.flink.size: {{ flink_job_names['enrolment-reconciliation'].jobmanager_memory }}
    taskmanager.memory.flink.size: {{ flink_job_names['enrolment-reconciliation'].taskmanager_memory }}
    taskmanager.numberOfTaskSlots: {{ flink_job_names['enrolment-reconciliation'].taskslots }}
    parallelism.default: 1
    jobmanager.execution.failover-strategy: region
    taskmanager.memory.network.fraction: 0.1

collection-cert-pre-processor:
  collection-cert-pre-processor: |+
    include file("/data/flink/conf/base-config.conf")
    kafka {
      input.topic = {{ kafka_topic_certificate_request }}
      output.topic = {{ kafka_topic_generate_certificate_request }}
      output.failed.topic = {{ kafka_topic_certificate_failed }}
      groupId = {{ kafka_group_collection_pre_processor }}
    }
    task {
      restart-strategy.attempts = {{ restart_attempts }} # max 3 restart attempts
      restart-strategy.delay = 240000 # in milli-seconds # on max restarts (3) within 4 min the job will fail.
      parallelism = {{collection_cert_pre_processor_consumer_parallelism}}
      consumer.parallelism = {{ collection_cert_pre_processor_consumer_parallelism }}
      generate_certificate.parallelism = {{generate_certificate_parallelism}}
    }
    lms-cassandra {
      keyspace = "{{ middleware_course_keyspace }}"
      consumption.table = "{{ middleware_consumption_table }}"
      user_enrolments.table = "{{ middleware_user_enrolments_table }}"
      course_batch.table = "{{ middleware_course_batch_table }}"
      assessment_aggregator.table = "{{ middleware_assessment_aggregator_table }}"
      user_activity_agg.table = "{{ middleware_user_activity_agg_table }}"
    }
    redis {
          database {
            contentCache.id = {{ redis_database_content_cache_index | default(5) }}
            collectionCache.id = {{ redis_collection_cache_index | default(0) }}
          }
    }
    cert_domain_url = "{{ cert_domain_url }}"
    user_read_api = "/private/user/v1/read"
    content_read_api = "/content/v3/read"
    service {
      content.basePath = "{{ content_service_base_url }}"
      learner.basePath = "{{ userorg_service_base_url }}"
    }
    enable.suppress.exception = {{ collection_certificate_pre_processor_enable_suppress_exception | lower }}
    redis-meta {
    {% if metadata2_redis_host is defined %}
      host = {{ metadata2_redis_host }}
    {% else %}
      host = {{ redis_host }}
    {% endif %}
      port = 6379
    }
    assessment.metrics.supported.contenttype = ["SelfAssess"]
    cert_cloud_storage_type = "{{ cert_cloud_storage_type }}"
    cert_cloud_storage_secret = "{{ cert_cloud_storage_secret }}"
    cert_cloud_storage_key = "{{ cert_cloud_storage_key }}"
    cloud_storage_base_url = "{{ cloud_storage_base_url }}"
    cloud_store_base_path_placeholder = "{{ cloud_store_base_path_placeholder | default('CLOUD_BASE_PATH')}}"
    content_cloud_storage_container = "{{ cloud_storage_content_bucketname }}"
    cloud_storage_cname_url = "{{ cloud_storage_cname_url }}"

  flink-conf: |+
    jobmanager.memory.flink.size: {{ flink_job_names['collection-cert-pre-processor'].jobmanager_memory }}
    taskmanager.memory.flink.size: {{ flink_job_names['collection-cert-pre-processor'].taskmanager_memory }}
    taskmanager.numberOfTaskSlots: {{ flink_job_names['collection-cert-pre-processor'].taskslots }}
    parallelism.default: 1
    jobmanager.execution.failover-strategy: region
    taskmanager.memory.network.fraction: 0.1

collection-certificate-generator:
  collection-certificate-generator: |+
    include file("/data/flink/conf/base-config.conf")
    kafka {
      input.topic = {{ kafka_topic_generate_certificate_request }}
      output.audit.topic = {{ kafka_topic_telemetry_raw }}
      groupId = {{ kafka_group_certificate_generator }}
    }
    task {
      restart-strategy.attempts = {{ restart_attempts }} # max 3 restart attempts
      restart-strategy.delay = 240000 # in milli-seconds # on max restarts (3) within 4 min the job will fail.
      consumer.parallelism = {{ collection_certificate_generator_consumer_parallelism }}
      parallelism = {{ collection_certificate_generator_parallelism }}
    }
    lms-cassandra {
      keyspace = "{{ middleware_course_keyspace }}"
      user_enrolments.table = "{{ middleware_user_enrolments_table }}"
      course_batch.table = "{{ middleware_course_batch_table }}"
      sbkeyspace = "{{ registry_sunbird_keyspace }}"
      certreg.table ="{{ cert_registry_table }}"
    }
    cert_domain_url = "{{ cert_domain_url }}"
    cert_container_name = "{{ cert_container_name }}"
    cert_cloud_storage_type = "{{ cert_cloud_storage_type }}"
    cert_cloud_storage_secret = "{{ cert_cloud_storage_secret }}"
    cert_cloud_storage_endpoint = "{{cert_cloud_storage_endpoint}}"
    cert_cloud_storage_key = "{{ cert_cloud_storage_key }}"
    cloud_storage_base_url = "{{ cloud_storage_base_url }}"
    cloud_store_base_path_placeholder = "{{ cloud_store_base_path_placeholder | default('CLOUD_BASE_PATH')}}"
    content_cloud_storage_container = "{{ cloud_storage_content_bucketname }}"
    cloud_storage_cname_url = "{{ cloud_storage_cname_url }}"
    service {
      certreg.basePath = "{{ cert_reg_service_base_url }}"
      learner.basePath = "{{ userorg_service_base_url }}"
      enc.basePath = "{{ enc_service_base_url }}"
      rc.basePath = "{{ cert_rc_base_url }}"
      rc.entity = "{{ cert_rc_entity }}"
      rc.rcApiKey = "{{ sunbird_api_auth_token }}"
    }
    enable.suppress.exception = {{ collection_certificate_generator_enable_suppress_exception | lower }}
    enable.rc.certificate = {{ collection_certificate_generator_enable_rc_certificate | lower }}
    task.rc.badcharlist = {{ collection_certificate_generator_rc_badcharlist }}


  flink-conf: |+
    jobmanager.memory.flink.size: {{ flink_job_names['collection-certificate-generator'].jobmanager_memory }}
    taskmanager.memory.flink.size: {{ flink_job_names['collection-certificate-generator'].taskmanager_memory }}
    taskmanager.numberOfTaskSlots: {{ flink_job_names['collection-certificate-generator'].taskslots }}
    parallelism.default: 1
    jobmanager.execution.failover-strategy: region
    taskmanager.memory.network.fraction: 0.1


legacy-certificate-migrator:
  legacy-certificate-migrator: |+
    include file("/data/flink/conf/base-config.conf")
    kafka {
      input.topic = {{ kafka_topic_migrate_certificate_request }}
      output.audit.topic = {{ kafka_topic_telemetry_raw }}
      groupId = {{ kafka_group_certificate_migrator }}
    }
    task {
      restart-strategy.attempts = {{ restart_attempts }} # max 3 restart attempts
      restart-strategy.delay = 240000 # in milli-seconds # on max restarts (3) within 4 min the job will fail.
      consumer.parallelism = {{ legacy_certificate_migrator_consumer_parallelism }}
      parallelism = {{ legacy_certificate_migrator_parallelism }}
    }
    lms-cassandra {
      keyspace = "{{ middleware_course_keyspace }}"
      user_enrolments.table = "{{ middleware_user_enrolments_table }}"
      course_batch.table = "{{ middleware_course_batch_table }}"
      sbkeyspace = "{{ registry_sunbird_keyspace }}"
      certreg.table ="{{ cert_registry_table }}"
    }
    cert_domain_url = "{{ cert_domain_url }}"
    cert_container_name = "{{ cert_container_name }}"
    cert_cloud_storage_type = "{{ cert_cloud_storage_type }}"
    cert_cloud_storage_secret = "{{ cert_cloud_storage_secret }}"
    cert_cloud_storage_key = "{{ cert_cloud_storage_key }}"
    cloud_storage_base_url = "{{ cloud_storage_base_url }}"
    cloud_store_base_path_placeholder = "{{ cloud_store_base_path_placeholder }}"
    content_cloud_storage_container = "{{ cloud_storage_content_bucketname }}"
    cloud_storage_cname_url = "{{ cloud_storage_cname_url }}"
    service {
      certreg.basePath = "{{ cert_reg_service_base_url }}"
      learner.basePath = "{{ userorg_service_base_url }}"
      enc.basePath = "{{ enc_service_base_url }}"
      rc.basePath = "{{ cert_rc_base_url }}"
      rc.entity = "{{ cert_rc_entity }}"
      rc.rcApiKey = "{{ sunbird_api_auth_token }}"
    }
    enable.suppress.exception = {{ certificate_migrator_enable_suppress_exception | lower }}
    enable.rc.certificate = {{ collection_certificate_generator_enable_rc_certificate | lower }}
    task.rc.badcharlist = {{ collection_certificate_generator_rc_badcharlist }}

  flink-conf: |+
    jobmanager.memory.flink.size: {{ flink_job_names['legacy-certificate-migrator'].jobmanager_memory }}
    taskmanager.memory.flink.size: {{ flink_job_names['legacy-certificate-migrator'].taskmanager_memory }}
    taskmanager.numberOfTaskSlots: {{ flink_job_names['legacy-certificate-migrator'].taskslots }}
    parallelism.default: 1
    jobmanager.execution.failover-strategy: region
    taskmanager.memory.network.fraction: 0.1


merge-user-courses:
  merge-user-courses: |+
    include file("/data/flink/conf/base-config.conf")
    kafka {
      input.topic = {{ kafka_topic_lms_user_account }}
      output.failed.topic = {{ kafka_topic_learning_failed }}
      groupId = {{ kafka_group_merge_courses }}
      output.course.batch.updater.topic = {{ kafka_topic_course_batch_job_request }}
    }
    task {
      consumer.parallelism = {{ merge_user_courses_consumer_parallelism }}
      parallelism = {{ merge_user_courses_parallelism }}
      course_batch_updater.parallelism = {{ merge_user_courses_course_batch_parallelism }}
    }
    lms-cassandra {
      keyspace = "{{ middleware_course_keyspace }}"
      content_consumption.table = "{{ middleware_consumption_table }}"
      user_enrolments.table = "{{ middleware_user_enrolments_table }}"
      user_activity_agg.table = "{{ middleware_user_activity_agg_table }}"
    }
    course.date.format = "{{ merge_user_courses_course_date_format }}"

  flink-conf: |+
    jobmanager.memory.flink.size: {{ flink_job_names['merge-user-courses'].jobmanager_memory }}
    taskmanager.memory.flink.size: {{ flink_job_names['merge-user-courses'].taskmanager_memory }}
    taskmanager.numberOfTaskSlots: {{ flink_job_names['merge-user-courses'].taskslots }}
    parallelism.default: 1
    jobmanager.execution.failover-strategy: region
    taskmanager.memory.network.fraction: 0.1

assessment-aggregator:
  assessment-aggregator: |+
    include file("/data/flink/conf/base-config.conf")
    kafka {
      producer.broker-servers = "{{ kafka_brokers }}"
      consumer.broker-servers = "{{ kafka_brokers }}"
      zookeeper = "{{ zookeepers }}"
      input.topic = {{ kafka_topic_assessment }}
      failed.topic= {{ kafka_topic_assessment_failed }}
      groupId = {{ kafka_group_assessment_aggregator }}
      output.certissue.topic = {{ kafka_topic_certificate_request }}
    }
    task {
      consumer.parallelism = {{ assessaggregator_consumer_parallelism }}
      downstream.parallelism = {{ assessaggregator_downstream_parallelism }}
      assessaggregator {
        parallelism = {{ assessaggregator_parallelism }}
      }
      scoreaggregator.parallelism = {{ assessaggregator_scoreaggregator_parallelism }}
    }
    lms-cassandra {
      keyspace = "{{ middleware_cassandra_courses_keyspace }}"
      table = "{{ middleware_cassandra_assessment_aggregator_table }}"
      questionudttype= "{{ middleware_cassandra_assessment_question_type }}"
      enrolmentstable = "{{ middleware_cassandra_user_enrolments_table }}"
      activitytable = "{{ middleware_cassandra_user_activity_agg_table }}"
    }
    redis {
      database {
        relationCache.id = {{ redis_database_relation_cache_index | default(10) }}
        contentCache.id = {{ redis_database_content_cache_index | default(5) }}
      }
    }
    assessment.skip.missingRecords = true
    content.read.api = "{{ content_read_api_host }}/{{ content_read_api_endpoint }}"
    user.activity.agg.type="attempt_metrics"

  flink-conf: |+
    jobmanager.memory.flink.size: {{ flink_job_names['assessment-aggregator'].jobmanager_memory }}
    taskmanager.memory.flink.size: {{ flink_job_names['assessment-aggregator'].taskmanager_memory }}
    taskmanager.numberOfTaskSlots: {{ flink_job_names['assessment-aggregator'].taskslots }}
    parallelism.default: 1
    jobmanager.execution.failover-strategy: region
    taskmanager.memory.network.fraction: 0.1

notification-job:
  notification-job: |+
    include file("/data/flink/conf/base-config.conf")
    kafka {
      input.topic = {{ kafka_topic_lms_notification }}
      groupId = {{ kafka_group_lms_notification }}
    }
    task {
      restart-strategy.attempts = {{ restart_attempts }} # max 3 restart attempts
      restart-strategy.delay = 240000 # in milli-seconds # on max restarts (3) within 4 min the job will fail.
      consumer.parallelism = {{ notification_job_consumer_parallelism }}
      parallelism = {{ notification_job_parallelism }}
    }
    fcm_account_key= "{{ core_vault_sunbird_fcm_account_key }}"
    sms_auth_key= "{{sunbird_msg_91_auth}}"
    mail_server_from_email= "{{sunbird_mail_server_from_email}}"
    sms_default_sender= "{{sunbird_notification_msg_default_sender}}"
    mail_server_username= "{{sunbird_mail_server_username}}"
    mail_server_password= "{{sunbird_mail_server_password}}"
    mail_server_host= "{{sunbird_mail_server_host}}"
    mail_server_port= "{{sunbird_mail_server_port}}"


  flink-conf: |+
    jobmanager.memory.flink.size: {{ flink_job_names['notification-job'].jobmanager_memory }}
    taskmanager.memory.flink.size: {{ flink_job_names['notification-job'].taskmanager_memory }}
    taskmanager.numberOfTaskSlots: {{ flink_job_names['notification-job'].taskslots }}
    parallelism.default: 1
    jobmanager.execution.failover-strategy: region
    taskmanager.memory.network.fraction: 0.1


user-cache-updater-v2:
  user-cache-updater-v2: |+
    include file("/data/flink/conf/base-config.conf")
    kafka {
      input.topic = {{ kafka_topic_telemetry_audit }}
      groupId = {{ kafka_group_user_cache_updater }}
    }
    task {
      restart-strategy.attempts = {{ restart_attempts }} # max 3 restart attempts
      restart-strategy.delay = 240000 # in milli-seconds # on max restarts (3) within 4 min the job will fail.
      consumer.parallelism = {{ user_cache_updater_job_consumer_parallelism }}
      parallelism = {{ user_cache_updater_job_parallelism }}
      usercache.updater.parallelism = {{ user_cache_updater_job_parallelism }}
    }
    regd.user.producer.pid = "userorg-service"
    user.self.signin.types = ["google","self"]
    user.validated.types = ["sso"]
    user.self.signin.key = "Self-Signed-In"
    user.valid.key = "Validated"
    user.read.url.fields = "locations,organisations"
    user.read.api.error = ["CLIENT_ERROR"]
    # redis-metadata
    redis-meta {
      database {
        userstore.id = {{ redis_meta_database_user_index | default(12) }}
        key.expiry.seconds = 3600
      }
    }
    #(user read api details)
    user_read_api = "/private/user/v1/read/"
    framework_read_api = "/api/framework/v1/read/"
    service {
      taxonomy.basePath = "{{ domain_url }}"
      userorg.basePath = "{{ userorg_service_base_url }}"
    }


  flink-conf: |+
    jobmanager.memory.flink.size: {{ flink_job_names['user-cache-updater-v2'].jobmanager_memory }}
    taskmanager.memory.flink.size: {{ flink_job_names['user-cache-updater-v2'].taskmanager_memory }}
    taskmanager.numberOfTaskSlots: {{ flink_job_names['user-cache-updater-v2'].taskslots }}
    parallelism.default: 1
    jobmanager.execution.failover-strategy: region
    taskmanager.memory.network.fraction: 0.1

program-user-info:
  program-user-info: |+
    include file("/data/flink/conf/base-config.conf")
    kafka {
      producer.broker-servers = "{{ kafka_brokers }}"
      consumer.broker-servers = "{{ kafka_brokers }}"
      zookeeper = "{{ zookeepers }}"
      input.topic = {{ kafka_topic_programuser_info }}
      groupId = {{ kafka_group_programuser_info }}
    }
    task {
      consumer.parallelism = {{ programuserinfo_consumer_parallelism }}
      downstream.parallelism = {{ programuserinfo_downstream_parallelism }}
      programuser {
        parallelism = {{ programuserinfo_programuser_parallelism }}
      }

    }
    ml-cassandra {
      keyspace = "{{ middleware_cassandra_programuserinfo_keyspace }}"
      table = "{{ middleware_cassandra_programuserinfo_table }}"
    }

  flink-conf: |+
    jobmanager.memory.flink.size: {{ flink_job_names['program-user-info'].jobmanager_memory }}
    taskmanager.memory.flink.size: {{ flink_job_names['program-user-info'].taskmanager_memory }}
    taskmanager.numberOfTaskSlots: {{ flink_job_names['program-user-info'].taskslots }}
    parallelism.default: 1
    jobmanager.execution.failover-strategy: region
    taskmanager.memory.network.fraction: 0.1

user-deletion-cleanup:
  user-deletion-cleanup: |+
    include file("/data/flink/conf/base-config.conf")
    kafka {
      input.topic = ${job.env}".delete.user"
      groupId = ${job.env}"-delete-user-group"
    }
    task {
      user.deletion.cleanup.parallelism = {{ user_deletion_cleanup_job_parallelism }}
    }

    service {
        lms {
            basePath = "{{ lms_service_base_url }}"
        }
        userorg {
            basePath = "{{ userorg_service_base_url }}"
        }
    }

    sunbird_keycloak_user_federation_provider_id={{core_vault_sunbird_keycloak_user_federation_provider_id}}
    user_read_api = "/user/v5/read/"
    batch_search_api = "/course/v1/batch/list"

    user {
        keyspace = "sunbird"
        lookup.table = "user_lookup"
        table = "user"
        externalIdentity.table = "usr_external_identity"
        org.table = "user_organisation"
    }


  flink-conf: |+
    jobmanager.memory.flink.size: {{ flink_job_names['user-deletion-cleanup'].jobmanager_memory }}
    taskmanager.memory.flink.size: {{ flink_job_names['user-deletion-cleanup'].taskmanager_memory }}
    taskmanager.numberOfTaskSlots: {{ flink_job_names['user-deletion-cleanup'].taskslots }}
    parallelism.default: 1
    jobmanager.execution.failover-strategy: region
    taskmanager.memory.network.fraction: 0.1

ml-user-delete:
  ml-user-delete: |+
    include file("/data/flink/conf/base-config.conf")
    kafka {
      input.topic = ${job.env}".delete.user"
      groupId = ${job.env}"-ml-user-delete-group"
    }
    task {
      ml.user.delete.parallelism = {{ ml_user_delete_job_parallelism }}
    }

  flink-conf: |+
    jobmanager.memory.flink.size: {{ flink_job_names['ml-user-delete'].jobmanager_memory }}
    taskmanager.memory.flink.size: {{ flink_job_names['ml-user-delete'].taskmanager_memory }}
    taskmanager.numberOfTaskSlots: {{ flink_job_names['ml-user-delete'].taskslots }}
    parallelism.default: 1
    jobmanager.execution.failover-strategy: region
    taskmanager.memory.network.fraction: 0.1

user-ownership-transfer:
  user-ownership-transfer: |+
    include file("/data/flink/conf/base-config.conf")
    kafka {
      input.topic = ${job.env}".user.ownership.transfer"
      groupId = ${job.env}"-user-ownership-transfer-group"
    }
    task {
      user.ownership.transfer.parallelism = {{ user_ownership_transfer_job_parallelism }}
    }
    lms-cassandra {
      course_batch.table = "{{ middleware_course_batch_table }}"
      keyspace = "{{ middleware_course_keyspace }}"
    }
    service {
         lms {
             basePath = "{{ lms_service_base_url }}"
         }
         userorg {
             basePath = "{{ userorg_service_base_url }}"
         }
    }
    user_read_api = "/user/v5/read/"
    batch_search_api = "/course/v1/batch/list"
    threshold.batch.write.size = {{ user_ownership_transfer_batch_write_size }}

  flink-conf: |+
    jobmanager.memory.flink.size: {{ flink_job_names['user-ownership-transfer'].jobmanager_memory }}
    taskmanager.memory.flink.size: {{ flink_job_names['user-ownership-transfer'].taskmanager_memory }}
    taskmanager.numberOfTaskSlots: {{ flink_job_names['user-ownership-transfer'].taskslots }}
    parallelism.default: 1
    jobmanager.execution.failover-strategy: region
    taskmanager.memory.network.fraction: 0.1