Parsely/pykafka

Messages lost when broker closes connection

Open

#814 opened on Jun 7, 2018

View on GitHub
 (5 comments) (5 reactions) (0 assignees)Python (1,114 stars) (231 forks)batch import
bughelp wanted

Description

Already queued messages are lost when a SocketDisconnectError is detected here.

This is occurring because the Producer._update method first closes all OwnedBrokers here, which results in the OwnedBroker setting self.running to False here. Then, Producer._update calls Producer._setup_owned_brokers to reestablish the broker connection.

Producer._setup_owned_brokers (which also does it's own OwnedBroker.stop call) calls OwnedBrokers.flush to collect all currently queued messages here.

The attempt to collect all currently queued messages will never return any messages, because OwnedBroker.flush checks to see if self.running is False and if so, returns an empty list here. self.running will be False, because that OwnedBroker was stopped.

I discovered this in a long running test that in the course of queuing 3,200,000 messages, a couple of dozen messages would go missing.

Contributor guide

Tech stack
pythonkafka
Domain
backend
Issue type
bug
DifficultyEstimated implementation difficulty for a new contributor, from 1 for very small changes to 5 for expert-level work.
3
Estimated timeA rough time range for an experienced contributor to investigate, implement, test, and prepare a pull request.
half day
Activity statusHow available the issue appears right now: fresh, active, stale, blocked, or waiting on maintainer input.
stale
ClarityHow clearly the issue explains the expected change, acceptance criteria, and next step.
clear
Prerequisites
Basic PythonKafka concepts
Newbie friendlinessA 1-100 score estimating how approachable this issue is for first-time contributors.
40
Research direction
Investigate the Producer. update method in pykafka/producer.py (lines 274-299) which first stops OwnedBrokers then attempts to flush them. The OwnedBroker.flush method (line 698) returns empty list when self.running is False. Consider modifying the order: flush messages before stopping the OwnedBroker, or save messages in a buffer before stopping. Ensure that messages queued are not lost during reconnection. Refer to linked PRs or comments for prior attempts.