export KAFKA09=$MYSW/kafka

function kafka-stop {
  $KAFKA09/bin/zookeeper-server-stop.sh $KAFKA09/config/zookeeper.properties
  sleep 3
  $KAFKA09/bin/kafka-server-stop.sh $KAFKA09/config/server.properties
}

function kafka-start {
  rm -rf /tmp/zookeeper /tmp/kafka-logs
  $KAFKA09/bin/zookeeper-server-start.sh $KAFKA09/config/zookeeper.properties &
  sleep 2
  $KAFKA09/bin/kafka-server-start.sh $KAFKA09/config/server.properties &
  sleep 1
}

function kafka-topic-produce-zombie-raw {
  kafka-topic-produce _test_events-1ebef931-8f30-49c9-9681-c5b22f5382d3
}
function kafka-topic-produce-zombie-sch {
  kafka-topic-produce _test_sch_events-1ebef931-8f30-49c9-9681-c5b22f5382d3
}

function kafka-topic-produce {
  TOPIC="${1-}"
  HOST="${2-localhost}"
  PORT="${3-9092}"

  if [ "$TOPIC" = "" ]; then exit 1; fi
  
  $KAFKA09/bin/kafka-console-producer.sh --topic $TOPIC --broker-list $HOST:$PORT
}

function kafka-topic-consume-zombie-raw {
  kafka-topic-consume _test_events-1ebef931-8f30-49c9-9681-c5b22f5382d3
}
function kafka-topic-consume-zombie-sch {
  kafka-topic-consume _test_sch_events-1ebef931-8f30-49c9-9681-c5b22f5382d3
}
function kafka-topic-consume {
  TOPIC="${1-}"
  HOST="${2-localhost}"
  PORT="${3-9092}"

  if [ "$TOPIC" = "" ]; then exit 1; fi
  
  $KAFKA09/bin/kafka-topics.sh --zookeeper $HOST:$PORT --describe --topic $TOPIC
}

function kafka-topic-create-zombie-raw {
  kafka-topic-create _test_events-1ebef931-8f30-49c9-9681-c5b22f5382d3
}
function kafka-topic-create-zombie-sch {
  kafka-topic-create _test_sch_events-1ebef931-8f30-49c9-9681-c5b22f5382d3
}
function kafka-topic-create {
  TOPIC="${1-}"
  HOST="${2-localhost}"
  PORT="${3-2181}"

  if [ "$TOPIC" = "" ]; then exit 1; fi

  $KAFKA09/bin/kafka-topics.sh --create --zookeeper $HOST:$PORT --replication-factor 1 --partitions 1 --topic $TOPIC
}