storage: remove kafka producer limits in sinks #24784
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
guswynn
reviewed
Jan 29, 2024
| options.insert("queue.buffering.max.kbytes", "2147483647".into()); | ||
| // Disable the default buffer limit of 100k messages. We don't want to impose any limit | ||
| // here as the operator has nothing better to do with the data than to buffer them. | ||
| options.insert("queue.buffering.max.messages", "0".into()); |
There was a problem hiding this comment.
Do we not have the fix to confluentinc/librdkafka#4018 in our fork? might explain the test failures
The kafka sink operator has nothing better to do with incoming data other than to buffer them, which provides no additional value than to send them to `librdkafka` and let that buffer them instead. The operator is already set up to never buffer messages that are ready to send but the current limits are too conservative. If a large snapshot arrives fast enough then it is possible to reach the 10M message limit and cause the sink to restart. I have observed this happening locally during benchmarking. For this reason this PR disables rdkafka message count and size limits. Signed-off-by: Petros Angelatos <petrosagg@gmail.com>
petrosagg
force-pushed
the
kafka-sink-limits
branch
from
ccd34d2
to
da3d237
Compare
guswynn
approved these changes
Jan 30, 2024
petrosagg
force-pushed
the
kafka-sink-limits
branch
from
da3d237
to
a8f4ee4
Compare
petrosagg
force-pushed
the
kafka-sink-limits
branch
3 times, most recently
from
b29320f
to
79a5814
Compare
Signed-off-by: Petros Angelatos <petrosagg@gmail.com>
petrosagg
force-pushed
the
kafka-sink-limits
branch
from
79a5814
to
83e5b88
Compare
This was referenced Feb 2, 2024
benesch
added a commit
to benesch/materialize
that referenced
this pull request
Jul 1, 2024
Now that we're on librdkafka v2.4.0, we don't need to catch and retry QueueFull errors, but can instead disable the queue limit. This commit is a combination of: * Reverting the QueueFull workaround from MaterializeInc#24871 * Reapplying Petros's original implementation in MaterializeInc#24784 Co-authored-by: Petros Angelatos <petrosagg@gmail.com>
5 tasks
umanwizard
pushed a commit
to umanwizard/materialize-1
that referenced
this pull request
Jul 3, 2024
Now that we're on librdkafka v2.4.0, we don't need to catch and retry QueueFull errors, but can instead disable the queue limit. This commit is a combination of: * Reverting the QueueFull workaround from MaterializeInc#24871 * Reapplying Petros's original implementation in MaterializeInc#24784 Co-authored-by: Petros Angelatos <petrosagg@gmail.com>
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
The kafka sink operator has nothing better to do with incoming data other than to buffer them, which provides no additional value than to send them to
librdkafkaand let that buffer them instead.The operator is already set up to never buffer messages that are ready to send but the current limits are too conservative. If a large snapshot arrives fast enough then it is possible to reach the 10M message limit and cause the sink to restart. I have observed this happening locally during benchmarking.
For this reason this PR disables rdkafka message count and size limits.
Tips for reviewer
Checklist
$T ⇔ Proto$Tmapping (possibly in a backwards-incompatible way), then it is tagged with aT-protolabel.