kafka-python with kerberos kafka.

Hello everyone!!! Confluent 3.0.1 - Kafka and kafka-python-3.0.1 When I use python code:

#!/opt/distr/confluent-3.0.1/venv/bin/python3.4
# coding: utf8

from kafka import KafkaProducer
from kafka import KafkaClient
import logging


logging.basicConfig(filename='example.log',level=logging.DEBUG)
params = {
  'bootstrap_servers': ['<hostname>:9092'],
  'security_protocol': 'SASL_PLAINTEXT',
  'sasl_mechanism': 'GSSAPI',
}

producer = KafkaProducer(**params)
producer.send('testsasl', b'this working code')
producer.flush()
producer.close()

I get following error:

INFO:kafka.conn:<BrokerConnection node_id=bootstrap host=Server_Name/Server_IP port=9092>: connecting to Server_IP:9092
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=Server_Name/Server_IP port=9092>: established TCP connection
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=Server_Name/Server_IP port=9092>: initiating SASL authentication
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=Server_Name/Server_IP port=9092> Request 1: SaslHandShakeRequest_v0(mechanism='GSSAPI')
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=Server_Name/Server_IP port=9092> Response 1: SaslHandShakeResponse_v0(error_code=0, enabled_mechanisms=['PLAIN', 'GSSAPI'])
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=Server_Name/Server_IP port=9092>: canonical Servicename: kafka/Server_Name@Kerberos_DOMAIN
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=Server_Name/Server_IP port=9092>: Security Context complete
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=Server_Name/Server_IP port=9092>: Successful GSSAPI handshake for client@Kerberos_DOMAIN
INFO:kafka.conn:<BrokerConnection node_id=bootstrap host=Server_Name/Server_IP port=9092>: Authenticated as kafka@Server_Name
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=Server_Name/Server_IP port=9092>: Connection complete.
DEBUG:kafka.client:Node bootstrap connected
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=Server_Name/Server_IP port=9092> Request 2: MetadataRequest_v0(topics=[])
ERROR:kafka.conn:<BrokerConnection node_id=bootstrap host=Server_Name/Server_IP port=9092>: socket disconnected
INFO:kafka.conn:<BrokerConnection node_id=bootstrap host=Server_Name/Server_IP port=9092>: Closing connection. ConnectionError: socket disconnected
DEBUG:kafka.conn:<BrokerConnection node_id=bootstrap host=Server_Name/Server_IP port=9092>: reconnect backoff 0.04484597723684187 after 1 failures
ERROR:kafka.client:Unable to bootstrap from [('Server_Name', 9092, <AddressFamily.AF_UNSPEC: 0>)]
INFO:kafka.producer.kafka:Kafka producer closed

In Kafka logs I see following:

[2018-01-07 21:24:36,475] DEBUG Got ping response for sessionid: 0x160c74c65dd0000 after 0ms (org.apache.zookeeper.ClientCnxn)
[2018-01-07 21:24:36,488] DEBUG Set SASL server state to HANDSHAKE_REQUEST (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2018-01-07 21:24:36,489] DEBUG Handle Kafka request SASL_HANDSHAKE (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2018-01-07 21:24:36,489] DEBUG Using SASL mechanism 'GSSAPI' provided by client (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2018-01-07 21:24:36,489] DEBUG Creating SaslServer for kafka/Server_Name@Kerberos_DOMAIN with mechanism GSSAPI (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2018-01-07 21:24:36,491] DEBUG Set SASL server state to AUTHENTICATE (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2018-01-07 21:24:36,521] INFO Successfully authenticated client: authenticationID=client@Kerberos_DOMAIN; authorizationID=client@Kerberos_DOMAIN. (org.apache.kafka.common.security.authenticator.SaslServerCallbackHandler)
[2018-01-07 21:24:36,521] INFO Setting authorizedID: client (org.apache.kafka.common.security.authenticator.SaslServerCallbackHandler)
[2018-01-07 21:24:36,521] DEBUG Set SASL server state to COMPLETE (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2018-01-07 21:24:37,612] DEBUG Set SASL server state to HANDSHAKE_REQUEST (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2018-01-07 21:24:37,612] DEBUG Handle Kafka request SASL_HANDSHAKE (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2018-01-07 21:24:37,612] DEBUG Using SASL mechanism 'GSSAPI' provided by client (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2018-01-07 21:24:37,612] DEBUG Creating SaslServer for kafka/Server_Name@Kerberos_DOMAIN with mechanism GSSAPI (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2018-01-07 21:24:37,613] DEBUG Set SASL server state to AUTHENTICATE (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2018-01-07 21:24:37,620] INFO Successfully authenticated client: authenticationID=client@Kerberos_DOMAIN; authorizationID=client@Kerberos_DOMAIN. (org.apache.kafka.common.security.authenticator.SaslServerCallbackHandler)
[2018-01-07 21:24:37,620] INFO Setting authorizedID: client (org.apache.kafka.common.security.authenticator.SaslServerCallbackHandler)
[2018-01-07 21:24:37,620] DEBUG Set SASL server state to COMPLETE (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2018-01-07 21:24:37,643] DEBUG Added sensor with name ProduceThrottleTime-console-producer (org.apache.kafka.common.metrics.Metrics)
[2018-01-07 21:24:37,643] DEBUG Added sensor with name Produce-console-producer (org.apache.kafka.common.metrics.Metrics)
[2018-01-07 21:24:38,478] DEBUG Got ping response for sessionid: 0x160c74c65dd0000 after 0ms (org.apache.zookeeper.ClientCnxn)
[2018-01-07 21:24:40,479] DEBUG Got ping response for sessionid: 0x160c74c65dd0000 after 0ms (org.apache.zookeeper.ClientCnxn)
[2018-01-07 21:24:42,481] DEBUG Got ping response for sessionid: 0x160c74c65dd0000 after 0ms (org.apache.zookeeper.ClientCnxn)
[2018-01-07 21:24:44,484] DEBUG Got ping response for sessionid: 0x160c74c65dd0000 after 0ms (org.apache.zookeeper.ClientCnxn)
[2018-01-07 21:24:46,486] DEBUG Got ping response for sessionid: 0x160c74c65dd0000 after 0ms (org.apache.zookeeper.ClientCnxn)
[2018-01-07 21:24:48,489] DEBUG Got ping response for sessionid: 0x160c74c65dd0000 after 0ms (org.apache.zookeeper.ClientCnxn)
[2018-01-07 21:24:50,491] DEBUG Got ping response for sessionid: 0x160c74c65dd0000 after 0ms (org.apache.zookeeper.ClientCnxn)
[2018-01-07 21:24:52,494] DEBUG Got ping response for sessionid: 0x160c74c65dd0000 after 0ms (org.apache.zookeeper.ClientCnxn)
[2018-01-07 21:24:54,495] DEBUG Got ping response for sessionid: 0x160c74c65dd0000 after 0ms (org.apache.zookeeper.ClientCnxn)
[2018-01-07 21:24:56,498] DEBUG Got ping response for sessionid: 0x160c74c65dd0000 after 0ms (org.apache.zookeeper.ClientCnxn)
[2018-01-07 21:24:58,499] DEBUG Got ping response for sessionid: 0x160c74c65dd0000 after 0ms (org.apache.zookeeper.ClientCnxn)
[2018-01-07 21:25:00,501] DEBUG Got ping response for sessionid: 0x160c74c65dd0000 after 0ms (org.apache.zookeeper.ClientCnxn)
[2018-01-07 21:25:02,504] DEBUG Got ping response for sessionid: 0x160c74c65dd0000 after 0ms (org.apache.zookeeper.ClientCnxn)
[2018-01-07 21:25:02,605] DEBUG Set SASL server state to HANDSHAKE_REQUEST (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2018-01-07 21:25:02,605] DEBUG Handle Kafka request SASL_HANDSHAKE (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2018-01-07 21:25:02,605] DEBUG Using SASL mechanism 'GSSAPI' provided by client (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2018-01-07 21:25:02,605] DEBUG Creating SaslServer for kafka/Server_Name@Kerberos_DOMAIN with mechanism GSSAPI (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2018-01-07 21:25:02,606] DEBUG Set SASL server state to AUTHENTICATE (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2018-01-07 21:25:02,609] DEBUG Set SASL server state to FAILED (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2018-01-07 21:25:02,609] DEBUG Connection with /Server_IP disconnected (org.apache.kafka.common.network.Selector)
java.io.IOException: javax.security.sasl.SaslException: Handshake expecting no response data from server
        at org.apache.kafka.common.security.authenticator.SaslServerAuthenticator.authenticate(SaslServerAuthenticator.java:243)
        at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:64)
        at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:318)
        at org.apache.kafka.common.network.Selector.poll(Selector.java:283)
        at kafka.network.Processor.poll(SocketServer.scala:472)
        at kafka.network.Processor.run(SocketServer.scala:412)
        at java.lang.Thread.run(Thread.java:748)
Caused by: javax.security.sasl.SaslException: Handshake expecting no response data from server
        at com.sun.security.sasl.gsskerb.GssKrb5Server.doHandshake1(GssKrb5Server.java:209)
        at com.sun.security.sasl.gsskerb.GssKrb5Server.evaluateResponse(GssKrb5Server.java:158)
        at org.apache.kafka.common.security.authenticator.SaslServerAuthenticator.authenticate(SaslServerAuthenticator.java:228)
        ... 6 more
[2018-01-07 21:25:04,506] DEBUG Got ping response for sessionid: 0x160c74c65dd0000 after 0ms (org.apache.zookeeper.ClientCnxn)
[2018-01-07 21:25:06,509] DEBUG Got ping response for sessionid: 0x160c74c65dd0000 after 0ms (org.apache.zookeeper.ClientCnxn)
[2018-01-07 21:25:08,510] DEBUG Got ping response for sessionid: 0x160c74c65dd0000 after 0ms (org.apache.zookeeper.ClientCnxn)

Please help to solve this issue.

Issue Analytics

State:
Created 6 years ago
Comments:11 (2 by maintainers)

Top GitHub Comments

4reactions

jeffwidmancommented, Feb 21, 2018

Yes, they are both python libraries used to create kafka consumers/producers.

kafka-python is pure python, so will be slower to produce / consume, although it send / pre-fetches messages using a buffer so unless your business logic is super fast this shouldn’t matter much. It will be faster to debug in production since it’s pure python. And it is basically a port of the Java reference implementation so it will generally be faster for someone comfortable with Kafka to understand what’s happening because the semantics are consistent.

The confluent-kafka-python wraps librdkafka which is C-based. So message throughput will be faster by a huge margin, as long as your business logic isn’t the bottleneck. Many languages have librdkafka wrappers, so if you’re supporting multiple languages, this can be useful to have only one core piece of code to understand. Because it’s C, sometimes there can be packaging issues. Also more difficult to debug sometimes because you can’t just drop into pdb willy-nilly to examine stuff.

There are pros/cons to both. The feature comparison shows each library is ahead/behind in different areas. This library is maintained by volunteers, whereas confluent-kafka-python is maintained by two really nice guys (hi @edenhill / @mhowlett) who are paid by confluent so they may pull ahead on features over time, although their time is divided among several different clients so they can’t move as fast as they’d like. Really depends what you’re looking for, how many teams/services you’re supporting, how you want to handle production emergencies, etc.

HTH!

1reaction

Sudheer312commented, Jul 4, 2019

Is it possible to connect a kerberized kafka cluster with keytab from python using kafka-python 1.4.6 or should i move to confluent-kafka-python.

Top Results From Across the Web

kafka-python example with Kerberos auth

kafka -python example with Kerberos auth. GitHub Gist: instantly share code, notes, ... Requirements: kafka-python gssapi krbticket. import os. import time.

Connecting Kafka-Python with a cluster with Kerberos

I'm trying to connect to a Kafka with Kafka-Python, the Kafka cluster has Kerberos that we need to build some commands to do...

How Do I Use Python3.x to Connect to Kafka in a Security ...

The user does not know how to connect to a Kafka cluster with Kerberos authentication enabled in the Python3.x environment.The user needs an...

Use Case: Kerberos Enabled Kafka Producer using Python ...

The use case is simple. We have a Python application which needs to work with Apache Kafka, which has been secured by Kerberos....

Release 2.0.1 Dana Powers

KafkaConsumer is a high-level message consumer, ... sasl_kerberos_domain_name (str) – kerberos domain name to use in GSSAPI sasl.