kafka admin scripts

reassign-partitions.sh

#!/usr/bin/env bash

set -e
BINDIR=/opt/kafka/kafka_2.11-0.10.0.1/bin/
TOPICS=${@:1}

if [[ "$TOPICS" == "" ]]; then
	echo "Usage: $0 topic1 topic2 ..."
	exit 1
fi

ZOOKEEPER=$(grep zookeeper.connect= /opt/kafka/config/server.properties  | cut -d'=' -f2 | cut -d '/' -f1)
echo "=> using zookeeper:"
echo $ZOOKEEPER

CMD="$BINDIR/kafka-reassign-partitions.sh"

BROKERS=$(kafkat brokers | tail -n +2  | awk '{print $1}' | tr '\n' ',' | sed -e 's/,$//')
echo "=> using kafka brokers:"
echo $BROKERS

JSON=/tmp/topics-to-move.json
cat > $JSON <<< '{"topics":['
for t in $TOPICS; do
	echo -n "{\"topic\":\"$t\"}," >> $JSON
done
echo -n '],"version":1}' >> $JSON
echo >> $JSON

sed -i.bak 's/,]/]/' $JSON
echo "=> $JSON"
cat $JSON | jq .

# Generate assignments
echo "=> generating assignments"
$CMD --zookeeper $ZOOKEEPER --generate --broker-list $BROKERS --topics-to-move-json-file $JSON > /tmp/assignments.json
< /tmp/assignments.json tail -1 > /tmp/new-assignments.json
< /tmp/assignments.json head -3 | tail -1 > /tmp/old-assignments.json

echo "=> new assignments (/tmp/new-assignments.json)"
cat /tmp/new-assignments.json | sed -e 's/{"topic"/\n{"topic"/g' | column -t -s , | sort -n

echo -n "=> execute assignment? [y] "
read answer
if echo "$answer" | grep -iq "^y" ;then
	$CMD --zookeeper $ZOOKEEPER --execute --reassignment-json-file /tmp/new-assignments.json
	echo '=> rollback command:'
	echo $CMD --zookeeper $ZOOKEEPER --execute --reassignment-json-file /tmp/old-assignments.json
	exit 0
else
	echo "=> cancelled"
	exit 1
fi