Kafka Scripts

ZOOKEEPER_CONNECT=zookeeper.service.consul:2181
LEDGER_TOPICS=(ledger-request-pre-release-2 ledger-result-pre-release-2 ledger-account-configuration-pre-release-1 ledger-account-pre-release-2 account-balance-updated-pre-release-3 ledger-entry-added-pre-release-2 __samza_checkpoint_ver_1_for_ledger-balance-tracker_1 __samza_checkpoint_ver_1_for_ledger-eventlog-builder_1 __samza_coordinator_ledger-balance-tracker_1 __samza_coordinator_ledger-eventlog-builder_1 ledger-account-configuration-changelog ledger-request-changelog ledger-result-changelog)

list-topics () {
    kafka-topics \
        --zookeeper $ZOOKEEPER_CONNECT \ 
        --list
  }

describe_topic () {
    kafka-topics \
        --zookeeper $ZOOKEEPER_CONNECT \
        --topic $1 \
        --describe
  }

delete_topic() {
    kafka-topics 
        --zookeeper $ZOOKEEPER_CONNECT \ 
        --topic $1 \ 
        --delete 
  }

set_retention() {
    kafka-configs \
        --zookeeper $ZOOKEEPER_CONNECT \ 
        --alter \
        --entity-name $1 \ 
        --entity-type topics \ 
        --add-config retention.ms=$2;
  }

reset_retention() {
    kafka-configs \
        --zookeeper $ZOOKEEPER_CONNECT \ 
        --alter \
        --entity-name $1 \
        --entity-type topics \ 
        --delete retention.ms;
  }

delete_ledger_topics() {
    for topic in $LEDGER_TOPICS; 
    do 
        delete_topic $topic; 
    done;
  }

drain_ledger_topics() {
    for topic in $LEDGER_TOPICS; 
    do 
        set_retention $topic 500; 
    done;
    sleep 600;
    for topic in $LEDGER_TOPICS; 
    do 
        reset_retention $topic; 
    done;
  }
  
consume () {
    kafka-console-consumer \
        --zookeeper $ZOOKEEPER_CONNECT \
        --property print.key=true \
        --topic $1 \
        --from-beginning; 
  }
  
ZOOKEEPER_CONNECT=zookeeper.service.consul:2181 avro-consume () { kafka-avro-console-consumer --zookeeper $ZOOKEEPER_CONNECT --property schema.registry.url=http://schema-registry.service.consul --topic $1 --from-beginning; }