Kafka consumer group info retrieval using Python

Written by

Sorin Tudor

in

,

Hi,

I’ve been playing with kafka-python module to grab the info i need in order to reconfigure Datadog integration.

Unfortunately, there is a catch also on this method. And i will show you below.

Here is a little bit of not so elegant code.

from kafka import BrokerConnection
from kafka.protocol.admin import *
import socket

fqdn = socket.getfqdn()
bc = BrokerConnection(fqdn,9092,socket.AF_INET)
try:
  bc.connect_blocking()
except Exception as e:
  print(e)
if bc.connected():
   print("Connection to", fqdn, " established")

def getgroup():
 list_groups_request = ListGroupsRequest_v1()

 future0 = bc.send(list_groups_request)
 while not future0.is_done:
     for resp, f in bc.recv():
         f.success(resp)
 group_ids = ()
 for group in future0.value.groups:
     group_ids += (group[0],)

 print(group_ids)
     
 description = DescribeGroupsRequest_v1(group_ids)
 future1 = bc.send(description)
 while not future1.is_done:
    for resp, f in bc.recv():
        f.success(resp)

 for groupid in future1.value.groups:
     print('For group ',groupid[1],':\n')
     for meta in groupid[5]:
         print(meta[0],meta[2],sep="\n")
         print(meta[3])
 if future1.is_done:
    print("Group query is done")

getgroup()

As you will see, print(meta[3]) will return a very ugly binary data with topic names in it, that is not converted if you try with meta[3].decode(‘utf-8’)

I hope i can find a way to decrypt it.

Cheers

More posts