MultiProcess consumer #35
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Collaborator
mahendra
commented
Jun 27, 2013
mahendra
commented
Currently the kafka SimpleConsumer consumes messages from all partitions. This commit will ensure that data is consumed only from partitions specified during init
The implementation is done by using simple options to Kafka Fetch Request Also in the SimpleConsumer iterator, update the offset before the message is yielded. This is so that the consumer state is not lost if certain cases. For eg: the message is yielded and consumed by the caller, but the caller does not come back into the generator again. The message will be consumed but the status is not updated in the consumer
Other changes
* Put a message size restriction on the shared queue
- to prevent message overload
* Wait for a while after each process is started (in constructor)
* Wait for a while in each child if the consumer does not return any messages
Just to be nice to the CPU.
* Control the start event more granularly - this prevents infinite loops
if the control does not return to the generator. For eg:
for msg in consumer:
assert False
* Update message status before yield
Conflicts: kafka/consumer.py
This was hidden because of another bug in offset management
|
@mumrah Hey, once this, #33 and #36 and are done, we can easily add gevent support. Matter of 20-30 lines of code. Basically, if you see, we are avoiding use of threads completely. With these three, we are using Queue() and Event() and Pool() exclusively. |
|
This will take care of #30 also |
In the current patch get_messages(count=1) would return zero messages the first time it is invoked after a consumer was initialized.
|
Merged manually, tests passing. Thanks again, @mahendra! |
|
cool. thanks for reviewing @mumrah . just got back to office. will work on updating other pull requests. |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.