Producing to Kafka cluster is very slow

[sevagsfl]

Hello,

I am using pykafka-2.0.0
In a scenario with single zk/kafka, producing and consuming behave normally, with fast response.

In a scenario with multiple zk/kafka, consuming is normal but producing is very slow. I need to wait 5-6 seconds after calling "producer.produce(msg)" for the message to actually be produced.

Java client and kafka-console-producer.sh both produce with normal speed:

Console producer:

./kafka1/bin/kafka-console-producer.sh --broker-list localhost:9092,localhost:9093 --topic test

With the Java client:

props.put("metadata.broker.list", "0.0.0.0:9092,0.0.0.0:9093");
props.put("serializer.class", "kafka.serializer.StringEncoder");

ProducerConfig config = new ProducerConfig(props);
Producer<String, String> producer = new Producer<>(config);

String msg = "this is produced by java";
KeyedMessage<String, String> data = new KeyedMessage<>("test", msg);
producer.send(data);

With the pykafka client:

client = KafkaClient(hosts="0.0.0.0:9092,0.0.0.0:9093")
producer = client.topics['test'].get_producer()
producer.produce('this is produced by python')
time.sleep(10)

Java consumer, pykafka consumer, and console-consumer.sh are consuming equally fast.

Console consumer:

./kafka1/bin/kafka-console-consumer.sh --zookeeper 0.0.0.0:2181 --topic test --from-beginning

Pykafka consumer code:

client = KafkaClient(hosts="0.0.0.0:9092,0.0.0.0:9093")
topic = client.topics['test']
consumer = topic.get_balanced_consumer(
consumer_group='cumtime',
zookeeper_connect='0.0.0.0:2181,0.0.0.0:2182',
reset_offset_on_start=True,
auto_offset_reset=OffsetType.LATEST
)
while True:
message=consumer.consume()
print(message.value)

I have seen this behaviour with a cluster on the same machine, or on different machines.

Thanks for any help/advice.

[emmettbutler]

Hi @sevagsfl - sounds like you want the min_queued_messages and linger_ms settings on Producer. The default min_queued_messages is very high, so if you're producing a single message and expecting it to be sent right away, you may want to use this kwarg:

producer = topic.get_producer(min_queued_messages=1)

linger_ms can also be helpful here depending on your use case.

[sevagsfl]

Thanks @emmett9001, using 'min_queued_messages=1' solved my problem. The 5 second delay was indeed "linger_ms".

Is there a reason why I didn't need this for a single-broker configuration? Are messages only queued for the multi-broker case?

[emmettbutler]

I was wondering about that myself. It sounds possible that it's a bug, actually. I wonder if there's any other information you can provide about that - maybe the source code of your single-broker test case?

[sevagsfl]

Actually, my mistake - it was the upgrade from pykafka-1.1.1 to 2.0.0 that caused this change in behaviour, and not the clustering.

When I first made the upgrade, I didn't run my test script. Instead I just ran my application (which never exits, unlike my test script) and missed this change in behaviour.

Thanks again.

[j-style]

min_queued_messages (int) – The minimum number of messages the producer can have waiting in a queue before it flushes that queue to its broker (must be greater than 0). This paramater can be used to control the number of messages sent in one batch during async production. This parameter is automatically overridden to 1 when sync=True.

i used xxxTopic.get_sync_producer() to get a producer, according to the wiki, it should overridden min_queued_messages to 1, but it didn't work. when task is heavy, the produce speed is around 5 seconds(using default linger_ms value). then i changed my code to xxxTopic.get_sync_producer(min_queued_messages=1), well, the problem solved...

[emmettbutler]

@j-style min_queued_messages is overridden when sync=True right here. Do you have any additional information we might use to figure out what was causing your issue?

[j-style]

@emmett9001 i think it's my fault.
i've checked my pykafka version, which was 2.2.1. the overridden feature seems to be added since 2.3.0

the following desc is from version 2.0.4, no overridden.
min_queued_messages (int) – The minimum number of messages the producer can have waiting in a queue before it flushes that queue to its broker (must be greater than 0).