Denorm optimisations - redis multiprocess fixes

ansible/roles/redis-multiprocess/tasks/main.yml

@@ -16,6 +16,7 @@
 
 - name: Copy redis conf files
   become: yes
+  become_user: "{{ redis_user }}"
   template: src=redis.conf.j2 dest={{ redis_dir }}/redis-stable/redis-{{ item.value.name }}.conf
   vars:
     _port_: "{{ item.value.port }}"

data-pipeline-flink/de-normalization/src/main/scala/org/sunbird/dp/denorm/task/DenormalizationStreamTask.scala

@@ -94,6 +94,6 @@ object DenormalizationStreamTask {
 class DenormKeySelector(config: DenormalizationConfig) extends KeySelector[Event, Int] {
   val shards = config.windowShards
   override def getKey(in: Event): Int = {
-    in.did().hashCode % shards
+    if (Option(in.did()) == None) "".hashCode % shards else in.did().hashCode % shards
   }
 }

data-pipeline-flink/de-normalization/src/test/scala/org/sunbird/dp/fixture/EventFixture.scala

@@ -142,6 +142,16 @@ object EventFixture {
        |"pdata":{"id":"sunbird.app","pid":"sunbird.app","ver":"2.3.144"},"sid":"df936f82-e982-41ec-8412-70d414458272",
        |"did":"45f32f48592cb9bcf26bef9178b7bd20abe24932"},"flags":{"dd_processed":true},
        |"mid":"mid6","type":"events","object":{"id":"PCZKA4","type":"qr",
+       |"version":"","rollup":{}}}""".stripMargin,
+
+      // event without did
+      s"""{"actor":{"type":"User","id":"610bab7d-1450-4e54-bf78-c7c9b14dbc82"},"eid":"INTERRUPT",
+       |"edata":{"type":"OTHER","subtype":"sheen-animation-ended","id":"library","pageid":"library","extra":{"pos":[]}},
+       |"ver":"3.0","syncts":1.579564974098E12,"@timestamp":"2020-01-21T00:02:54.098Z","ets":$currentDate,
+       |"context":{"cdata":[],"env":"home","channel":"505c7c48ac6dc1edc9b08f21db5a571d",
+       |"pdata":{"id":"sunbird.app","pid":"sunbird.app","ver":"2.3.144"},"sid":"df936f82-e982-41ec-8412-70d414458272"},
+       |"flags":{"dd_processed":true},
+       |"mid":"mid11","type":"events","object":{"id":"PCZKA4","type":"qr",
        |"version":"","rollup":{}}}""".stripMargin
 
   )

data-pipeline-flink/de-normalization/src/test/scala/org/sunbird/dp/spec/DenormalizationStreamTaskTestSpec.scala

@@ -228,7 +228,7 @@ class DenormalizationStreamTaskTestSpec extends BaseTestSpec {
     BaseMetricsReporter.gaugeMetrics(s"${denormConfig.jobName}.${denormConfig.deviceTotal}").getValue() should be (9)
 
     BaseMetricsReporter.gaugeMetrics(s"${denormConfig.jobName}.${denormConfig.eventsExpired}").getValue() should be (1)
-    BaseMetricsReporter.gaugeMetrics(s"${denormConfig.jobName}.${denormConfig.eventsSkipped}").getValue() should be (1) // Skipped INTERRUPT event
+    BaseMetricsReporter.gaugeMetrics(s"${denormConfig.jobName}.${denormConfig.eventsSkipped}").getValue() should be (2) // Skipped INTERRUPT event
 
   }