cjmatta

A method for orchestrating distributed Kafka benchmarks using Ansible

Kafka benchmarks are typically run using a single producer and consumer against a single topic, and the producer and consumer are run at close to maximum write/read speeds. In the real world, a Kafka cluster is more often serving many lower throughput producers and consumers. Ansible allows for a benchmarking method that sets up any number of topics and many producers and consumers.

Ansible playbooks allow us to run a number of tasks against a distributed set of clients both synchronously and asynchronously.

Topic setup

Before we can run tests we need topics to test against. This play sets up a number of topics with various partition configurations:

- name : Setup

cjmatta

---
- hosts: all
  vars:
    certificate_email: [email protected]
    domain_name: my.domain.com
  become: yes
  tasks:
    - name: install certbot prereq
      yum:
        name: epel-release

cjmatta

---
- hosts: broker
  tasks:
    - name: create filesystem
      filesystem:
        fstype: ext4
        dev: /dev/xvdf
    - name: create /var/lib/kafka directory
      file:
        path: /var/lib/kafka

cjmatta

#!/usr/bin/env python
# This script is meant to be used in conjunction with the JSON formatted
# output of `https://github.com/cjmatta/cp-poc-terraform
# Usage: terraform output -json` | ./create_ansible_inventory.py
import json
from jinja2 import Template
import sys

template = Template("""all:
  vars:

cjmatta

Configuration for Connect worker(s)

Set these settings in your Kafka Connect worker config (Confluent Doc for this here: https://docs.confluent.io/current/kafka/authentication_sasl_plain.html#kafka-connect)

# Connect worker
security.protocol=SASL_SSL
ssl.truststore.location=/var/ssl/private/kafka.client.truststore.jks
ssl.truststore.password=test1234
sasl.mechanism=PLAIN

cjmatta

{
  "name": "wikipedia-irc",
  "config": {
      "producer.interceptor.classes": "io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor",
      "connector.class": "org.cmatta.kafka.connect.irc.IrcSourceConnector",
      "irc.server": "irc.wikimedia.org",
      "kafka.topic": "wikipedia.raw",
      "irc.channels": "#en.wikipedia,#en.wiktionary,#en.wikibooks,#en.wikinews,#es.wikipedia,#fr.wikipedia",
      "tasks.max": "2"
  }

cjmatta

Complete Confluent Platform docker-compose.yml file. Includes an nginx configuration to load-balance between the rest-proxy and schema-registry components.

Running - make sure both docker-compose.yml and nginx_kafka.conf are in the same directory:

$ docker-compose up

cjmatta

[
  {"name": "id", "class": "id"},
  {"name": "name", "class": "name", "type": "first_last"},
  {"name": "address", "class": "address"},
  {"name":"gender", "class":"string", "dist":{"MALE":0.5, "FEMALE":0.5, "OTHER":0.02}},
  {"name": "zip", "class": "zip"},
  {"name": "ssn", "class": "ssn"}

]

cjmatta

#!/bin/bash

USERNAME=
PASSWORD=

DRILL_VER=drill-1.4.0
DRILL_LOC=/opt/mapr/drill
URL=jdbc:drill:

DPROP=~/prop$$

cjmatta

package streamflow.spout.core;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BoltDeclarer;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;