GSSAPI authentication handshake is marked successful before fully complete

[vneeruk]

KafkaProducer(bootstrap_servers=bootstrap,security_protocol="SASL_PLAINTEXT",sasl_mechanism="GSSAPI",value_serializer=lambda m: json.dumps(m).encode('ascii'))

DEBUG:kafka.conn:: Successful GSSAPI handshake for vneeruko@DEMO.INTERNAL.LOCAL
INFO:kafka.conn:: Authenticated as None
DEBUG:kafka.conn:: 372 Connection complete.
DEBUG:kafka.client:297 Node bootstrap connected
DEBUG:kafka.client:253 sending metadata request: MetadataRequest_v0(topics=[])
DEBUG:kafka.conn: 687 Request 2: MetadataRequest_v0(topics=[])
DEBUG:kafka.client:client_async 257 before recv
ERROR:kafka.conn:: 745 socket disconnected
INFO:kafka.conn:: Closing connection. ConnectionError: socket disconnected
DEBUG:kafka.conn:: reconnect backoff 0.0472508888635 after 1 failures
DEBUG:kafka.client:client_async 259 after recv
ERROR:kafka.client:Unable to bootstrap from [('broker1', 6667, 0), ('broker2', 6667, 0), ('broker3', 6667, 0)]
Traceback (most recent call last):
File "kafka.py", line 11, in
kafkap = KafkaProducer(bootstrap_servers=bootstrap,security_protocol="SASL_PLAINTEXT",sasl_mechanism="GSSAPI",value_serializer=lambda m: json.dumps(m).encode('ascii'))
File "/kafka_test/kafka-python/kafka/producer/kafka.py", line 347, in init
**self.config)
File "/kafka_test/kafka-python/kafka/client_async.py", line 220, in init
self.config['api_version'] = self.check_version(timeout=check_timeout)
File "/kafka_test/kafka-python/kafka/client_async.py", line 830, in check_version
raise Errors.NoBrokersAvailable()
kafka.errors.NoBrokersAvailable: NoBrokersAvailable
INFO:kafka.producer.kafka:Kafka producer closed

[natedogs911]

I'm getting a similar error when trying to use plaintext+gssapi

[jeffwidman]

1. Did the console consumer or producer work using this connection string?
2. What version of kafka-python are you using?

[vneeruk]

I am trying to produce messages and its failing. Do you have snippet of console consumer, I will try that.
I am using 1.3.5.dev version of kafka-python

[jeffwidman]

The Kafka quickstart docs show how to use a console producer/consumer to send in messages: https://kafka.apache.org/quickstart

This lets you quickly bisect whether the problem is your Kafka cluster or your client, and if client whether producer or consumer side.

[vneeruk]

Console consumer and producer working fine with out issues.

[qkiss]

Any update on this? I get the same error after "Request 2: MetadataRequest_v1(topics=NULL)". The server does not respond to this request and the socket disconnects.

[tvoinarovskyi]

The support was introduced in #1152, maybe @Harald-Berghoff can give a hint as to why this happens.
I would like to help but have 0 understanding of Kerberos to put together any kind of test for this. If anyone could provide some sort of test so I could just run it and debug (for example a docker container/swarm and instructions on how to run it). Without a Kerberos instance not much I can do.

[funes79]

I have the same problem. Using SASL_GSSAPI as security_protocol. Any hints?

DEBUG:kafka.producer.kafka:Starting the Kafka producer
DEBUG:kafka.metrics.metrics:Added sensor with name connections-closed
DEBUG:kafka.metrics.metrics:Added sensor with name connections-created
DEBUG:kafka.metrics.metrics:Added sensor with name select-time
DEBUG:kafka.metrics.metrics:Added sensor with name io-time
INFO:kafka.client:Bootstrapping cluster metadata from [('10.197.23.81', 9092, 2)]
DEBUG:kafka.client:Attempting to bootstrap via node at 10.197.23.81:9092
DEBUG:kafka.metrics.metrics:Added sensor with name bytes-sent-received
DEBUG:kafka.metrics.metrics:Added sensor with name bytes-sent
DEBUG:kafka.metrics.metrics:Added sensor with name bytes-received
DEBUG:kafka.metrics.metrics:Added sensor with name request-latency
DEBUG:kafka.metrics.metrics:Added sensor with name node-bootstrap.bytes-sent
DEBUG:kafka.metrics.metrics:Added sensor with name node-bootstrap.bytes-received
DEBUG:kafka.metrics.metrics:Added sensor with name node-bootstrap.latency
DEBUG:kafka.conn:: creating new socket
DEBUG:kafka.conn:: setting socket option (6, 1, 1)
INFO:kafka.conn:: connecting to 10.197.23.81:9092
DEBUG:kafka.conn:: established TCP connection
DEBUG:kafka.conn:: Connection complete.
DEBUG:kafka.client:Node bootstrap connected
DEBUG:kafka.conn: Request 1: MetadataRequest_v0(topics=[])
ERROR:kafka.conn:: socket disconnected

[funes79]

Update: I see this error in Kafka broker.

java.io.IOException: org.apache.kafka.common.errors.IllegalSaslStateException: Unexpected Kafka request of type METADATA during SASL handshake

Based on this https://issues.apache.org/jira/browse/KAFKA-5458

How to set jaas client properties?

[qkiss]

Is this what you need?
export KAFKA_OPTS=-Djava.security.auth.login.config=/home/user/kafka_client_jaas.conf

[funes79]

For java it is clear. But for python I just start a shell and run a python code. How can I define this in python?

[Harald-Berghoff]

Are you having a cluster configuration with zookeeper?

@funes79 : can you provide (or check) the log of the message exchange just before the error occured? Looks like the broker is expecting some confirmation message, while the client already considers the authentication to be finished.

[jeffwidman]

nudge @funes79

[funes79]

Hi Harald.
Went back to this test cluster, maybe something changed but log dont displays anything.
So I tried the console client and console producer approach. This scenario, writing into a kafka in one window and reading in the second is fine.

[ec2-user@ip-xx-xx-xx-xx ~]$ ./writekafka-test
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/opt/cloudera/parcels/KAFKA-2.2.0-1.2.2.0.p0.68/lib/kafka/libs/slf4j-log4j12-1.7.21.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/cloudera/parcels/KAFKA-2.2.0-1.2.2.0.p0.68/lib/kafka/libs/slf4j-log4j12-1.7.5.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
17/10/06 10:45:48 INFO producer.ProducerConfig: ProducerConfig values:
acks = 1
batch.size = 16384
block.on.buffer.full = false
bootstrap.servers = [ip-xx-xx-xx-yy.eu-west-1.compute.internal:9092]
buffer.memory = 33554432
client.id = console-producer
compression.type = none
connections.max.idle.ms = 540000
interceptor.classes = null
key.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
linger.ms = 1000
max.block.ms = 60000
max.in.flight.requests.per.connection = 5
max.request.size = 1048576
metadata.fetch.timeout.ms = 60000
metadata.max.age.ms = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.sample.window.ms = 30000
partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
receive.buffer.bytes = 32768
reconnect.backoff.ms = 50
request.timeout.ms = 1500
retries = 3
retry.backoff.ms = 100
sasl.jaas.config = null
sasl.kerberos.kinit.cmd = /usr/bin/kinit
sasl.kerberos.min.time.before.relogin = 60000
sasl.kerberos.service.name = kafka
sasl.kerberos.ticket.renew.jitter = 0.05
sasl.kerberos.ticket.renew.window.factor = 0.8
sasl.mechanism = GSSAPI
security.protocol = SASL_PLAINTEXT
send.buffer.bytes = 102400
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
ssl.endpoint.identification.algorithm = null
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.location = null
ssl.keystore.password = null
ssl.keystore.type = JKS
ssl.protocol = TLS
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS
timeout.ms = 30000
value.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer

17/10/06 10:45:48 INFO authenticator.AbstractLogin: Successfully logged in.
17/10/06 10:45:48 INFO kerberos.KerberosLogin: [Principal=myuser@MYREALM.LOCAL]: TGT refresh thread started.
17/10/06 10:45:48 INFO kerberos.KerberosLogin: [Principal=myuser@MYREALM.LOCAL]: TGT valid starting at: Fri Oct 06 10:45:48 CEST 2017
17/10/06 10:45:48 INFO kerberos.KerberosLogin: [Principal=myuser@MYREALM.LOCAL]: TGT expires: Sun Oct 08 10:45:48 CEST 2017
17/10/06 10:45:48 INFO kerberos.KerberosLogin: [Principal=myuser@MYREALM.LOCAL]: TGT refresh sleeping until: Sun Oct 08 01:56:34 CEST 2017
17/10/06 10:45:48 INFO utils.AppInfoParser: Kafka version : 0.10.2-kafka-2.2.0
17/10/06 10:45:48 INFO utils.AppInfoParser: Kafka commitId : unknown
Helloworkld

/* Different window: */

[ec2-user@ip-xx-xx-xx-xx ~]$ ./readkafka-test
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/opt/cloudera/parcels/KAFKA-2.2.0-1.2.2.0.p0.68/lib/kafka/libs/slf4j-log4j12-1.7.21.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/cloudera/parcels/KAFKA-2.2.0-1.2.2.0.p0.68/lib/kafka/libs/slf4j-log4j12-1.7.5.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
17/10/06 10:44:17 INFO consumer.ConsumerConfig: ConsumerConfig values:
auto.commit.interval.ms = 5000
auto.offset.reset = latest
bootstrap.servers = [ip-xx-xx-xx-yy.eu-west-1.compute.internal:9092]
check.crcs = true
client.id =
connections.max.idle.ms = 540000
enable.auto.commit = true
exclude.internal.topics = true
fetch.max.bytes = 52428800
fetch.max.wait.ms = 500
fetch.min.bytes = 1
group.id = console-consumer-78149
heartbeat.interval.ms = 3000
interceptor.classes = null
internal.leave.group.on.close = true
key.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
max.partition.fetch.bytes = 1048576
max.poll.interval.ms = 300000
max.poll.records = 500
metadata.max.age.ms = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.recording.level = INFO
metrics.sample.window.ms = 30000
partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
receive.buffer.bytes = 65536
reconnect.backoff.ms = 50
request.timeout.ms = 305000
retry.backoff.ms = 100
sasl.jaas.config = null
sasl.kerberos.kinit.cmd = /usr/bin/kinit
sasl.kerberos.min.time.before.relogin = 60000
sasl.kerberos.service.name = kafka
sasl.kerberos.ticket.renew.jitter = 0.05
sasl.kerberos.ticket.renew.window.factor = 0.8
sasl.mechanism = GSSAPI
security.protocol = SASL_PLAINTEXT
send.buffer.bytes = 131072
session.timeout.ms = 10000
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
ssl.endpoint.identification.algorithm = null
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.location = null
ssl.keystore.password = null
ssl.keystore.type = JKS
ssl.protocol = TLS
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS
value.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer

17/10/06 10:44:17 INFO authenticator.AbstractLogin: Successfully logged in.
17/10/06 10:44:17 INFO kerberos.KerberosLogin: [Principal=myuser@MYREALM.LOCAL]: TGT refresh thread started.
17/10/06 10:44:17 INFO kerberos.KerberosLogin: [Principal=myuser@MYREALM.LOCAL]: TGT valid starting at: Fri Oct 06 10:44:17 CEST 2017
17/10/06 10:44:17 INFO kerberos.KerberosLogin: [Principal=myuser@MYREALM.LOCAL]: TGT expires: Sun Oct 08 10:44:17 CEST 2017
17/10/06 10:44:17 INFO kerberos.KerberosLogin: [Principal=myuser@MYREALM.LOCAL]: TGT refresh sleeping until: Sun Oct 08 03:19:19 CEST 2017
17/10/06 10:44:17 INFO utils.AppInfoParser: Kafka version : 0.10.2-kafka-2.2.0
17/10/06 10:44:17 INFO utils.AppInfoParser: Kafka commitId : unknown
17/10/06 10:44:17 INFO internals.AbstractCoordinator: Discovered coordinator ip-xx-xx-xx-zz.eu-west-1.compute.internal:9092 (id: 2147483568 rack: null) for group console-consumer-78149.
17/10/06 10:44:17 INFO internals.ConsumerCoordinator: Revoking previously assigned partitions [] for group console-consumer-78149
17/10/06 10:44:17 INFO internals.AbstractCoordinator: (Re-)joining group console-consumer-78149
17/10/06 10:44:17 INFO internals.AbstractCoordinator: Successfully joined group console-consumer-78149 with generation 1
17/10/06 10:44:17 INFO internals.ConsumerCoordinator: Setting newly assigned partitions [test-1, test-0, test-2] for group console-consumer-78149
null:Helloworkld

/* Active Controller, broker log: */

2017-10-06 10:33:10,484 INFO kafka.coordinator.GroupMetadataManager: [Group Metadata Manager on Broker 77]: Removed 0 expired offsets in 0 milliseconds.

2017-10-06 10:35:28,544 INFO org.apache.kafka.common.security.authenticator.SaslServerCallbackHandler: Successfully authenticated client: authenticationID=myuser@MYREALM.LOCAL; authorizationID=myuser@MYREALM.LOCAL.
2017-10-06 10:35:28,544 INFO org.apache.kafka.common.security.authenticator.SaslServerCallbackHandler: Setting authorizedID: myuser
2017-10-06 10:35:28,552 INFO kafka.coordinator.GroupCoordinator: [GroupCoordinator 77]: Preparing to restabilize group console-consumer-77645 with old generation 0
2017-10-06 10:35:28,558 INFO kafka.coordinator.GroupCoordinator: [GroupCoordinator 77]: Stabilized group console-consumer-77645 generation 1
2017-10-06 10:35:28,565 INFO kafka.coordinator.GroupCoordinator: [GroupCoordinator 77]: Assignment received from leader for group console-consumer-77645 for generation 1
2017-10-06 10:35:28,622 INFO org.apache.kafka.common.security.authenticator.SaslServerCallbackHandler: Successfully authenticated client: authenticationID=myuser@MYREALM.LOCAL; authorizationID=myuser@MYREALM.LOCAL.
2017-10-06 10:35:28,622 INFO org.apache.kafka.common.security.authenticator.SaslServerCallbackHandler: Setting authorizedID: myuser
2017-10-06 10:35:37,811 INFO kafka.coordinator.GroupCoordinator: [GroupCoordinator 77]: Preparing to restabilize group console-consumer-77645 with old generation 1
2017-10-06 10:35:37,812 INFO kafka.coordinator.GroupCoordinator: [GroupCoordinator 77]: Group console-consumer-77645 with generation 2 is now empty
2017-10-06 10:43:10,487 INFO kafka.coordinator.GroupMetadataManager: [Group Metadata Manager on Broker 77]: Removed 0 expired offsets in 3 milliseconds.
2017-10-06 10:44:17,915 INFO kafka.log.Log: Rolled new log segment for '__consumer_offsets-10' in 1 ms.
2017-10-06 10:44:17,976 INFO org.apache.kafka.common.security.authenticator.SaslServerCallbackHandler: Successfully authenticated client: authenticationID=myuser@MYREALM.LOCAL; authorizationID=myuser@MYREALM.LOCAL.
2017-10-06 10:44:17,976 INFO org.apache.kafka.common.security.authenticator.SaslServerCallbackHandler: Setting authorizedID: myuser
2017-10-06 10:44:23,236 INFO org.apache.kafka.common.security.authenticator.SaslServerCallbackHandler: Successfully authenticated client: authenticationID=myuser@MYREALM.LOCAL; authorizationID=myuser@MYREALM.LOCAL.
2017-10-06 10:44:23,236 INFO org.apache.kafka.common.security.authenticator.SaslServerCallbackHandler: Setting authorizedID: myuser
2017-10-06 10:44:28,766 INFO kafka.log.LogCleaner: Cleaner 0: Beginning cleaning of log __consumer_offsets-10.
2017-10-06 10:44:28,766 INFO kafka.log.LogCleaner: Cleaner 0: Building offset map for __consumer_offsets-10...
2017-10-06 10:44:28,786 INFO kafka.log.LogCleaner: Cleaner 0: Building offset map for log __consumer_offsets-10 for 1 segments in offset range [0, 900).
2017-10-06 10:44:28,800 INFO kafka.log.LogCleaner: Cleaner 0: Offset map for log __consumer_offsets-10 complete.
2017-10-06 10:44:28,801 INFO kafka.log.LogCleaner: Cleaner 0: Cleaning log __consumer_offsets-10 (cleaning prior to Tue Aug 08 11:02:46 CEST 2017, discarding tombstones prior to Thu Jan 01 01:00:00 CET 1970)...
2017-10-06 10:44:28,803 INFO kafka.log.LogCleaner: Cleaner 0: Cleaning segment 0 in log __consumer_offsets-10 (largest timestamp Tue Aug 08 11:02:46 CEST 2017) into 0, retaining deletes.
2017-10-06 10:44:28,818 INFO kafka.log.LogCleaner: Cleaner 0: Swapping in cleaned segment 0 for segment(s) 0 in log __consumer_offsets-10.
2017-10-06 10:44:28,820 INFO kafka.log.LogCleaner: [kafka-log-cleaner-thread-0],
Log cleaner thread 0 cleaned log __consumer_offsets-10 (dirty section = [0, 0])
0.1 MB of log processed in 0.1 seconds (1.6 MB/sec).
Indexed 0.1 MB in 0.0 seconds (2.5 Mb/sec, 64.8% of total time)
Buffer utilization: 0.0%
Cleaned 0.1 MB in 0.0 seconds (4.7 Mb/sec, 35.2% of total time)
Start size: 0.1 MB (900 messages)
End size: 0.0 MB (8 messages)
99.4% size reduction (99.1% fewer messages)

2017-10-06 10:45:28,819 INFO kafka.log.Log: Deleting segment 0 from log __consumer_offsets-10.
2017-10-06 10:45:28,820 INFO kafka.log.OffsetIndex: Deleting index /data/data5/kafka/__consumer_offsets-10/00000000000000000000.index.deleted
2017-10-06 10:45:28,820 INFO kafka.log.TimeIndex: Deleting index /data/data5/kafka/__consumer_offsets-10/00000000000000000000.timeindex.deleted
2017-10-06 10:45:54,149 INFO org.apache.kafka.common.security.authenticator.SaslServerCallbackHandler: Successfully authenticated client: authenticationID=myuser@MYREALM.LOCAL; authorizationID=myuser@MYREALM.LOCAL.
2017-10-06 10:45:54,149 INFO org.apache.kafka.common.security.authenticator.SaslServerCallbackHandler: Setting authorizedID: myuser

/* another Broker log: */

2017-10-06 10:33:36,332 INFO kafka.coordinator.GroupMetadataManager: [Group Metadata Manager on Broker 79]: Removed 0 expired offsets in 0 milliseconds.
2017-10-06 10:35:28,657 INFO org.apache.kafka.common.security.authenticator.SaslServerCallbackHandler: Successfully authenticated client: authenticationID=myuser@MYREALM.LOCAL; authorizationID=myuser@MYREALM.LOCAL.
2017-10-06 10:35:28,657 INFO org.apache.kafka.common.security.authenticator.SaslServerCallbackHandler: Setting authorizedID: myuser
2017-10-06 10:43:36,332 INFO kafka.coordinator.GroupMetadataManager: [Group Metadata Manager on Broker 79]: Removed 0 expired offsets in 0 milliseconds.
2017-10-06 10:44:17,886 INFO org.apache.kafka.common.security.authenticator.SaslServerCallbackHandler: Successfully authenticated client: authenticationID=myuser@MYREALM.LOCAL; authorizationID=myuser@MYREALM.LOCAL.
2017-10-06 10:44:17,886 INFO org.apache.kafka.common.security.authenticator.SaslServerCallbackHandler: Setting authorizedID: myuser
2017-10-06 10:44:17,892 INFO kafka.coordinator.GroupCoordinator: [GroupCoordinator 79]: Preparing to restabilize group console-consumer-78149 with old generation 0
2017-10-06 10:44:17,897 INFO kafka.coordinator.GroupCoordinator: [GroupCoordinator 79]: Stabilized group console-consumer-78149 generation 1
2017-10-06 10:44:17,904 INFO kafka.coordinator.GroupCoordinator: [GroupCoordinator 79]: Assignment received from leader for group console-consumer-78149 for generation 1
2017-10-06 10:44:17,911 INFO kafka.log.Log: Rolled new log segment for '__consumer_offsets-10' in 1 ms.
2017-10-06 10:44:17,982 INFO org.apache.kafka.common.security.authenticator.SaslServerCallbackHandler: Successfully authenticated client: authenticationID=myuser@MYREALM.LOCAL; authorizationID=myuser@MYREALM.LOCAL.
2017-10-06 10:44:17,982 INFO org.apache.kafka.common.security.authenticator.SaslServerCallbackHandler: Setting authorizedID: myuser
2017-10-06 10:44:26,269 INFO kafka.log.LogCleaner: Cleaner 0: Beginning cleaning of log __consumer_offsets-10.
2017-10-06 10:44:26,269 INFO kafka.log.LogCleaner: Cleaner 0: Building offset map for __consumer_offsets-10...
2017-10-06 10:44:26,289 INFO kafka.log.LogCleaner: Cleaner 0: Building offset map for log __consumer_offsets-10 for 1 segments in offset range [0, 900).
2017-10-06 10:44:26,300 INFO kafka.log.LogCleaner: Cleaner 0: Offset map for log __consumer_offsets-10 complete.
2017-10-06 10:44:26,301 INFO kafka.log.LogCleaner: Cleaner 0: Cleaning log __consumer_offsets-10 (cleaning prior to Tue Aug 08 11:02:46 CEST 2017, discarding tombstones prior to Thu Jan 01 01:00:00 CET 1970)...
2017-10-06 10:44:26,303 INFO kafka.log.LogCleaner: Cleaner 0: Cleaning segment 0 in log __consumer_offsets-10 (largest timestamp Tue Aug 08 11:02:46 CEST 2017) into 0, retaining deletes.
2017-10-06 10:44:26,320 INFO kafka.log.LogCleaner: Cleaner 0: Swapping in cleaned segment 0 for segment(s) 0 in log __consumer_offsets-10.
2017-10-06 10:44:26,322 INFO kafka.log.LogCleaner: [kafka-log-cleaner-thread-0],
Log cleaner thread 0 cleaned log __consumer_offsets-10 (dirty section = [0, 0])
0.1 MB of log processed in 0.1 seconds (1.7 MB/sec).
Indexed 0.1 MB in 0.0 seconds (2.9 Mb/sec, 59.6% of total time)
Buffer utilization: 0.0%
Cleaned 0.1 MB in 0.0 seconds (4.2 Mb/sec, 40.4% of total time)
Start size: 0.1 MB (900 messages)
End size: 0.0 MB (8 messages)
99.4% size reduction (99.1% fewer messages)

2017-10-06 10:45:26,321 INFO kafka.log.Log: Deleting segment 0 from log __consumer_offsets-10.
2017-10-06 10:45:26,322 INFO kafka.log.OffsetIndex: Deleting index /data/data4/kafka/__consumer_offsets-10/00000000000000000000.index.deleted
2017-10-06 10:45:26,322 INFO kafka.log.TimeIndex: Deleting index /data/data4/kafka/__consumer_offsets-10/00000000000000000000.timeindex.deleted

[funes79]

Writing with python throws this error:

[ec2-user@ip-xx-xx-xx-xx ~]$ python kafka-producer.py test ./big.txt
Running kafka producer with 2 retries and all acks.
Traceback (most recent call last):
File "kafka-producer.py", line 11, in
producer = KafkaProducer(bootstrap_servers=['xx.xx.xx.zz:9092'],security_protocol='SASL_GSSAPI',sasl_mechanism='GSSAPI', acks='all', retries=2)
File "/usr/lib/python2.7/site-packages/kafka/producer/kafka.py", line 347, in init
**self.config)
File "/usr/lib/python2.7/site-packages/kafka/client_async.py", line 221, in init
self.config['api_version'] = self.check_version(timeout=check_timeout)
File "/usr/lib/python2.7/site-packages/kafka/client_async.py", line 826, in check_version
raise Errors.NoBrokersAvailable()
kafka.errors.NoBrokersAvailable: NoBrokersAvailable

[ec2-user@ip-xx-xx-xx-xx ~]$ more kafka-producer.py
import sys
import logging
logging.basicConfig(filename='example.log',level=logging.DEBUG)
from kafka import KafkaProducer
if len(sys.argv) < 3:
print('Usage: producer.py <file_path>')
print('Running kafka producer with 2 retries and all acks.')
fpath = sys.argv[2]
producer = KafkaProducer(bootstrap_servers=['10.197.xx.zz:9092'],security_protocol='SASL_GSSAPI',sasl_mechanism='GSSAPI', acks='all', retries=2)
logger.info('Reading from file '+fpath)
print('Reading from file: ' + fpath)

f = open(fpath, 'rb')
print('Writing to kafka topic: ' + sys.argv[1])
for line in f:
producer.send(sys.argv[1], line.strip())
f.close()
producer.flush()
print('Done.')
[ec2-user@ip-xx-xx-xx-xx ~]$
[ec2-user@ip-xx-xx-xx-xx ~]$
[ec2-user@ip-xx-xx-xx-xx ~]$ more example.log
DEBUG:kafka.producer.kafka:Starting the Kafka producer
DEBUG:kafka.metrics.metrics:Added sensor with name connections-closed
DEBUG:kafka.metrics.metrics:Added sensor with name connections-created
DEBUG:kafka.metrics.metrics:Added sensor with name select-time
DEBUG:kafka.metrics.metrics:Added sensor with name io-time
INFO:kafka.client:Bootstrapping cluster metadata from [('10.197.xx.zz', 9092, 2)]
DEBUG:kafka.client:Attempting to bootstrap via node at 10.197.xx.zz:9092
DEBUG:kafka.metrics.metrics:Added sensor with name bytes-sent-received
DEBUG:kafka.metrics.metrics:Added sensor with name bytes-sent
DEBUG:kafka.metrics.metrics:Added sensor with name bytes-received
DEBUG:kafka.metrics.metrics:Added sensor with name request-latency
DEBUG:kafka.metrics.metrics:Added sensor with name node-bootstrap.bytes-sent
DEBUG:kafka.metrics.metrics:Added sensor with name node-bootstrap.bytes-received
DEBUG:kafka.metrics.metrics:Added sensor with name node-bootstrap.latency
DEBUG:kafka.conn:: creating new socket
DEBUG:kafka.conn:: setting socket option (6, 1, 1)
INFO:kafka.conn:: connecting to 10.197.xx.zz:9092
DEBUG:kafka.conn:: established TCP connection
DEBUG:kafka.conn:: Connection complete.
DEBUG:kafka.client:Node bootstrap connected
DEBUG:kafka.conn: Request 1: MetadataRequest_v0(topics=[])
ERROR:kafka.conn:: socket disconnected
INFO:kafka.conn:: Closing connection. ConnectionError: socket disconnected
DEBUG:kafka.conn:: reconnect backoff 0.0518856326407 after 1 failures
ERROR:kafka.client:Unable to bootstrap from [('10.197.xx.zz', 9092, 2)]
INFO:kafka.producer.kafka:Kafka producer closed
[ec2-user@ip-xx-xx-xx-xx ~]$`