Allow stream to be interrupted when there is no traffic #1251
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
When a partition is lost while there is no traffic, `PartitionStreamControl` is blocked waiting for data. We fix this by racing with the `interruptPromise`. (Thanks @josdirksen for the analysis!) Note: this situation can only occur with lost partitions. Timeouts (the other reason for interrupts) do not occur when there is no traffic. Currently, we have no way to test lost partitions. Therefore, there are no added tests. See #1250.
svroonland
approved these changes
Jun 10, 2024
There was a problem hiding this comment.
This would indeed fix the situation as described in #1250.
So now the only remaining distinction between handling revoked and lost is that for lost we terminate immediately. Like you mentioned before, I think that lost partitions will not have received any data for a long time anyway. But since we can't test that, this is probably a safe implementation.
zio-kafka/src/main/scala/zio/kafka/consumer/internal/PartitionStreamControl.scala
Show resolved
Hide resolved
svroonland
added a commit
that referenced
this pull request
Nov 10, 2024
…#1371) The issue became apparent from the tests. It unfortunately means that #1251 did not fix the issue of blocking when partitions are lost when there is no more data in the queue. * Fix: race should be raceFirst, because interruptionPromise will only ever fail (relates to #1251) * Reset queue size on lost * Abstract requesting data for testing purposes
|
Note: @svroonland found a bug in this code after adding a unit test. See #1371. |
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.
When a partition is lost while there is no traffic,
PartitionStreamControlis blocked waiting for data. We fix this by racing with theinterruptPromise. (Thanks @josdirksen for the analysis! See #1250.)Note: this situation can only occur with lost partitions. Timeouts (the other reason for interrupts) do not occur when there is no traffic. Currently, we have no way to test lost partitions. Therefore, there are no added tests.
This PR does not change how lost partitions are handled. That is, the stream for the partition that is lost is interrupted, the other streams are closed gracefully, the consumer aborts with an error.