Add Kafka Consumer Factory to cache consumers for FromKafkaBatched #226
Conversation
|
I assume this is working for you locally? We typically found that kafka running on travis is well slower than locally, possibly leading to the failure to sync you are seeing in the test. When running locally, did you find the speedup you expected? Would be interesting to see benchmarks before/after. |
|
I think the tests related to this implementation are passing here, |
| "Use group.id: {} for restarting this job from a check point in future.".format(group_id)) | ||
| consumer_params['group.id'] = group_id | ||
|
|
||
| consumer_params['enable.auto.commit'] = 'false' |
There was a problem hiding this comment.
What was the reason behind this and the next (commented) line?
There was a problem hiding this comment.
I am manually committing the offsets after each batch is read successfully. And the job would restart from this checkpoint and not from an auto-commit in case of a failure.
This configuration specifies the maximum time a consumer can stay idle between two poll operations. A consumer might stay idle for a longer period with current implementation. For example if Dask cluster is still processing older batches and the consumer is waiting for a thread to poll next batch.
Inspired by python-streamz#226, but in this case, consumers are thread-local, because they should not be called simultaneously from separate threads, and this should also work across processes.
|
Closing in favour of #233 - please add any further ideas there |
2 participants
I implemented a Kafka consumer factory that caches consumers and supplies consumer on demand to poll a message batch. This prevents unnecessary creation of consumers in the current implementation. Also, cached consumer is already seeked to latest offset(
lowof the next message batch ) to start polling messages immediately after a request for the next batch. One consumer per partition is created and reused for polling next batch in that partition.This also adds checkpointing for the job using Kafka consumer group. Consumer commits offsets after reading each batch. Restarting a job with same consumer group name would result in processing messages from last committed offset.