kafka---python访问Kerberos环境下的Kafka

[AronChung]

先确保环境装有kafka-python和gssapi

sudo pip3 install kafka-python -i https://mirrors.aliyun.com/pypi/simple/
sudo pip3 install gssapi -i https://mirrors.aliyun.com/pypi/simple/

代码

#!/usr/bin/env python3
# coding=utf-8

# 执行前需要先kinit

import time

from kafka import KafkaProducer
from kafka import KafkaConsumer

def kafka_python_producer_main():
    producer = KafkaProducer(bootstrap_servers='hadoop.data.com:9092,hadoop2.data.com:9092',
                             security_protocol='SASL_PLAINTEXT',
                             sasl_mechanism='GSSAPI',
                             sasl_kerberos_service_name='kafka',
                             sasl_plain_username='zcm')
    for _ in range(100):
        print(_)
        producer.send('zcm_test', b'some_message_bytes')
    producer.flush()
    producer.close()
    print('done')

def kafka_python_consumer_main():
    consumer = KafkaConsumer('zcm_test',
                             bootstrap_servers='hadoop.data.com:9092,hadoop2.data.com:9092',
                             group_id='testgroup',
                             auto_offset_reset='earliest',
                             security_protocol='SASL_PLAINTEXT',
                             sasl_mechanism='GSSAPI',
                             sasl_kerberos_service_name='kafka',
                             sasl_plain_username='zcm')
    for msg in consumer:
        print(msg.value)
        print(msg.partition)

if __name__ == '__main__':
    kafka_python_producer_main()
    time.sleep(1)
    kafka_python_consumer_main()

具体的参数可参考：
https://kafka-python.readthedocs.io/en/master/apidoc/KafkaProducer.html