common error topic, no aggregation needed during error publishing

config/default.js

@@ -35,9 +35,8 @@ module.exports = {
   KAFKA_MESSAGE_ORIGINATOR: process.env.KAFKA_MESSAGE_ORIGINATOR || 'u-bahn-api',
 
   // topics
-  UBAHN_CREATE_USER_TOPIC: process.env.UBAHN_CREATE_USER_TOPIC || 'user.action.topic.create',
-  UBAHN_UPDATE_USER_TOPIC: process.env.UBAHN_UPDATE_USER_TOPIC || 'user.action.topic.update',
-  UBAHN_DELETE_USER_TOPIC: process.env.UBAHN_DELETE_USER_TOPIC || 'user.action.topic.delete',
+  UBAHN_ERROR_TOPIC: process.env.UBAHN_ERROR_TOPIC || 'u-bahn.action.error',
+
   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',

src/common/helper.js

@@ -160,6 +160,25 @@ async function postEvent (topic, payload) {
   await busApiClient.postEvent(message)
 }
 
+/**
+ * Send error event to Kafka
+ * @params {String} topic the topic name
+ * @params {Object} payload the payload
+ * @params {String} action for which operation error occurred
+ */
+ async function publishError (topic, payload, action) {
+  logger.debug(`Publish error to Kafka topic ${topic}, ${JSON.stringify(payload, null, 2)}`)
+  _.set(payload, 'apiAction', action)
+  const message = {
+    topic,
+    originator: config.KAFKA_MESSAGE_ORIGINATOR,
+    timestamp: new Date().toISOString(),
+    'mime-type': 'application/json',
+    payload
+  }
+  await busApiClient.postEvent(message)
+}
+
 module.exports = {
   validProperties,
   getAuthUser,
@@ -168,5 +187,6 @@ module.exports = {
   injectSearchMeta,
   getControllerMethods,
   getSubControllerMethods,
-  postEvent
+  postEvent,
+  publishError
 }

src/modules/user/service.js

@@ -43,7 +43,7 @@ async function create (entity, auth) {
     return result
   } catch (e) {
     if (payload) {
-      helper.postEvent(config.UBAHN_CREATE_USER_TOPIC, payload)
+      helper.publishError(config.UBAHN_ERROR_TOPIC, payload, 'user.create')
     }
     throw e
   }
@@ -74,16 +74,14 @@ async function patch (id, entity, auth, params) {
     const result = await sequelize.transaction(async (t) => {
       const newEntity = await dbHelper.update(User, id, entity, auth, null, t)
       payload = newEntity.dataValues
-
       await serviceHelper.patchRecordInEs(resource, newEntity.dataValues, true)
-
       return newEntity
     })
 
     return result
   } catch (e) {
     if (payload) {
-        helper.postEvent(config.UBAHN_UPDATE_USER_TOPIC, payload)
+      helper.publishError(config.UBAHN_ERROR_TOPIC, payload, 'user.update')
     }
     throw e
   }
@@ -203,7 +201,7 @@ async function beginCascadeDelete (id, params) {
     })
 
   } catch (e) {
-    helper.postEvent(config.UBAHN_DELETE_USER_TOPIC, payload)
+    helper.publishError(config.UBAHN_ERROR_TOPIC, payload, 'user.delete')
     throw e
   }
 }