KAFKA-17351: Improved handling of compacted topics in share partition (2/N) #19010
+435
−50
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
github-actions
bot
added
triage
AndrewJSchofield
requested changes
Feb 23, 2025
| * | ||
| * @param memberId The member id of the client that is fetching the record. | ||
| * @param batchSize The number of records per acquired records batch. | ||
| * @param maxFetchRecords The maximum number of records that should be acquired, this is a soft | ||
| * limit and the method might acquire more records than the maxFetchRecords, | ||
| * if the records are already part of the same fetch batch. | ||
| * * @param fetchOffset The fetch offset for which the records are fetched. | ||
| * @param fetchOffset The fetch offset for which the records are fetched. |
There was a problem hiding this comment.
There's now an out-of-date comment on the declaration of fetchOffset. You've now started to use the parameter.
| * @param map The map containing the in-flight records. | ||
| * @return A boolean which indicates whether any record is archived or not. | ||
| */ | ||
| private boolean archiveAvailableRecords(long startOffset, long endOffset, NavigableMap<Long, InFlightBatch> map) { | ||
| lock.writeLock().lock(); |
There was a problem hiding this comment.
Just observing that all callers already hold the write lock.
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.
The PR handles fetch for
compactedtopics. The fix was required only when complete batch disappears from the topic log, and same batch is marked re-available in Share Partition state cache. Subsequent log reads will not result the disappeared batch in read response hence respective batch will be left as avaialable in the state cache.The PR checks for the first fetched/read batch base offset and if it's greater than the position from where the read occurred (fetch offset) then if there exists any
availablebatches in the state cache then they will be archived.Adds following test cases:
testAcquireAndAcknowledgeWithRecordsAheadOfRecordBatchStartOffset- Added to verify the existing code for compacted topics.testAcquireWhenBatchesAreRemovedFromBetweenInSubsequentFetchData- Added to verify the existing code for compacted topics.testAcquireWhenBatchesRemovedForFetchOffset- Verifies the changes in the current PR.testAcquireWhenBatchesRemovedForFetchOffsetWithinBatch- Verifies the changes in the current PR.testAcquireWhenBatchesRemovedForFetchOffsetForSameCachedBatch- Verifies the changes in the current PR.