Add aggregate topic support

Author: callmekatootie

README.md

@@ -23,6 +23,7 @@ The following parameters can be set in config files or in env variables:
 - UBAHN_CREATE_TOPIC: the create ubahn entity Kafka message topic, default value is 'u-bahn.action.create'
 - UBAHN_UPDATE_TOPIC: the update ubahn entity Kafka message topic, default value is 'u-bahn.action.update'
 - UBAHN_DELETE_TOPIC: the delete ubahn entity Kafka message topic, default value is 'u-bahn.action.delete'
+- UBAHN_AGGREGATE_TOPIC: the ubahn entity aggregate topic, that contains create, update and delete topics. Default value is 'u-bahn.action.aggregate'
 - ES.HOST: Elasticsearch host, default value is 'localhost:9200'
 - ES.AWS_REGION: The Amazon region to use when using AWS Elasticsearch service, default value is 'us-east-1'
 - ES.API_VERSION: Elasticsearch API version, default value is '7.4'

VERIFICATION.md

@@ -1,5 +1,28 @@
 # Verification
 
+**NOTE** - For all kafka message below, update the topic to be the one set in config.UBAHN_AGGREGATE_TOPIC and inside the payload object, create a new attribute named `originalTopic` with the value of the original topic. Example:
+
+```
+{
+  "topic": "u-bahn.action.aggregate",
+  "originator": "u-bahn-api",
+  "timestamp": "2019-07-08T00:00:00.000Z",
+  "mime-type": "application/json",
+  "payload": {
+    "originalTopic": "u-bahn.action.create"
+    "resource": "user",
+    "id": "391a3656-9a01-47d4-8c6d-64b68c44f212",
+    "handle": "user"
+  }
+}
+```
+
+Additionally, you will be entering the messages into only one topic:
+
+```
+docker exec -it ubahn-data-processor-es_kafka /opt/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic u-bahn.action.aggregate
+```
+
 1. start kafka server, start elasticsearch, initialize Elasticsearch, start processor app
 2. start kafka-console-producer to write messages to `u-bahn.action.create`
 topic:

config/default.js

@@ -17,6 +17,7 @@ module.exports = {
   UBAHN_CREATE_TOPIC: process.env.UBAHN_CREATE_TOPIC || 'u-bahn.action.create',
   UBAHN_UPDATE_TOPIC: process.env.UBAHN_UPDATE_TOPIC || 'u-bahn.action.update',
   UBAHN_DELETE_TOPIC: process.env.UBAHN_DELETE_TOPIC || 'u-bahn.action.delete',
+  UBAHN_AGGREGATE_TOPIC: process.env.UBAHN_AGGREGATE_TOPIC || 'u-bahn.action.aggregate',
 
   ES: {
     HOST: process.env.ES_HOST || 'localhost:9200',

docker-kafka-es/docker-compose.yml

@@ -12,7 +12,8 @@ services:
       - "9092:9092"
     environment:
       KAFKA_ADVERTISED_HOST_NAME: localhost
-      KAFKA_CREATE_TOPICS: "u-bahn.action.create:1:1,u-bahn.action.update:1:1,u-bahn.action.delete:1:1"
+      # KAFKA_CREATE_TOPICS: "u-bahn.action.create:1:1,u-bahn.action.update:1:1,u-bahn.action.delete:1:1"
+      KAFKA_CREATE_TOPICS: "u-bahn.action.aggregate:1:1"
       KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
   # esearch:
   #   image: docker.elastic.co/elasticsearch/elasticsearch:7.4.2

src/app.js

@@ -69,7 +69,7 @@ const dataHandler = (messageSet, topic, partition) => Promise.each(messageSet, a
   }
   const transactionId = _.uniqueId('transaction_')
   try {
-    switch (topic) {
+    switch (messageJSON.payload.originalTopic) {
       case config.UBAHN_CREATE_TOPIC:
         await ProcessorService.processCreate(messageJSON, transactionId)
         break
@@ -79,6 +79,8 @@ const dataHandler = (messageSet, topic, partition) => Promise.each(messageSet, a
       case config.UBAHN_DELETE_TOPIC:
         await ProcessorService.processDelete(messageJSON, transactionId)
         break
+      default:
+        throw new Error(`Unknown original topic: ${messageJSON.payload.originalTopic}`)
     }
 
     logger.debug(`Successfully processed message with count ${messageCount}`)
@@ -106,7 +108,8 @@ const check = () => {
   return connected
 }
 
-const topics = [config.UBAHN_CREATE_TOPIC, config.UBAHN_UPDATE_TOPIC, config.UBAHN_DELETE_TOPIC]
+// const topics = [config.UBAHN_CREATE_TOPIC, config.UBAHN_UPDATE_TOPIC, config.UBAHN_DELETE_TOPIC]
+const topics = [config.UBAHN_AGGREGATE_TOPIC]
 
 consumer
   .init([{

src/common/helper.js

@@ -128,7 +128,8 @@ async function updateUser (userId, body, seqNo, primaryTerm, transactionId) {
     transactionId,
     body: { doc: body },
     if_seq_no: seqNo,
-    if_primary_term: primaryTerm
+    if_primary_term: primaryTerm,
+    refresh: 'wait_for'
   })
 }
 
@@ -161,7 +162,8 @@ async function updateOrg (organizationId, body, seqNo, primaryTerm, transactionI
     transactionId,
     body: { doc: body },
     if_seq_no: seqNo,
-    if_primary_term: primaryTerm
+    if_primary_term: primaryTerm,
+    refresh: 'wait_for'
   })
 }
 

src/services/ProcessorService.js

@@ -11,6 +11,7 @@ const {
   userResources,
   organizationResources
 } = require('../common/constants')
+const config = require('config')
 
 /**
  * Process create entity message
@@ -27,8 +28,8 @@ async function processCreate (message, transactionId) {
       index: topResources[resource].index,
       type: topResources[resource].type,
       id: message.payload.id,
-      body: _.omit(message.payload, 'resource'),
-      refresh: 'true'
+      body: _.omit(message.payload, ['resource', 'originalTopic']),
+      refresh: 'wait_for'
     })
   } else if (_.includes(_.keys(userResources), resource)) {
     // process user resources such as userSkill, userAttribute...
@@ -45,7 +46,7 @@ async function processCreate (message, transactionId) {
       logger.error(`Can't create existed ${resource} with the ${userResource.relateKey}: ${relateId}, userId: ${message.payload.userId}`)
       throw helper.getErrorWithStatus('[version_conflict_engine_exception]', 409)
     } else {
-      user[userResource.propertyName].push(_.omit(message.payload, 'resource'))
+      user[userResource.propertyName].push(_.omit(message.payload, ['resource', 'originalTopic']))
       await helper.updateUser(message.payload.userId, user, seqNo, primaryTerm, transactionId)
     }
   } else if (_.includes(_.keys(organizationResources), resource)) {
@@ -63,7 +64,7 @@ async function processCreate (message, transactionId) {
       logger.error(`Can't create existing ${resource} with the ${orgResources.relateKey}: ${relateId}, organizationId: ${message.payload.organizationId}`)
       throw helper.getErrorWithStatus('[version_conflict_engine_exception]', 409)
     } else {
-      org[orgResources.propertyName].push(_.omit(message.payload, 'resource'))
+      org[orgResources.propertyName].push(_.omit(message.payload, ['resource', 'originalTopic']))
       await helper.updateOrg(message.payload.organizationId, org, seqNo, primaryTerm, transactionId)
     }
   } else {
@@ -78,7 +79,8 @@ processCreate.schema = {
     timestamp: Joi.date().required(),
     'mime-type': Joi.string().required(),
     payload: Joi.object().keys({
-      resource: Joi.string().required()
+      resource: Joi.string().required(),
+      originalTopic: Joi.string().required().valid(config.UBAHN_CREATE_TOPIC)
     }).required().unknown(true)
   }).required(),
   transactionId: Joi.string().required()
@@ -105,10 +107,11 @@ async function processUpdate (message, transactionId) {
       id,
       transactionId,
       body: {
-        doc: _.assign(source._source, _.omit(message.payload, 'resource'))
+        doc: _.assign(source._source, _.omit(message.payload, ['resource', 'originalTopic']))
       },
       if_seq_no: source._seq_no,
-      if_primary_term: source._primary_term
+      if_primary_term: source._primary_term,
+      refresh: 'wait_for'
     })
   } else if (_.includes(_.keys(userResources), resource)) {
     // process user resources such as userSkill, userAttribute...
@@ -127,7 +130,7 @@ async function processUpdate (message, transactionId) {
       throw helper.getErrorWithStatus('[resource_not_found_exception]', 404)
     } else {
       const updateIndex = _.findIndex(user[userResource.propertyName], [userResource.relateKey, relateId])
-      user[userResource.propertyName].splice(updateIndex, 1, _.omit(message.payload, 'resource'))
+      user[userResource.propertyName].splice(updateIndex, 1, _.omit(message.payload, ['resource', 'originalTopic']))
       logger.info(`Updating ${user.id} and ${relateId}`)
       await helper.updateUser(message.payload.userId, user, seqNo, primaryTerm, transactionId)
       logger.info(`Updated ${user.id} and ${relateId}`)
@@ -146,7 +149,7 @@ async function processUpdate (message, transactionId) {
       throw helper.getErrorWithStatus('[resource_not_found_exception]', 404)
     } else {
       const updateIndex = _.findIndex(org[orgResource.propertyName], [orgResource.relateKey, relateId])
-      org[orgResource.propertyName].splice(updateIndex, 1, _.omit(message.payload, 'resource'))
+      org[orgResource.propertyName].splice(updateIndex, 1, _.omit(message.payload, ['resource', 'originalTopic']))
       await helper.updateOrg(message.payload.organizationId, org, seqNo, primaryTerm, transactionId)
     }
   } else {
@@ -161,7 +164,8 @@ processUpdate.schema = {
     timestamp: Joi.date().required(),
     'mime-type': Joi.string().required(),
     payload: Joi.object().keys({
-      resource: Joi.string().required()
+      resource: Joi.string().required(),
+      originalTopic: Joi.string().required().valid(config.UBAHN_UPDATE_TOPIC)
     }).required().unknown(true)
   }).required(),
   transactionId: Joi.string().required()
@@ -182,7 +186,7 @@ async function processDelete (message, transactionId) {
       index: topResources[resource].index,
       type: topResources[resource].type,
       id: message.payload.id,
-      refresh: 'true'
+      refresh: 'wait_for'
     })
   } else if (_.includes(_.keys(userResources), resource)) {
     // process user resources such as userSkill, userAttribute...
@@ -226,7 +230,8 @@ processDelete.schema = {
     timestamp: Joi.date().required(),
     'mime-type': Joi.string().required(),
     payload: Joi.object().keys({
-      resource: Joi.string().required()
+      resource: Joi.string().required(),
+      originalTopic: Joi.string().required().valid(config.UBAHN_DELETE_TOPIC)
     }).required().unknown(true)
   }).required(),
   transactionId: Joi.string().required()