This add-on is operated by 84codes AB

Message streaming as a Service

CloudKarafka

Last updated 29 May 2017

CloudKarafka is an add-on that provides Apache Kafka as a service. Apache Kafka is a message bus optimized for high-ingress data streams and replay written in Scala and Java.

Installing the add-on

CloudKarafka can be installed to a Heroku application via the CLI:

$ heroku addons:create cloudkarafka
-----> Adding cloudkarafka to sharp-mountain-4005... done, v18 (free)

Example for the “Mini Mouse” plan:

$ heroku addons:create cloudkarafka:mouse

After you provision CloudKarafka, the CLOUDKARAFKA_BROKERS config var is available in your app’s configuration. It contains the canonical URL used to access the Apache Kafka cluster. You can confirm this with the heroku config command:

$ heroku config | grep CLOUDKARAFKA_BROKERS
CLOUDKAFKA_BROKERS    => "dogsled-01.srvs.cloudkarafka.com:9093,dogsled-02.srvs.cloudkarafka.com:9093,dogsled-03.srvs.cloudkarafka.com:9093"

Also available in the Heroku config are the following variables:

After you provision CloudKarafka, your application should be configured to fully integrate with the add-on.

About CloudKarafka

CloudKarafka provides managed Apache Kafka servers in the cloud. Kafka is a distributed publish-subscribe messaging system that is designed to be fast, scalable, and durable. It’s an open-source message broker written in Scala that can support a large number of consumers and retain large amounts of data with very little overhead.

CloudKarafka automates every part of setup, running, and scaling of Apache Kafka. We have support staff available 24/7 to help in any event or with any questions. CloudKarafka lets developers focus on the core part of their applications, instead of managing and maintaining servers.

We offer five different plans for different needs. You can try CloudKarafka for free with the Developer Duck plan.

All plans are billed by the second, meaning you can try out even the largest instance types without spending much. As soon as you delete your instance, you won’t be charged for it anymore. Billing occurs at the end of each month, and you are only charged for the time an instance was active.

Instance details and console

Your Kafka instance is immediately created after provisioning the add-on. You can view your instance’s details from the Heroku CLI like so:

$ heroku addons:open cloudkarafka

The dashboard contains useful information about your instance, along with configuration options for Kafka topics, certificates, and more. The following sections describe features of the CloudKarafka dashboard.

Add and delete topics

When you create your instance, a default Kafka topic is created for you. You can add up to 5 topics to your free instance. Add new topics to your instance from the Topics tab of your CloudKarafka dashboard.

Cert management

We allow connections via certificate-based authentication. You can find information about how to set up the certificates in the Certificate tab of your CloudKarafka dashboard.

After you complete all steps in the guide, you will have the following credentials that will allow you to connect:

• A private key
• A CA certificate
• A signed client certificate

Language-specific code can be found in the documentation menu.

Log

This dashboard tab shows the live log from Kafka.

Metrics

The Metrics tab helps you to measure performance metrics from your server. CloudKarafka shows monitoring for CPU Usage, Memory Usage, and Disk Usage.

Kafka Manager

Kafka Manager is a tool for management of Apache Kafka. It is enabled for dedicated instances by default.

Shared instances (Developer Duck plan)

Instances on the Developer Duck plan reside on a multi-tenant Kafka server. You share this server with other users. To get started with your shared instance, simply use the CLOUDKARAFKA_BROKERS config var that was set when you installed the add-on.

In the language-specific documentation below, you will see examples of how to handle the config vars (certificates) in different languages. Some languages require a path to the cert file, in which case you first need to write the .env certificate information to a file. Other languages can read and handle the certificates straight from the environment variables. See the language-specific documentation for more information.

Dedicated instances

To use a dedicated Kafka instance, you must establish a secure connection between your clients and your CloudKarafka server. You need to create certificates, or you need to peer your instance with your AWS VPC.

Ruby

Ruby-kafka allows you to produce and consume messages to/from the Apache Kafka distributed messaging service, and it is the recommended library for Ruby. Ruby-kafka support the new consumers and producers introduced in Apache Kafka 0.9. It also supports the new certificate based authentication, which is required for the shared plans on CloudKarafka.

Add ruby-kafka as a dependency to your Gemfile and execute bundle install.

First thing that you have to do is connect to the Kafka server. You can get all the connection variables you need from Heroku to create the CloudKarafka instance.

The following code snippet demonstrates how to publish and subscribe:

#Producer.rb
require 'kafka'

# Create an array with the broker host names.
brokers = ENV['CLOUDKAFKA_BROKERS'].split(',')

# Create a Apache Kafka client.
kafka = Kafka.new(seed_brokers: brokers,
                  ssl_ca_cert: ENV['CLOUDKAFKA_CA'],
                  ssl_client_cert: ENV['CLOUDKAFKA_CERT'],
                  ssl_client_cert_key: ENV['CLOUDKAFKA_PRIVATE_KEY'])

producer = kafka.producer

topic_prefix = ENV['CLOUDKAFKA_TOPIC_PREFIX']
loop do
  producer.produce("hello", topic: "#{topic_prefix}default")

  # If this line fails with Kafka::DeliveryFailed we *may* have succeeded in delivering
  # the message to Kafka but won't know for sure.
  producer.deliver_messages
  # If we get to this line we can be sure that the message has been delivered to Kafka!
  sleep 1
end

#Consumer.rb
require 'kafka'

# Create an array with the broker host names.
brokers = ENV['CLOUDKAFKA_BROKERS'].split(',')

# Create a Apache Kafka client.
kafka = Kafka.new(seed_brokers: brokers,
                  ssl_ca_cert: ENV['CLOUDKAFKA_CA'],
                  ssl_client_cert: ENV['CLOUDKAFKA_CERT'],
                  ssl_client_cert_key: ENV['CLOUDKAFKA_PRIVATE_KEY'])

consumer = kafka.consumer(group_id: "my-consumer")

topic_prefix = ENV['CLOUDKAFKA_TOPIC_PREFIX']
consumer.subscribe("#{topic_prefix}default")

# This will loop indefinitely, yielding each message in turn.
consumer.each_message do |message|
  puts message.topic, message.partition
  puts message.offset, message.key, message.value
end

Python

The Apache Kafka Python client can be installed as follows:

$ pip install kafka-python

First, connect to the Kafka server. You can get all the connection variables you need from Heroku when you create your CloudKarafka instance. The following code snippet demonstrates how to publish and consume messages:

# Producer.py

from kafka import KafkaProducer
from kafka.errors import KafkaError
import os, ssl

# First get all info from the env
ca = os.environ.get('CLOUDKAFKA_CA')
cert = os.environ.get('CLOUDKAFKA_CERT')
key = os.environ.get('CLOUDKAFKA_PRIVATE_KEY')
with open("/tmp/ca.pem", "w") as f:
    f.write(ca)
with open("/tmp/cert.pem", "w") as f:
    f.write(cert)
with open("/tmp/key.pem", "w") as f:
    f.write(key)

ssl_context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)

ssl_context.verify_mode = ssl.CERT_REQUIRED
ssl_context.check_hostname = True
ssl_context.load_verify_locations("/tmp/ca.pem")
ssl_context.load_cert_chain('/tmp/cert.pem', '/tmp/key.pem')

# Load the rest of the env variables
brokers = os.environ.get('CLOUDKAFKA_BROKERS').split(',')

Go

An excellent client for Apache Kafka written in Go is Sarama

First, connect to the Kafka server. You can get all the connection variables you need from Heroku when you create the CloudKarafka instance.

In the following code snippet you can see how you can publish messages:

//producer.go

package main

import (
    "gopkg.in/Shopify/sarama.v1"

    "crypto/tls"
    "crypto/x509"
    "fmt"
    "os"
    "strings"
)

func main() {
    brokers := os.Getenv("CLOUDKAFKA_BROKERS")
    ca := os.Getenv("CLOUDKAFKA_CA")
    cert := os.Getenv("CLOUDKAFKA_CERT")
    key := os.Getenv("CLOUDKAFKA_PRIVATE_KEY")
    topicPrefix := os.Getenv("CLOUDKAFKA_TOPIC_PREFIX")

    brokerList := strings.Split(brokers, ",")

    c, err := tls.X509KeyPair([]byte(cert), []byte(key))
    if err != nil {
        fmt.Println(err)
        return
    }

    caCertPool := x509.NewCertPool()
    caCertPool.AppendCertsFromPEM([]byte(ca))

    tlsConfig := &tls.Config{
        Certificates:       []tls.Certificate{c},
        RootCAs:            caCertPool,
        InsecureSkipVerify: false,
    }

    config := sarama.NewConfig()
    // Wait for all in-sync replicas to ack the message
    config.Producer.RequiredAcks = sarama.WaitForAll
    config.Producer.Retry.Max = 10 // Retry up to 10 times to produce the message

    config.Net.TLS.Config = tlsConfig
    config.Net.TLS.Enable = true

    producer, err := sarama.NewSyncProducer(brokerList, config)
    if err != nil {
        fmt.Println(err)
        return
    }
    defer func() {
        if err := producer.Close(); err != nil {
            fmt.Println(err)
            return
        }
    }()
    topic := topicPrefix + "default"
    msg := &sarama.ProducerMessage{Topic: topic, Value: sarama.StringEncoder("testing 123")}
    partition, offset, err := producer.SendMessage(msg)
    if err != nil {
        fmt.Printf("FAILED to send message: %s\n", err)
    } else {
        fmt.Printf("> message sent to partition %d at offset %d\n", partition, offset)
    }
}

package main

import (
    "gopkg.in/Shopify/sarama.v1"

    "crypto/tls"
    "crypto/x509"
    "fmt"
    "os"
    "os/signal"
    "strings"
)

func main() {
    brokers := os.Getenv("CLOUDKAFKA_BROKERS")
    ca := os.Getenv("CLOUDKAFKA_CA")
    cert := os.Getenv("CLOUDKAFKA_CERT")
    key := os.Getenv("CLOUDKAFKA_PRIVATE_KEY")
    topicPrefix := os.Getenv("CLOUDKAFKA_TOPIC_PREFIX")

    brokerList := strings.Split(brokers, ",")

    c, err := tls.X509KeyPair([]byte(cert), []byte(key))
    if err != nil {
        panic(err)
    }

    caCertPool := x509.NewCertPool()
    caCertPool.AppendCertsFromPEM([]byte(ca))

    tlsConfig := &tls.Config{
        Certificates:       []tls.Certificate{c},
        RootCAs:            caCertPool,
        InsecureSkipVerify: false,
    }

    config := sarama.NewConfig()

    config.Net.TLS.Config = tlsConfig
    config.Net.TLS.Enable = true

    client, err := sarama.NewClient(brokerList, config)
    if err != nil {
        panic(err)
    }
    fmt.Println("New Client")

    consumer, err := sarama.NewConsumer(brokerList, config)
    if err != nil {
        panic(err)
    }
    fmt.Println("New consumer")

    defer func() {
        if err := consumer.Close(); err != nil {
            panic(err)
        }
    }()
    topic := topicPrefix + "default"

    partitions, err := client.Partitions(topic)
    if err != nil {
        panic(err)
    }

    messages := make(chan *sarama.ConsumerMessage)
    for _, part := range partitions {
        go consumePartition(consumer, topic, part, messages)
    }

    signals := make(chan os.Signal, 1)
    signal.Notify(signals, os.Interrupt)

    consumed := 0
ConsumerLoop:
    for {
        select {
        case msg := <-messages:
            fmt.Printf("Consumed message offset %d\n", msg.Offset)
            consumed++
        case <-signals:
            fmt.Println("break")
            break ConsumerLoop
        }
    }

    fmt.Printf("Consumed: %d\n", consumed)
}

func consumePartition(consumer sarama.Consumer, topic string, partition int32, out chan *sarama.ConsumerMessage) {
    partitionConsumer, err := consumer.ConsumePartition(topic, partition, sarama.OffsetOldest)
    if err != nil {
        panic(err)
    }

    defer func() {
        if err := partitionConsumer.Close(); err != nil {
            panic(err)
        }
    }()

    for msg := range partitionConsumer.Messages() {
        out <- msg
    }
}

Java

To get started with Apache Kafka, add the kafka-client dependency to your project’s pom.xml file:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.9.0.1</version>
</dependency>

Because the certs we provide are PEM-encoded, we need to convert them to a pcs12 keystore using OpenSSL.

//TestConsumer.java

import org.apache.kafka.clients.consumer.*;
import java.util.Arrays;
import java.util.Properties;
import java.lang.System;
import java.io.*;
import java.lang.Runtime;
import java.io.PrintWriter;

public class TestConsumer {
    public static void main(String[] args) {
        try {
            PrintWriter caWriter = new PrintWriter("/tmp/cert.pem", "UTF-8");
            String ca = System.getenv("CLOUDKAFKA_CA");
            caWriter.println(ca);
            caWriter.close();

            PrintWriter certWriter = new PrintWriter("/tmp/cert.pem", "UTF-8");
            String cert = System.getenv("CLOUDKAFKA_CERT");
            certWriter.println(cert);
            certWriter.close();

            PrintWriter keyWriter = new PrintWriter("/tmp/key.pem", "UTF-8");
            String privateKey = System.getenv("CLOUDKAFKA_PRIVATE_KEY");
            keyWriter.println(privateKey);
            keyWriter.close();

            Runtime r = Runtime.getRuntime();
            Process p = r.exec("openssl pkcs12 -export -password pass:test1234 -out /tmp/store.pkcs12 -inkey /tmp/key.pem -certfile /tmp/ca.pem -in /tmp/cert.pem -caname 'CA Root' -name client");
            p.waitFor();

            p = r.exec("keytool -importkeystore -noprompt -srckeystore /tmp/store.pkcs12 -destkeystore /tmp/keystore.jks -srcstoretype pkcs12 -srcstorepass test1234 -srckeypass test1234 -destkeypass test1234 -deststorepass test1234 -alias client");
            p.waitFor();

            p = r.exec("keytool -noprompt -keystore /tmp/truststore.jks -alias CARoot -import -file /tmp/ca.pem -storepass test1234");
            p.waitFor();

            String brokers = System.getenv("CLOUDKAFKA_BROKERS");
            String topicPrefix = System.getenv("CLOUDKAFKA_TOPIC_PREFIX");

            Properties props = new Properties();
            props.put("bootstrap.servers", brokers);
            props.put("group.id", "test");
            props.put("enable.auto.commit", "true");
            props.put("auto.commit.interval.ms", "1000");
            props.put("session.timeout.ms", "30000");
            props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            props.put("security.protocol", "SSL");
            props.put("ssl.truststore.location", "/tmp/truststore.jks");
            props.put("ssl.truststore.password", "test1234");
            props.put("ssl.keystore.location", "/tmp/keystore.jks");
            props.put("ssl.keystore.password", "test1234");
            props.put("ssl.keypassword", "test1234");

            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
            consumer.subscribe(Arrays.asList(topicPrefix + "default"));
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records)
                    System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value());
            }
        } catch (FileNotFoundException e) {
            System.out.println(e);
        } catch (UnsupportedEncodingException e) {
            System.out.println(e);
        } catch (IOException e) {
            System.out.println(e);
        } catch (InterruptedException e) {
            System.out.println(e);
        }
    }
}

//TestProducer.java

import org.apache.kafka.clients.producer.*;
import java.util.Arrays;
import java.util.Properties;
import java.lang.System;
import java.io.*;
import java.lang.Runtime;
import java.io.PrintWriter;

public class TestProducer {
    public static void main(String[] args) {
        try {
            PrintWriter caWriter = new PrintWriter("/tmp/cert.pem", "UTF-8");
            String ca = System.getenv("CLOUDKAFKA_CA");
            caWriter.println(ca);
            caWriter.close();

            PrintWriter certWriter = new PrintWriter("/tmp/cert.pem", "UTF-8");
            String cert = System.getenv("CLOUDKAFKA_CERT");
            certWriter.println(cert);
            certWriter.close();

            PrintWriter keyWriter = new PrintWriter("/tmp/key.pem", "UTF-8");
            String privateKey = System.getenv("CLOUDKAFKA_PRIVATE_KEY");
            keyWriter.println(privateKey);
            keyWriter.close();

            Runtime r = Runtime.getRuntime();
            Process p = r.exec("openssl pkcs12 -export -password pass:test1234 -out /tmp/store.pkcs12 -inkey /tmp/key.pem -certfile /tmp/ca.pem -in /tmp/cert.pem -caname 'CA Root' -name client");
            p.waitFor();

            p = r.exec("keytool -importkeystore -noprompt -srckeystore /tmp/store.pkcs12 -destkeystore /tmp/keystore.jks -srcstoretype pkcs12 -srcstorepass test1234 -srckeypass test1234 -destkeypass test1234 -deststorepass test1234 -alias client");
            p.waitFor();

            p = r.exec("keytool -noprompt -keystore /tmp/truststore.jks -alias CARoot -import -file /tmp/ca.pem -storepass test1234");
            p.waitFor();

            String brokers = System.getenv("CLOUDKAFKA_BROKERS");
            String topicPrefix = System.getenv("CLOUDKAFKA_TOPIC_PREFIX");

            Properties props = new Properties();
            props.put("bootstrap.servers", brokers);
            props.put("group.id", "test");
            props.put("enable.auto.commit", "true");
            props.put("auto.commit.interval.ms", "1000");
            props.put("session.timeout.ms", "30000");
            props.put("key.serializer", org.apache.kafka.common.serialization.StringSerializer.class.getName());
            props.put("value.serializer", org.apache.kafka.common.serialization.StringSerializer.class.getName());
            props.put("security.protocol", "SSL");
            props.put("ssl.truststore.location", "/tmp/truststore.jks");
            props.put("ssl.truststore.password", "test1234");
            props.put("ssl.keystore.location", "/tmp/keystore.jks");
            props.put("ssl.keystore.password", "test1234");
            props.put("ssl.keypassword", "test1234");

            Producer<String, String> producer = new KafkaProducer<>(props);
            for(int i = 0; i < 100; i++)
                producer.send(new ProducerRecord<String, String>(topicPrefix + "default", Integer.toString(i), Integer.toString(i)));

            producer.close();
        } catch (FileNotFoundException e) {
            System.out.println(e);
        } catch (UnsupportedEncodingException e) {
            System.out.println(e);
        } catch (IOException e) {
            System.out.println(e);
        } catch (InterruptedException e) {
            System.out.println(e);
        }
    }
}

Node.js

The best Kafka library for node.js right now is Blizzard’s node-rdkafka. You can install the node-rdkafka module by using npm: npm install node-rdkafka

# Consumer

var Kafka = require('node-rdkafka');

var kafkaConf = {
  'group.id': 'my-consumer',
  'metadata.broker.list': 'localhost:9092',
  'socket.keepalive.enable': true,
  'enable.auto.commit': true
};

var topics = ["my-topic"];
if (process.env.CLOUDKARAFKA_TOPIC_PREFIX) {
  topics = topics.map(function(t) { return process.env.CLOUDKARAFKA_TOPIC_PREFIX + t; });
  fs.writeFileSync("/tmp/kafka.ca", process.env.CLOUDKARAFKA_CA);
  fs.writeFileSync("/tmp/kafka.crt", process.env.CLOUDKARAFKA_CERT);
  fs.writeFileSync("/tmp/kafka.key", process.env.CLOUDKARAFKA_PRIVATE_KEY);
  kafkaConf["ssl.ca.location"] = "/tmp/kafka.ca";
  kafkaConf["ssl.certificate.location"] = "/tmp/kafka.crt";
  kafkaConf["ssl.key.location"] = "/tmp/kafka.key";
  kafkaConf["security.protocol"] = "ssl";
  kafkaConf["metadata.broker.list"] = process.env.CLOUDKARAFKA_BROKERS.split(",");
}

var consumer = new Kafka.KafkaConsumer(kafkaConf, {
  'auto.offset.reset': 'beginning'
});

consumer
  .on('ready', function() {
    consumer.subscribe(topics);
    consumer.consume();
  })
  .on('error', function(err) {
    console.log(err);
    process.exit(1);
  })
  .on('data', function(data) {
    var msg = JSON.parse(data.value);
    console.log(msg);
  });

consumer.connect();

Alternative client libraries include franz-kafka and a spin-off project from ordinary node-kafka: https://github.com/terrancesnyder/node-kafka

Removing the add-on

You can remove the add-on directly from the Heroku CLI:

$ heroku addon:destroy cloudkarafka --confirm <name of app>

Support

All CloudKarafka support and runtime issues should be logged with Heroku Support at support.heroku.com. Any non-support related issues or product feedback is welcome at support@cloudkarafka.com. For our plans Bouncing Bat and larger, we provide 24/7 critical support with a 30-minutes maximum initial response time. For the plans Fast Fox and larger we also provide direct phone numbers.