我有一个启用了SASL_SSL(身份验证(JAAS)和授权)的Kafka10集群。可以使用Java客户端和以下道具通过SASL进行连接。
ssl.keystore.location="client_keystore.jks" ssl.keystore.password="password" ssl.truststore.location="clienttruststore" ssl.truststore.password="password"
and passing the JAAS conf file thru the JVM params.
-Djava.security.auth.login.config=/path/to/client_jaas.conf
Is there anyway to achieve the same thing with the python client?
我一直在使用以下代码连接到位于底层的kafka的IBM Message Hub:
from kafka import KafkaProducer from kafka.errors import KafkaError import ssl sasl_mechanism = 'PLAIN' security_protocol = 'SASL_SSL' # Create a new context using system defaults, disable all but TLS1.2 context = ssl.create_default_context() context.options &= ssl.OP_NO_TLSv1 context.options &= ssl.OP_NO_TLSv1_1 producer = KafkaProducer(bootstrap_servers = app.config['KAFKA_BROKERS_SASL'], sasl_plain_username = app.config['KAFKA_USERNAME'], sasl_plain_password = app.config['KAFKA_PASSWORD'], security_protocol = security_protocol, ssl_context = context, sasl_mechanism = sasl_mechanism, api_version = (0,10), retries=5) def send_message(message): try: producer.send(app.config['KAFKA_TOPIC'], message.encode('utf-8')) except: print("Unexpected error:", sys.exc_info()[0]) raise