kafka_bashrc.sh

# Source global definitions
if [ -f /etc/bashrc ]; then
	. /etc/bashrc
fi

export KAFKA_URL='kafka.service.consul:9092'
export ZOOKEEPER_URL='zookeeper.service.consul:2181'
export SCHEMA_REGISTRY_URL='http://schema-registry.localservice'
export FC_CONNECT_URL='http://workers-fc-connect.service.consul'

count_messages() {
  kafka-console-consumer --bootstrap-server $KAFKA_URL --topic $1 --timeout-ms 500 --from-beginning 2>&1 >/dev/null | grep "Processed a total" | grep -o "[0-9]*"
}

start_topic_drain() {
  kafka-configs --zookeeper $ZOOKEEPER_URL --entity-type topics --entity-name $1 --alter --add-config retention.ms=1000
}

stop_topic_drain() {
  kafka-configs --zookeeper $ZOOKEEPER_URL --entity-type topics --entity-name $1 --alter --delete-config retention.ms
}

kc_connector_status() {
  curl $FC_CONNECT_URL/connectors/$1/status | jq .
}

kc_connector_config() {
  curl $FC_CONNECT_URL/connectors/$1 | jq .
}

pause_kc_connector() {
  curl -X PUT $FC_CONNECT_URL/connectors/$1/pause | jq .
}

restart_kc_connector() {
  curl -X POST $FC_CONNECT_URL/connectors/$1/restart | jq .
}

delete_kc_connector() {
  curl -X DELETE $FC_CONNECT_URL/connectors/$1 | jq .
}

docker_top() {
  CONTAINER_ID=$(sudo docker ps | grep $1 | cut -c 1-12 | head -n 1)

  if [ -z "$CONTAINER_ID" ]
  then
    >&2 echo "No container matching $1"
    return 1
  fi

  ROOT_PID=$(sudo docker inspect -f '{{ .State.Pid }}' $CONTAINER_ID)
  ps xao pid,pgid,rss,lstart,args | grep $ROOT_PID | grep -v grep
}

alias reset_streams='kafka-streams-application-reset --bootstrap-servers $KAFKA_URL --zookeeper $ZOOKEEPER_URL '

alias avro_consume='kafka-avro-console-consumer --bootstrap-server $KAFKA_URL --property schema.registry.url="$SCHEMA_REGISTRY_URL" --topic'
alias json_consume='kafka-console-consumer --bootstrap-server $KAFKA_URL --topic'
alias list_topics='kafka-topics --zookeeper $ZOOKEEPER_URL --list'
alias describe_topic='kafka-topics --zookeeper $ZOOKEEPER_URL --describe --topic'
alias topic_configs='kafka-configs --zookeeper $ZOOKEEPER_URL --entity-type topics --describe --entity-name'
alias alter_topic_configs='kafka-configs --zookeeper $ZOOKEEPER_URL --entity-type topics --alter --entity-name'
alias list_consumers='kafka-consumer-groups --bootstrap-server $KAFKA_URL --list'
alias describe_consumer='kafka-consumer-groups --bootstrap-server $KAFKA_URL --describe --group'
alias reset_consumer_offsets='kafka-consumer-groups --bootstrap-server $KAFKA_URL --reset-offsets --group '
alias list_old_consumers='kafka-consumer-groups --zookeeper $ZOOKEEPER_URL --list'
alias delete_topic='kafka-topics --zookeeper $ZOOKEEPER_URL --delete --topic '
alias create_topic='kafka-topics --zookeeper $ZOOKEEPER_URL --create --topic '

alias tail_logs='sudo tail -f /var/log/messages'

if [ "$SHOW_CUSTOM_COMMANDS" != "false" ]; then
  echo "Loaded custom functions"
  declare -F | cut -c 12- | grep -v "^_" | sed 's/^/* /'
  echo ""

  echo "Loaded custom aliases"
  grep -o "^alias [^=]*" .bashrc | cut -c 7- | sed 's/^/* /'
  echo ""
fi