Tag: kafka

kafka-node with Heroku Kafka

The other day I was building a demo for a customer around Heroku and Heroku Kafka. My language of choice these days is node.js so I needed a Kafka library for node.js and I settled on kafka-node. Now I needed to figure how to use the library using the environment variables provided by Heroku to access my Kafka cluster.

After having bound my Kafka add-on to the app I had 5 environment variables added to my app. The variables are well explained in the Heroku DevCenter but how to use them with kafka-node had me spend some time on it. The root cause seemed to be that the name in the certificate presented by the Kafka brokers did not match the name in the certificate provided to me by Heroku. Since the library uses the node.js TLS module under the covers the solution was to implement some of the verification login myself using the checkServerIdentity method. In the method I verify the cryptographic fingerprint of the root issuer certificate as provided by Heroku (KAFKA_TRUSTED_CERT) with that of the issuing certificate of the Kafka broker. I also had to remove the “kafka+ssl://” part from the Kafka broker URL’s. I do that using regular expressions.

Below is the code. YMMW.

const kafka = require("kafka-node");
const x509 = require('x509');
const Client = kafka.KafkaClient;

const kafkaHosts = process.env.KAFKA_URL.replace(/kafka\+ssl:\/\//gi, "");
const kafkaCert = x509.parseCert(process.env.KAFKA_TRUSTED_CERT);

const options = {
  "key": process.env.KAFKA_CLIENT_CERT_KEY,
  "cert": process.env.KAFKA_CLIENT_CERT,
  "ca": [process.env.KAFKA_TRUSTED_CERT],
  "checkServerIdentity": (host, cert) => {
    if (kafkaCert.fingerPrint === cert.issuerCertificate.fingerprint) return undefined;
    return Error('Not authentic')
  }
}

module.exports = {
    "client": () => {
        return new Client({
            "kafkaHost": kafkaHosts,
            "sslOptions": options
        });
    },
    "topic": `${process.env.KAFKA_PREFIX}safe-habour`
}