Example KafkaFilter Repo

io_chrisCatiagnani_kafkaSpraynozzle_KafkaThrottlingFilter.java

package io.chrisCatiagnani.kafkaSpraynozzle;
// Pretend this file is actually located in io/chrisCatiagnani/kafkaSpraynozzle/KafkaThrottlingFilter.java

import com.uber.kafkaSpraynozzle.KafkaFilter;
import org.apache.http.entity.ByteArrayEntity;

import java.util.Random;

/**
 * Filter which throttles incoming requests by dropping a percentage of the messages
 */
public class KafkaThrottlingFilter implements KafkaFilter {

    private double allowedPercentage;
    private Random random;

    /**
     * @param allowedPercentage - String between 0 and 1
     */
    public KafkaThrottlingFilter(String allowedPercentage) {
        this.allowedPercentage = Double.parseDouble(allowedPercentage);
        this.random = new Random();
    }

    @Override
    public boolean filter(ByteArrayEntity jsonEntity) {
        return random.nextDouble() < allowedPercentage;
    }

    @Override
    public String toString() {
        return "KafkaThrottlingFilter{allowedPercentage=" + allowedPercentage + '}';
    }
}

Makefile

JAVA=java
JC=javac
JAR=jar
SPRAYNOZZLEPATH=./kafka-spraynozzle
HTTPCOMPONENTSPATH="$(SPRAYNOZZLEPATH)/httpcomponents-client-4.3.6"
CLASSPATH="$(SPRAYNOZZLEPATH)/kafkaSpraynozzle.jar:$(HTTPCOMPONENTSPATH)/lib/commons-codec-1.6.jar:$(HTTPCOMPONENTSPATH)/lib/commons-logging-1.1.3.jar:$(HTTPCOMPONENTSPATH)/lib/fluent-hc-4.3.6.jar:$(HTTPCOMPONENTSPATH)/lib/httpclient-4.3.6.jar:$(HTTPCOMPONENTSPATH)/lib/httpclient-cache-4.3.6.jar:$(HTTPCOMPONENTSPATH)/lib/httpcore-4.3.3.jar:$(HTTPCOMPONENTSPATH)/lib/httpmime-4.3.6.jar"

.PHONY: download build clean build-spraynozzle

download: kafka-spraynozzle

build-spraynozzle: download
    cd kafka-spraynozzle && make build

build: build-spraynozzle
    $(JC) -Xlint:unchecked -cp $(CLASSPATH) io/chrisCatignani/kafkaSpraynozzle/*.java
    rm -rf build
    mkdir build
    cp io/chrisCatignani/kafkaSpraynozzle/*.class build/

clean:
    rm -rf kafka-spraynozzle
    rm -rf build

kafka-spraynozzle:
    git clone [email protected]/uber/kafka-spraynozzle

throttledKafkaSpraynozzle.sh

#!/bin/bash

set -vex

TOPIC=$1
URL=$2
FILTERARG=$3

PATH=/usr/lib/jvm/java-1.7.0-openjdk-amd64/bin:$PATH
export PATH

cd ./kafka-spraynozzle && ./kafka-spraynozzle.sh -c $PWD/../ -f io.chrisCatiagnani.kafkaSpraynozzle.KafkaThrottlingFilter -a $FILTERARG -u http://localhost:12345$URL -n 128 $TOPIC 1>$PWD/kafka-spraynozzle-$TOPIC.log 2>$PWD/kafka-spraynozzle-$TOPIC.err