feat: allow AsyncProducer to have MaxOpenRequests inflight produce requests per broker #1686
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
|
@bai install dependencies failed, pls check and run again |
|
Could you please rebase your branch off of latest master? 🙏 |
This comment has been minimized.
This comment has been minimized.
ghost
added
stale
bai
force-pushed
the
shopify-master
branch
2 times, most recently
from
0778cd0 to
d6201da
Compare
|
I rebased this PR off of latest master. |
|
@dnwe Do you think it's something we want to get merged? Asking since this PR's been open for a while. |
dnwe
changed the title
dnwe
changed the title
|
@bai thanks for the reminder — I reviewed the changes and they seemed sensible, essentially in-lining the existing sendAndReceive promise-based approach but modifying it to allow for buffering MaxOpenRequests. I did some quick benchmarking in the regular case (when brokers are responsive) and confirmed no noticable dip in producer throughput, so I think it is safe to merge this on the assumption that it is only likely to help with a small uptick in memory consumption if brokers are slow to response to the produce requests |
dnwe
approved these changes
Jan 12, 2022
This was referenced Jan 19, 2022
This was referenced Mar 16, 2022
dnwe
pushed a commit
that referenced
this pull request
Mar 30, 2022
…2182) * producer: ensure that the management message (fin) is never "leaked" Since async producer now support multiple inflight messages thanks to #1686 and #2094, it now may "leak" the "fin" internal management message to Kafka (and to the client) when broker producer is reconnecting to Kafka broker and retries multiple inflight messages at the same time. * test:async-producer: test broker restart (this fixes #2150) * tests:async-producer: disable logger in TestAsyncProducerBrokerRestart * tests:async-producer: protect leader with a mutex to make race detector happy * test:async-producer: set 5mn default finish timeout * async-producer: do not clear bp.currentRetries when fin message is received just after syn * async-producer: use debug logger when fin message is handled for a healthy brokerProducer * test:async-producer:restart: make emptyValues atomic to avoid races * test:async-producer:restart: rename produceRequestTest to countRecordsWithEmptyValue * test:async-producer:restart: reduce retry backoff timeout to speed up the test * test:async-producer:restart: remove bogus 0
This was referenced Apr 19, 2022
Merged
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.
MOTIVATION
now, producer use method
sendAndReceiveof Broker to produceMessage, which is blocking model for next send, producer will wait too long when broker slows down, which cause later request waste too much time.CHANGE
we split produce and receive method, proudcer send proudcer produceSet one by one, and will push promise to pending channel to wait, so prodcuer will not block on produce, which decrease the cost of next send request. pending channel is consumed one by one too, because the request/response is in order, so next response always arrives later