SimpleConsumer hangs

[wizzat]

Kafka version: 0.8.1

• localhost:9092
• localhost:9093
• localhost:9094

Messages are deposited from Play into Kafka on a topic. The message contents aren't particularly important because it's just heartbeat information right now, but the message sizes are somewhat beefy and expected to grow in production. The maximum message size is ~2mb. From the Kafka console consumer:

$ bin/ktail -b $TOPIC | perl -ne 'print length($_) . "\n"'
51632
92376
30768
48216
57068
78792
26292
27792
56208
60612
38288
78008
33300
127116
58132
44616
53708
88892
55324
79552
^CConsumed 20 messages

Kafka-Python: master

Code:

#!/usr/bin/env python
import threading, logging, time

from kafka.client import KafkaClient
from kafka.consumer import SimpleConsumer
from kafka.producer import SimpleProducer

logging.basicConfig(level=logging.DEBUG)

client = KafkaClient("localhost", 9092)
consumer = SimpleConsumer(client, "importer", "TOPIC", max_buffer_size=None)

for message in consumer:
    print(message.message.value[:100])

The log file:

DEBUG:kafka:About to send 30 bytes to Kafka, request 0
DEBUG:kafka:Reading response 0 from Kafka
DEBUG:kafka:About to read 4 bytes from Kafka
DEBUG:kafka:Read 4/4 bytes from Kafka
DEBUG:kafka:About to read 941 bytes from Kafka
DEBUG:kafka:Read 941/941 bytes from Kafka
DEBUG:kafka:Broker metadata: {0: BrokerMetadata(nodeId=0, host='localhost', port=9092), 1: BrokerMetadata(nodeId=1, host='localhost', port=9093), 2: BrokerMetadata(nodeId=2, host='localhost', port=9094)}
DEBUG:kafka:Topic metadata: -- snipped --
DEBUG:kafka:About to send 45 bytes to Kafka, request 1
DEBUG:kafka:Reading response 1 from Kafka
DEBUG:kafka:About to read 4 bytes from Kafka
DEBUG:kafka:Read 4/4 bytes from Kafka
DEBUG:kafka:About to read 192 bytes from Kafka
DEBUG:kafka:Read 192/192 bytes from Kafka
DEBUG:kafka:Broker metadata: {0: BrokerMetadata(nodeId=0, host='localhost', port=9092), 1: BrokerMetadata(nodeId=1, host='localhost', port=9093), 2: BrokerMetadata(nodeId=2, host='localhost', port=9094)}
DEBUG:kafka:Topic metadata: -- snipped --
DEBUG:kafka:About to send 77 bytes to Kafka, request 2
DEBUG:kafka:Reading response 2 from Kafka
DEBUG:kafka:About to read 4 bytes from Kafka
DEBUG:kafka:Read 4/4 bytes from Kafka
DEBUG:kafka:About to read 4141 bytes from Kafka
DEBUG:kafka:Read 4096/4141 bytes from Kafka
DEBUG:kafka:Read 4141/4141 bytes from Kafka
DEBUG:kafka:About to send 77 bytes to Kafka, request 3
DEBUG:kafka:Reading response 3 from Kafka
DEBUG:kafka:About to read 4 bytes from Kafka
DEBUG:kafka:Read 4/4 bytes from Kafka
DEBUG:kafka:About to read 45 bytes from Kafka
DEBUG:kafka:Read 45/45 bytes from Kafka
DEBUG:kafka:About to send 77 bytes to Kafka, request 4
DEBUG:kafka:Reading response 4 from Kafka
DEBUG:kafka:About to read 4 bytes from Kafka
DEBUG:kafka:Read 4/4 bytes from Kafka
DEBUG:kafka:About to read 45 bytes from Kafka
DEBUG:kafka:Read 45/45 bytes from Kafka
WARNING:kafka:Fetch size too small, increase to 8192 (2x) and retry
DEBUG:kafka:About to send 77 bytes to Kafka, request 5
DEBUG:kafka:Reading response 5 from Kafka
DEBUG:kafka:About to read 4 bytes from Kafka
DEBUG:kafka:Read 4/4 bytes from Kafka
DEBUG:kafka:About to read 8237 bytes from Kafka
DEBUG:kafka:Read 4096/8237 bytes from Kafka
DEBUG:kafka:Read 8192/8237 bytes from Kafka
DEBUG:kafka:Read 8237/8237 bytes from Kafka
DEBUG:kafka:About to send 77 bytes to Kafka, request 6
DEBUG:kafka:Reading response 6 from Kafka
DEBUG:kafka:About to read 4 bytes from Kafka

[wizzat]

The problem still persists when setting parameters:

• buffer_size = 2 * 1024 * 1024
• fetch_size_bytes = 2 * 1024 * 1024
• max_buffer_size = None

I actually get messages if I set iter_timeout very low, but then the application exits between messages.

[wizzat]

The same behavior happens with Kafka 0.8.0, though the code must be modified to turn off autocommit.

[wizzat]

I have independently fixed this by forking the library. I'm sure the patch is in one of my pull requests.