confluentinc / confluentinc/examples

Suggested Enhancements to Python Producer

Open
#1,106 0 comments 0 reactions 0 assignees View on GitHub
enhancement
Dominant language
Java
Stars
2.1k
Forks
1.1k
PR merge metrics
No merged PRs in 30d

Description

I wrote a python producer app that was consistently getting `Local: Queue full` errors (`BufferError`) and found [this code sample](https://github.com/confluentinc/confluent-kafka-python/issues/341#issuecomment-376067215) from Magnus (author of librdkafka) helpful:

```python
for i in range(msg_count):
while True:
try:
producer.produce(topic, value=msg_payload)
producer.poll(0)
break
except BufferError as e:
print(e, file=sys.stderr)
producer.poll(1)
```
The `while` loop just means that normally a record will be produced and then there's a `break` to move out of the while loop and back into the for loop, but in case we hit `BufferError`, we need to retry the produce before we move back to the for loop.

I would also suggest including some performance tuning knobs in the example.

I have a code snippet that might help. My app reads from a CSV file to produce to two different topics. Here is a relevant snippet:

```python
# configure schema registry client and avro serializer
schema_registry_client = SchemaRegistryClient(SCHEMA_REGISTRY_CONFIG)
my_avro_serializer = AvroSerializer(
schema_registry_client=schema_registry_client,
schema_str=MY_AVRO_SCHEMA
)
producer_config = KAFKA_CONFIG.copy()
producer_config['value.serializer'] = my_avro_serializer

# Configure other producer properties to tune performance.
# See https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
producer_config['queue.buffering.max.messages'] = 1000000
producer_config['linger.ms'] = 500
producer_config['batch.size'] = 2000000
producer_config['compression.type'] = 'snappy'

# Create producer and produce data
producer = SerializingProducer(producer_config)
records_produced = 0
with open('./my_data.csv') as data:
reader = csv.DictReader(data,fieldnames=MY_FIELDNAMES)
for row in reader:
# preprocess csv row to use float values instead of scientific notation
row_with_float_values = {field:float(value) for field, value in row.items()}
while True:
try:
producer.produce(
topic=MY_TOPIC,
value=row_with_float_values,
on_delivery=log_kafka_errors)
producer.poll(0)
records_produced += 1
break
except BufferError as err:
print(err, file=sys.stderr)
print(f"produced {records_produced} records so far", file=sys.stdout)
producer.flush()
print(f"produced a totoal of {records_produced} records", file=sys.stdout)
producer.flush()
```

There is some discussion about the extent to which calling `poll(0)` after each produce affects performance. Magnus says it's negligible but [here's someone who doesn't agree](https://github.com/confluentinc/confluent-kafka-python/issues/16#issuecomment-974255058).

It made sense to me to do `producer.flush()` when we hit a `BufferError` to clear everything out, but Magnus' example does `poll(1)` to wait for just 1 second to register callbacks for sent messages. I found that to cause me to hit the BufferError more often. To be honest, I don't really have a good sense of how `poll` works or what I should do with it from reading [the docs](https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#confluent_kafka.Consumer.poll), so maybe some more comments around how to use `poll()` and `flush()` would be good.

For reference, my full working example is here:
- https://github.com/confluentinc/demo-10x-storage#confluent-10x-storage-demo

Contributor guide

No contributing guide indexed for this repository

Research direction

Start by locating the Python producer example corresponding to the reported app and review its use of SerializingProducer, produce(), poll(), and flush(). Compare it with the proposed BufferError retry loop and performance settings. Done means the example explains or demonstrates retry behavior, polling and flushing, and the suggested tuning knobs without leaving their effects ambiguous.

Written by the indexing model from the issue text.

Assessment

Tech stack
kafka, python
Domain
data-engineering, documentation
Issue type
Documentation
Difficulty
3/5
Estimated time
1-2 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.