Apache Kafka Automation, Part 2: Kafka as a Service

What is Kafka

Initially conceived as a messaging queue, Kafka is based on an abstraction of a distributed commit log. Since being created and open sourced by LinkedIn in 2011, Kafka has quickly evolved from a messaging queue to a fully-fledged event streaming platform.

Kafka has three key capabilities:

• Publish and subscribe to streams of records, similar to a message queue or enterprise messaging system.
• Store streams of records in a fault-tolerant, durable way.
• Process streams of records as they occur.

Kafka is generally used for two broad classes of applications:

• Real-time data streaming pipelines that reliably move data between systems or applications.
• Real-time streaming applications that transform or react to continual flow of data.

What is Confluent

Founded by the original developers of Apache Kafka, Confluent provides the most complete distribution of Kafka with Confluent Platform. Confluent Platform improves Kafka with additional community and commercial features that enhance the streaming experience and support massive scales.

Challenges in the Cloud (AWS)

Provisioning a Kafka cluster on AWS can be as simple as installing the required Java and Confluent Platform packages. A more reliable approach is to use Chef and Packer to create an Amazon Machine Image (AMI) with Kafka and all dependencies installed. The challenge comes in configuring Kafka:

• How do we determine the broker id for each broker?
• How we handle broker replacement? On in other words, how we tell to the other brokers about new broker IP addresses?
• In the event of new broker initialization, how do we manage data replication?

I’ll address each of these questions, but first, I’ll make a few assumptions:

1. We have a cluster of 3 brokers.
2. Each broker is deployed on an EC2 instance in one Available Zone (AZ).
3. We are leveraging an Auto Scaling Group (ASG).
4. We are using Chef and Packer to bake Confluent Kafka and all its dependencies.
5. We are using Keystore for secret and configuration management.

When we deploy Kafka cluster (Confluent flavor) on AWS as a service, there is no need for manual installation or configuration. In the case of any broker going down or being replaced, configuration information (such as broker id assignment) is automated and no manual configuration is needed! How is this achieved?

Broker Assignment

Broker id is a mandatory configuration in Kafka. It can be string or number, but it needs to be unique in the cluster. How do we set this value?

I list two approaches:

1. Use the EC2 instance id: This is very simple to setup, but when a broker is replaced, replication may put a lot of traffic on the network. Another downside is that if all the brokers are terminated, we will lose our data. This option is viable for a large cluster.
2. Use a number, 0 to N: This requires custom code to query Zookeeper, determine what ids are available, and pick one. The script needs to be smart enough to handle multiple simultaneous requests and not return duplicate ids. The main benefit of this option is that we can predict the broker id and provision some resources like Elastic Block Store (EBS) and Elastic Network Interface (ENI) and apply them to the broker, as described in the next section. This option is good for small to medium-sized clusters. Here is some sample python that manages the broker assignment:

   from kazoo.client import KazooClient
   
   zk = KazooClient(hosts=zk_hosts, logger=logger)
   zk.start()
   zk_broker_ids = zk.get_children('/brokers/ids')
   zk.stop()
   zk.close()
   # our cluster size is 3 in our example
   max_broker_count = 3
   set_broker_ids = set(map(int, zk_broker_ids))
   possible_broker_ids = set(range(max_broker_count))
   broker_id = sorted(possible_broker_ids - set_broker_ids)[0]

For the rest of this article, I’ll use a small pool of broker ids: 0, 1 and 2.

ENI Attachment

Since we know the broker id, attaching ENI is easy, because we simply can map broker ids to the ENI tags:

1. id = 0 map to ENI tag KAFKA-0.
2. id = 1 map to ENI tag KAFKA-1.
3. id = 2 map to ENI tag KAFKA-2.

Before we go farther, let’s see why we need ENI. We need to create an environment with static internal IP addresses for the brokers, so that if a broker is replaced, the new one gets the same IP address. With ENI, we do this easily.

In our ASG launch configuration, we should follow this order to attach the ENI:

Step 1. Get broker id:

require 'zookeeper'
def get_broker_id(zk_host)
  zk = Zookeeper.new(zk_host)
  assigned_ids = zk.get_children(:path => '/brokers/ids')[:children]
  # our cluster size is 3 in our example
  max_broker_count = 3
  all_ids =* (0..(max_broker_count - 1)).map(&:to_s)
  possible_ids = all_ids - assigned_ids
  possible_ids.size == 0 ? -1 : possible_ids[0]
end

zk_host = '10.100.1.100:2181,10.100.2.100:2181,10.100.3.100:2181'

broker_id = get_broker_id(zk_host)
@ec2 = Aws::EC2::Client.new(region: region)
# get the available eni
eni = @ec2.describe_network_interfaces(
  filters: [
    {
       name: 'tag:Name', 
       values: ["KAFKA-#{broker_id}"]
    },
    {
       name: 'status',
       values: ['available']
    }
  ]).network_interfaces[0]

metadata_endpoint = 'http://169.254.169.254/latest/meta-data/'
instance_id = Net::HTTP.get(URI.parse(metadata_endpoint + 'instance-id'))
eni.attach(instance_id: instance_id, device_index: 1)

#!/bin/bash -e
export GATEWAY=`route -n | grep "^0.0.0.0" | tr -s " " | cut -f2 -d" "`

if [ -f /etc/network/interfaces.d/eth1.cfg ]; then mv -f /etc/network/interfaces.d/eth1.cfg /etc/network/interfaces.d/backup.eth1.cfg.backup; fi
cat > /etc/network/interfaces.d/eth1.cfg < lookup eth1_rt prio 1000
ETHCFG

mv /etc/iproute2/rt_tables /etc/iproute2/backup.rt_tables.backup
cat > /etc/iproute2/rt_tables <

from boto import ec2
import commands

broker_id = get_broker_id()
conn = ec2.connect_to_region(region_name)

volume = conn.get_all_volumes(
  filters = {
    'tag:Name': "KAFKA-%s" % broker_id
  })[0]
# attach the volume
conn.attach_volume(volume.id, instance_id, '/dev/xvdg')
# mount it
commands.getstatusoutput('mount /dev/xvdg /kafkalogs')

NetworkInterface1:
  Type: AWS::EC2::NetworkInterface
  Properties:
    SubnetId:
      Ref: Subnet1
    PrivateIpAddress: 10.100.1.200
    Description: ENI for Kafka broker 0
    GroupSet:
    - Ref: InstanceSecurityGroup
    Tags:
    - Key: Name
      Value: KAFKA-0

      Value: KAFKA-0

$ cfn-signal -e 0 --stack $stack_name --resource InstanceAsg --region us-east-1