Ekaanbj: /* Kafka Connect */

2025-08-26T12:43:00Z

Kafka Connect

New page

=What does it mean:=
akhq Apache Kafka HQ. web-based user interface and management tool for Apache Kafka clusters.
broker managing messages. Receive and deliver.
strimzi Apache Kafka on Kubernetes
cdc change data capture

=using kafka client=
==List topics==
kubectl exec -it -n kafka kafka-cluster-kafka-0 -- bin/kafka-topics.sh --bootstrap-server kafka-cluster-kafka-bootstrap:9092 --list
==Describe topic==
kubectl exec -it -n kafka $(oc get kafka --no-headers -o custom-columns=:.metadata.name)-kafka-0 -- bin/kafka-topics.sh --bootstrap-server $(oc get kafka -o json | jq -r '.items[0].status.listeners[].bootstrapServers' | grep 9092) --describe --topic <topic>
kubectl exec -it -n kafka $(oc get kafka --no-headers -o custom-columns=:.metadata.name)-kafka-0 -- bin/kafka-configs.sh --bootstrap-server $(oc get kafka -o json | jq -r '.items[0].status.listeners[].bootstrapServers' | grep 9092) --describe --entity-type topics --entity-name <topic>

=Read topic=
export TOPIC=<topic> ; kubectl exec -it -n kafka kafka-cluster-kafka-0 -- bin/kafka-console-consumer.sh --bootstrap-server kafka-cluster-kafka-bootstrap:9092 --topic ${TOPIC} --property print.key=true --max-messages 10 --from-beginning --timeout-ms 10000
==Describe all topics==
kubectl exec -i -n kafka kafka-cluster-kafka-0 -- bin/kafka-topics.sh --bootstrap-server kafka-cluster-kafka-bootstrap:9092 --list | while read i ; do echo '*' $i ; kubectl exec -i -n kafka kafka-cluster-kafka-0 -- bin/kafka-topics.sh --bootstrap-server kafka-cluster-kafka-bootstrap:9092 --describe --topic $i </dev/null ; done

=consumergroups=
List consumer groups
kubectl exec -it -n kafka kafka-cluster-kafka-0 -- bin/kafka-consumer-groups.sh --bootstrap-server kafka-cluster-kafka-bootstrap:9092 --list

=quick overview=
kubectl api-resources | grep -i kafka | awk '{print $1}' | while read i ; do echo '*' $i ; kubectl get $i -A ; done

=which users exit=
kubectl get kafkausers -n kafka

=which password does user have=
kubectl get secret ifs -o json | jq -r .data.password | base64 -d
=list topics=
kubectl get kafkatopics -o wide
=Kafka version=
kubectl exec -it -n kafka $(oc get kafka --no-headers -o custom-columns=:.metadata.name)-kafka-0 -- bin/kafka-topics.sh --bootstrap-server $(oc get kafka -o json | jq -r '.items[0].status.listeners[].bootstrapServers' | grep 9092) --version

=bootstrap server internally=
kubectl get kafka kafka-cluster -o=jsonpath='{.status.listeners[?(@.type=="tls")].bootstrapServers}{"\n"}'
=bootstrap sever external=
kubectl get kafka kafka-cluster -o=jsonpath='{.status.listeners[?(@.type=="external")].bootstrapServers}{"\n"}'
=produce to topic=
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic <topic>
Produce to topic with key.
kubectl exec -it -n kafka kafka-cluster-kafka-0 -- bash
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic <topic> --property "parse.key=true" --property "key.separator=:"
>emp_info: {"emp_id":100,"first_name":"Keshav","last_name":"Lodhi","designation":"DataEngineer"}
CTRL + c
kubectl exec -it -n kafka kafka-cluster-kafka-2 -- bin/kafka-console-consumer.sh --bootstrap-server kafka-cluster-kafka-bootstrap:9092 --topic customer-blacklistmail-dotnet-v1 --property print.key=true --max-messages 10 --from-beginning

=read topic=
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic FedoraTopic --from-beginning
=zookeeper=
Who is the leader.
for i in 0 1 2 ; do echo kafka-cluster-zookeeper-$i $(kubectl exec -it -n kafka kafka-cluster-zookeeper-$i -- bash -c 'echo stat | nc 127.0.0.1 12181' | grep Mode:) ; done
Get various info from zookeeper.
kubectl exec -it -n kafka kafka-cluster-zookeeper-0 -- bash -c 'echo envi | nc 127.0.0.1 12181'
Query all zookeepers
for i in 0 1 2 ; do echo '*' $i ; kubectl exec -it $(oc get kafka --no-headers -o custom-columns=:.metadata.name)-zookeeper-$i -n kafka -- bash -c 'echo conf | nc 127.0.0.1 12181' ; echo ; done
The following commands are availablew.
conf Print details about serving configuration.
cons List full connection/session details for all clients connected to this server. Includes information on numbers of packets
received/sent, session id, operation latencies, last operation performed, etc...
crst Reset connection/session statistics for all connections.
dump Lists the outstanding sessions and ephemeral nodes. This only works on the leader.
envi Print details about serving environment
ruok respond with imok if it is running.
srst Reset server statistics.
srvr Lists full details for the server.
stat Lists brief details for the server and connected clients.
wchs Lists brief information on watches for the server.
wchc Lists detailed information on watches for the server, by session
wchp Lists detailed information on watches for the server, by path.
mntr Outputs a list of variables that could be used for monitoring the health of the cluster.
Zookeeper shell
kubectl exec -it $(oc get kafka --no-headers -o custom-columns=:.metadata.name)-zookeeper-0 -n kafka -- bin/zookeeper-shell.sh 127.0.0.1:12181

=kafka-exporter=
Get stats from kafka-exporter.
kubectl -it exec deployment/kafka-cluster-kafka-exporter -- curl -s localhost:9404/metrics
=connectors=
These are the actual plugins that Kafka Connect uses to interface with external systems. Connectors can be either sources (pulling data into Kafka) or sinks (pushing data from Kafka). (io.debezium.connector.postgresql.PostgresConnector)

=Kafka Connect=
Is a framework and set of tools for connecting Kafka with external systems such as databases, key-value stores, search indexes, and file systems.
==
=Test kafka client=
<pre>
# Get username password
oc get secrets -l strimzi.io/kind=KafkaUser -o custom-columns=NAME:.metadata.name,PASSWORD:.data.password --no-headers | head -1 | while read USERNAME PASSWORD ; do echo export USERNAME=$USERNAME PASSWORD=$(base64 -d <<< "${PASSWORD}") ; done
# Start bash in kafka pod.
kafka_bash ()
{
BOOTSTRAP_SERVERS=$(kubectl get kafka -o=jsonpath='{.items[0].status.listeners[?(@.name=="plain")].bootstrapServers}{"\n"}');
if [ ! "${BOOTSTRAP_SERVERS}" ]; then
echo kafka not available.;
return 1;
fi;
oc exec -n kafka $(oc get pods -l strimzi.io/component-type=kafka --no-headers -o custom-columns=:.metadata.name --field-selector "status.phase=Running" | head -1) -it -- bash -c "export BOOTSTRAP_SERVERS=${BOOTSTRAP_SERVERS}; export PATH=\$PATH:/opt/kafka/bin ; exec bash"
}
kafka_bash
# In pod paste reply from command above. export USERNAME=...
# In pod. Get p12 password.
export CERTS_STORE_PASSWORD=$(ls /proc/[0-9]*/environ | while read i ; do cat $i 2>/dev/null | tr '\000' '\n' ; done | sed -n 's/^CERTS_STORE_PASSWORD=//p' | uniq)
# Get bootstrap url
export BOOTSTRAP_SERVERS_TLS=$(openssl s_client -connect localhost:9094 < /dev/null 2> /dev/null | openssl x509 -inform pem -text 2>&1 | grep -A1 " Subject Alternative Name: " | tail -1 | sed 's/ //g;s/:/\n/g;s/DNS//g;s/,//g'| awk '{ print length, $0 }' | sort -n | cut -d" " -f2- | grep -v ^$ | while read i ; do ANSWER=$(nc -zv -w1 $i 9094 2>&1 | cat -v | xargs) ; if [[ "${ANSWER}" == *"Connected"* ]]; then echo $i:9094 ; break ; fi ; done)
# in pod generate config
cat << EOF > /tmp/kafka_login.properties
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="${USERNAME}" password="${PASSWORD}";
ssl.truststore.location=/tmp/kafka/cluster.truststore.p12
ssl.truststore.password=${CERTS_STORE_PASSWORD}
ssl.truststore.type=PKCS12
EOF
# In pod. Test connectivity.
bin/kafka-topics.sh --list --bootstrap-server ${BOOTSTRAP_SERVERS_TLS} --command-config /tmp/kafka_login.properties
...
# test from other computer
# Get external Loadbalancer
echo export BOOTSTRAP_SERVERS_TLS=$(oc get service kafka-cluster-kafka-external-bootstrap -o jsonpath='{.status.loadBalancer.ingress[0].hostname}'):9094
# Copy connection file and certificates.
oc cp $(oc get pods -l strimzi.io/component-type=kafka --no-headers -o custom-columns=:.metadata.name --field-selector "status.phase=Running" | head -1):/tmp/kafka_login.properties kafka_login.properties
oc cp $(oc get pods -l strimzi.io/component-type=kafka --no-headers -o custom-columns=:.metadata.name --field-selector "status.phase=Running" | head -1):/tmp/kafka/cluster.truststore.p12 cluster.truststore.p12
scp <bastion>:/tmp
# Login to bastion
# Download client.
(cd /tmp/ ; curl https://dlcdn.apache.org/kafka/4.0.0/kafka_2.13-4.0.0.tgz -O)
# Update config to where files are.
sed -i 's%ssl.truststore.location=/tmp/kafka/cluster.truststore.p12%ssl.truststore.location=/tmp/cluster.truststore.p12%g' /tmp/kafka_login.properties
# Test connection. You should get all kafka queues
kafka*/bin/kafka-topics.sh --list --bootstrap-server ${BOOTSTRAP_SERVERS_TLS} --command-config /tmp/kafka_login.properties
</pre>

Kafka - Revision history

Ekaanbj: /* Kafka Connect */