Pasted

# pip install confluent-kafka
 
from confluent_kafka import Consumer, KafkaError
 
topic = 'CLIENT_NAME-tests'
 
c = Consumer({
  'bootstrap.servers': 'server.name:9094',
  'group.id': 'mygroup',
 
  # set to False to make sure you do not update offsets
  'enable.auto.commit': False,
  #'enable.auto.commit': True,
 
  # set to earliest to start --from-beginning
  'auto.offset.reset': 'earliest',
 
  'security.protocol': 'SASL_SSL',
  'ssl.ca.location': 'ca-cert',
  'sasl.mechanisms': 'PLAIN',
  'sasl.username': 'CLIENT_USERNAME',
  'sasl.password': 'XXXX',
  })
 
c.subscribe([topic])
 
n = 0
while True:
  msg = c.poll(0)
 
  if msg is None:
    continue
 
  if msg.error():
    print("Consumer error: {}".format(msg.error))
    continue
 
  n += 1
  print("message: {} | {}".format(n, msg.value()))
 
c.close()