use destruct method to flush the producer #341
+4
−3
Conversation
This file contains hidden or 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
… binding the class to a callback in terminate
|
@sash can you take a look at the test failures please? |
|
and sorry for the delay on checking this, but GitHub is not notifying me for some reason |
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.
I had issues with the new async producer. Because the reference is captured in the terminate callback, in fact the rdkafka producer is never garbage collcted and because of that in the queue the connection to kafka is never terminated.
That causes issue when used in the queue as new connections indeed created, but the old ones are never closed.
The solution would be to use the destructor. In that way - if you use the main facade method - the reference is kept around for a second publish, and if you do Kafka::fresh(), the message will be flushed when it foes out of scope and then when it is destroyed the connection is closed too. That works great for the Queue because the Facade cache is reset after each job execution! If someone wants to implement a persistent publisher that works cross-jobs - he can create a singleton!