Develope kafka rest api with nodejs to make it callable from CPI through sap cloud connector Part II

In previous blog Part I, we have prepared the on premise kafka envirement . Today we will try build nodejs Kafka rest api proxy and use cpi to produce and consume kafka message by deployed rest api .

We can follow the following steps :

Step 1 : deploy CPI iflow to consume kafka topic’s msessage

Step 2 : deploy nodejs rest api to produce message to Kafka and consume kafka message and forward the message to cpi .

code snippy :

Server.js

const express = require('express');
const bodyParser = require('body-parser');
const { Kafka, logLevel } = require('kafkajs');
const axios = require('axios');
const oauth = require('axios-oauth-client');
const cpiurl = 'https://1s4hcextension.it-cpi001-rt.cfapps.eu10.hana.ondemand.com/http/kafka/sender';

const getClientCredentials = oauth.client(axios.create(), {
    url: 'https://1s4hcextension.authentication.eu10.hana.ondemand.com/oauth/token',
    grant_type: 'client_credentials',
    client_id: client_id_from_cpi_runtime_service_key,
    client_secret: client_secret_from_cpi_runtime_service_key,
    scope: ''
  });

const app = express();

const kafka = new Kafka({
    clientId: 'my-app',
    requestTimeout: 25000,
    connectionTimeout: 30000,
    authenticationTimeout:30000,
    retry: {
      initialRetryTime: 3000,
      retries: 0
    },
    logLevel: logLevel.INFO,
    brokers: ['localhost:9092']
  });

app.use(express.text())
app.use(bodyParser.json());

app.listen(4004,  () => { console.log('===> Server started') })


// depoly api to  produce  message to kafka
app.post('/kafka/:topic', (req, res) => {
    const topicv = req.params.topic;
    console.log(topicv);
const kafka = new Kafka({
    clientId: 'my-app',
    requestTimeout: 25000,
    connectionTimeout: 30000,
    authenticationTimeout:30000,
    retry: {
      initialRetryTime: 3000,
      retries: 0
    },
    brokers: ['localhost:9092']
  })
  const producer = kafka.producer();
  const run = async () => {
  await producer.connect();
  await   producer.send({
    topic: topicv,
    messages: [
      { value: req.body.msg.toString() },
    ],
  });
  await producer.disconnect();

  res.status(200).send('message send');
}
run().catch((e)=>{
    console.log(e);
    res.status(500).send('error');
})   
})  ;


// consume kafka message and forward the message to CPI 
const consumer = kafka.consumer({ groupId: 'test-group' });
 
const run = async () => {
  // Consuming

  await consumer.connect()
  await consumer.subscribe({ topic: 'dblab01', fromBeginning: true })
 
  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {

      console.log({
        partition,
        offset: message.offset,
        value: message.value.toString(),
      });
      getClientCredentials().then((token1)=>{  
          // console.log(token1.access_token);    
          const config = {
            headers:{
                "Authorization": `Bearer ${token1.access_token}`,
              "Content-Type": "application/json"
            }
          };  
          const data ={
            msg: message.value.toString()
          };
        axios.post(cpiurl,data,config).then(res=>{
            console.log(res);
        }).catch(e=>{
            console.log(e);
        });
    
    }).catch(e=>{
            console.log(e);
        })

    },
  })
}
 
run().catch((e)=>{
    console.log(e);
})

package.json

{
  "name": "capkafka",
  "version": "1.0.0",
  "description": "capkafka",
  "main": "server.js",
  "dependencies": {
    "axios": "^0.27.2",
    "axios-oauth-client": "^1.4.4",
    "body-parser": "latest",
    "express": "^4.17.1",
    "kafkajs": "latest",
    "passport": "^0.4.0"
  },
  "scripts": {
    "test": "echo "Error: no test specified" && exit 1"
  },
  "author": "jacky liu",
  "license": "ISC"
}