#!/bin/bash

if [ "$1" = "-h" ] || [ "$#" -lt 1 ]; then
  echo "Usage: check_queue.sh [ -d domain ] jobtype"
  exit 0
fi

if [ "$1" = "-d" ]; then
  DOMAIN="$2"
  JOB_TYPE="$3"
else
  JOB_TYPE="$1"
fi

KAFKA_INFO="`kafka run-class  kafka.admin.ConsumerGroupCommand \
 --describe \
 --group change-prop-${JOB_TYPE} \
 --bootstrap-server localhost:9092 \
 --new-consumer | grep eqiad.mediawiki.job.${JOB_TYPE}`"

COMMITTED_OFFSET="$(echo ${KAFKA_INFO} | tr ',' '\n' | sed -n 4p | tr -d '[:space:]')"
BACKLOG="$(echo ${KAFKA_INFO} | tr ',' '\n' | sed -n 6p | tr -d '[:space:]')"

echo "Committed offset for ${JOB_TYPE}: ${COMMITTED_OFFSET}"
echo "Full backlog for ${JOB_TYPE}: ${BACKLOG}"

if [ "x$DOMAIN" = "x" ]; then
  echo "Top 10 domains by job backlog count:"
  kafkacat -b localhost:9092 \
    -p 0 \
    -t eqiad.mediawiki.job.${JOB_TYPE} \
    -o ${COMMITTED_OFFSET} \
    -c ${BACKLOG} 2> /dev/null | jq -c '.meta.domain' | sort | uniq -ic | sort -n -k 1 -r | head -n 10
else
  echo -n "Job count for ${DOMAIN}: "
  kafkacat -b localhost:9092 \
    -p 0 \
    -t eqiad.mediawiki.job.${JOB_TYPE} \
    -o ${COMMITTED_OFFSET} \
    -c ${BACKLOG} 2> /dev/null | jq -c "select(.meta.domain == \"${DOMAIN}\") | ." | wc -l
fi