Undo changes carried out for the issue of ES not updating all users and instead add mutex at top level

Author: callmekatootie

src/app.js

@@ -5,7 +5,6 @@
 global.Promise = require('bluebird')
 const config = require('config')
 const Kafka = require('no-kafka')
-const _ = require('lodash')
 const healthcheck = require('topcoder-healthcheck-dropin')
 const logger = require('./common/logger')
 const helper = require('./common/helper')
@@ -17,79 +16,62 @@ logger.info('Starting kafka consumer')
 // create consumer
 const consumer = new Kafka.GroupConsumer(helper.getKafkaOptions())
 
-let count = 0
 let mutex = new Mutex()
 
-async function getLatestCount () {
-  const release = await mutex.acquire()
-
-  try {
-    count = count + 1
-
-    return count
-  } finally {
-    release()
-  }
-}
-
 /*
  * Data handler linked with Kafka consumer
  * Whenever a new message is received by Kafka consumer,
  * this function will be invoked
  */
 const dataHandler = (messageSet, topic, partition) => Promise.each(messageSet, async (m) => {
+  const release = await mutex.acquire()
   const message = m.message.value.toString('utf8')
   logger.info(`Handle Kafka event message; Topic: ${topic}; Partition: ${partition}; Offset: ${
     m.offset}; Message: ${message}.`)
   let messageJSON
-  let messageCount = await getLatestCount()
 
-  logger.debug(`Current message count: ${messageCount}`)
   try {
     messageJSON = JSON.parse(message)
   } catch (e) {
     logger.error('Invalid message JSON.')
     logger.logFullError(e)
 
-    logger.debug(`Commiting offset after processing message with count ${messageCount}`)
-
     // commit the message and ignore it
     await consumer.commitOffset({ topic, partition, offset: m.offset })
     return
+  } finally {
+    release()
   }
 
   if (messageJSON.topic !== topic) {
     logger.error(`The message topic ${messageJSON.topic} doesn't match the Kafka topic ${topic}.`)
 
-    logger.debug(`Commiting offset after processing message with count ${messageCount}`)
-
     // commit the message and ignore it
     await consumer.commitOffset({ topic, partition, offset: m.offset })
+    release()
     return
   }
-  const transactionId = _.uniqueId('transaction_')
+
   try {
     switch (topic) {
       case config.UBAHN_CREATE_TOPIC:
-        await ProcessorService.processCreate(messageJSON, transactionId)
+        await ProcessorService.processCreate(messageJSON)
         break
       case config.UBAHN_UPDATE_TOPIC:
-        await ProcessorService.processUpdate(messageJSON, transactionId)
+        await ProcessorService.processUpdate(messageJSON)
         break
       case config.UBAHN_DELETE_TOPIC:
-        await ProcessorService.processDelete(messageJSON, transactionId)
+        await ProcessorService.processDelete(messageJSON)
         break
     }
 
-    logger.debug(`Successfully processed message with count ${messageCount}`)
+    logger.debug('Successfully processed message')
   } catch (err) {
     logger.logFullError(err)
   } finally {
-    helper.checkEsMutexRelease(transactionId)
-    logger.debug(`Commiting offset after processing message with count ${messageCount}`)
-
     // Commit offset regardless of error
     await consumer.commitOffset({ topic, partition, offset: m.offset })
+    release()
   }
 })
 

src/common/helper.js

@@ -7,17 +7,11 @@ const config = require('config')
 const elasticsearch = require('elasticsearch')
 const _ = require('lodash')
 const Joi = require('@hapi/joi')
-const { Mutex } = require('async-mutex')
-const logger = require('./logger')
 
 AWS.config.region = config.ES.AWS_REGION
 
 // Elasticsearch client
 let esClient
-let transactionId
-// Mutex to ensure that only one elasticsearch action is carried out at any given time
-const esClientMutex = new Mutex()
-const mutexReleaseMap = {}
 
 /**
  * Get Kafka options
@@ -57,31 +51,6 @@ async function getESClient () {
       host
     })
   }
-
-  // Patch the transport to enable mutex
-  esClient.transport.originalRequest = esClient.transport.request
-  esClient.transport.request = async (params) => {
-    const tId = _.get(params.query, 'transactionId')
-    params.query = _.omit(params.query, 'transactionId')
-    if (!tId || tId !== transactionId) {
-      const release = await esClientMutex.acquire()
-      mutexReleaseMap[tId || 'noTransaction'] = release
-      transactionId = tId
-    }
-    try {
-      return await esClient.transport.originalRequest(params)
-    } finally {
-      if (params.method !== 'GET' || !tId) {
-        const release = mutexReleaseMap[tId || 'noTransaction']
-        delete mutexReleaseMap[tId || 'noTransaction']
-        transactionId = undefined
-        if (release) {
-          release()
-        }
-      }
-    }
-  }
-
   return esClient
 }
 
@@ -102,66 +71,50 @@ function validProperties (payload, keys) {
 /**
  * Function to get user from es
  * @param {String} userId
- * @param {String} transactionId
  * @returns {Object} user
  */
-async function getUser (userId, transactionId) {
+async function getUser (userId) {
   const client = await getESClient()
-  const user = await client.get({ index: config.get('ES.USER_INDEX'), type: config.get('ES.USER_TYPE'), id: userId, transactionId })
-  return { seqNo: user._seq_no, primaryTerm: user._primary_term, user: user._source }
+  return client.getSource({ index: config.get('ES.USER_INDEX'), type: config.get('ES.USER_TYPE'), id: userId })
 }
 
 /**
  * Function to update es user
  * @param {String} userId
- * @param {Number} seqNo
- * @param {Number} primaryTerm
- * @param {String} transactionId
  * @param {Object} body
  */
-async function updateUser (userId, body, seqNo, primaryTerm, transactionId) {
+async function updateUser (userId, body) {
   const client = await getESClient()
   await client.update({
     index: config.get('ES.USER_INDEX'),
     type: config.get('ES.USER_TYPE'),
     id: userId,
-    transactionId,
-    body: { doc: body },
-    if_seq_no: seqNo,
-    if_primary_term: primaryTerm
+    body: { doc: body }
   })
 }
 
 /**
  * Function to get org from es
  * @param {String} organizationId
- * @param {String} transactionId
  * @returns {Object} organization
  */
-async function getOrg (organizationId, transactionId) {
+async function getOrg (organizationId) {
   const client = await getESClient()
-  const org = await client.get({ index: config.get('ES.ORGANIZATION_INDEX'), type: config.get('ES.ORGANIZATION_TYPE'), id: organizationId, transactionId })
-  return { seqNo: org._seq_no, primaryTerm: org._primary_term, org: org._source }
+  return client.getSource({ index: config.get('ES.ORGANIZATION_INDEX'), type: config.get('ES.ORGANIZATION_TYPE'), id: organizationId })
 }
 
 /**
  * Function to update es organization
  * @param {String} organizationId
- * @param {Number} seqNo
- * @param {Number} primaryTerm
- * @param {String} transactionId
  * @param {Object} body
  */
-async function updateOrg (organizationId, body, seqNo, primaryTerm, transactionId) {
+async function updateOrg (organizationId, body) {
   const client = await getESClient()
   await client.update({
     index: config.get('ES.ORGANIZATION_INDEX'),
     type: config.get('ES.ORGANIZATION_TYPE'),
     id: organizationId,
-    transactionId,
-    body: { doc: body },
-    if_seq_no: seqNo,
-    if_primary_term: primaryTerm
+    body: { doc: body }
   })
 }
 
@@ -177,21 +130,6 @@ function getErrorWithStatus (message, statusCode) {
   return error
 }
 
-/**
- * Ensure the esClient mutex is released
- * @param {String} tId transactionId
- */
-function checkEsMutexRelease (tId) {
-  if (tId === transactionId) {
-    const release = mutexReleaseMap[tId]
-    delete mutexReleaseMap[tId]
-    transactionId = undefined
-    if (release) {
-      release()
-    }
-  }
-}
-
 module.exports = {
   getKafkaOptions,
   getESClient,
@@ -200,6 +138,5 @@ module.exports = {
   updateUser,
   getOrg,
   updateOrg,
-  getErrorWithStatus,
-  checkEsMutexRelease
+  getErrorWithStatus
 }