Confluent Kafka scripts for OPS people

find-connector-by-topic.sh

#!/bin/bash

# do not use proxied connections as we connect localhost only
unset http_proxy

# path to JQ binary for parsing JSON in Bash
JQ="$HOME/bin/jq"

# set topic name from input
topic=$1

found=0
for connector in $( curl -sk http://localhost:8084/connectors/ | jq -r '.[]' );
do
  curl -s "http://localhost:8083/connectors/${connector}/config" | grep -i $topic &>/dev/null

  # if matching topic name found in the connector config
  # show the status and details
  if [[ $? -eq 0 ]];
  then
    echo -e "\n\n====== CHECK CONNECTOR STATUS =======\n\n"
    curl -s "http://localhost:8083/connectors/${connector}/status" | $JQ -r

    echo -e "\n\n====== SHOW CONNECTOR CONFIG =======\n\n"
    curl -s "http://localhost:8083/connectors/${connector}/config" | $JQ -r

    # increase the found var by 1
    ((found++))
    # break the loop
    #break
  fi
done


# if no matching connector config found
# display a message
[[ $found -eq 0  ]] && echo -e "\n\n**** CONNECTOR NOT FOUND ****\n\n"

list-of-helpful-kafka-commands.md

List of helpful Kafka commands

Please note some commands may be Confluent Kafka specific

update-license-key.md

How to update Confluent License key on a Kafka Source/Sink connector

Step 1

Before renewing the license strings on connectors it's RECOMMENDED to create a backup of JSON configurations

First, you may list all connectors doing

$ http_proxy= curl -s http://localhost:8083/connectors/ | jq
[
  "LQ.REDACTED",
  "LQ.REDACTED",
  "LQ.REDACTED",
  "LQ.REDACTED",
  .....

Second, you may create a backup for a particular connect doing

http_proxy= curl -s http://localhost:8083/connectors/CONNECTOR_NAME/config | jq -r '.' > CONNECTOR_NAME.json

Third, you may create a backup of all the connectors in bulk action (RECOMMENDED)

$ mkdir -p ~/backups/connectors_$(date +%F)
$ cd ~/backups/connectors_$(date +%F)
$ http_proxy= curl -s http://localhost:8083/connectors/ | jq -r '.[]' | while read connector; do http_proxy= curl -s http://localhost:8083/connectors/${connector}/config | jq -r '.' > "${connector}.json"; done

Step 2

Renew the license string per each Source and Sink connector using the Confluent REST API (Note alternatively you may use Confluent Control Centre for the task)

PATCH request is not allowed, see Connect REST Interface | Confluent Documentation so the below example would NOT work

# the below does not work
$ http_proxy= curl -sk -H 'Content-Type: application/json' -X PATCH -g -d '{"confluent.license": "REDACTEDREDACTED"}' http://localhost:8083/connectors/CONNECTOR_NAME/config | jq -r '.'
 
{
  "error_code": 405,
  "message": "HTTP 405 Method Not Allowed"
}

Instead (!) we have to issue PUT request. You may use the attached update-license-key.sh for that.

Step 3

Restart the Kafka Connect service

$ systemctl status confluent-kafka-connect.service
$ sudo systemctl restart confluent-kafka-connect.service ; sudo journalctl -u confluent-kafka-connect.service -f
$ systemctl status confluent*

TIP: After the restart, Check if the new license keys got saved properly for the affected connectors.

update-license-key.sh

#!/bin/bash

# Update license key on a Source/Sink connector

JQ="$HOME/bin/jq"

connector=$1
LICENSE="REDACTED"


update_license() {
  http_proxy= curl -s "http://localhost:8083/connectors/${connector}/config" | \
    $JQ -c --arg confluentLicense $LICENSE '."confluent.license" |= $confluentLicense'
}

# ------------- Apply the new license string ------------------

http_proxy= curl -sk -H 'Content-Type: application/json' \
  -X PUT \
  -g -d "$(update_license)" \
  "http://localhost:8083/connectors/${connector}/config" | $JQ -r '.'


# ------------ Check connector status ----------------------

echo -e "\n\n====== CHECK CONNECTOR STATUS =======\n\n"
http_proxy= curl -s "http://localhost:8083/connectors/${connector}/status" | $JQ