adixitconfluent commented on code in PR #23911:
URL: https://github.com/apache/kafka/pull/23911#discussion_r4350440203
##########
core/src/test/java/kafka/server/share/DelayedShareFetchTest.java:
##########
@@ -222,12 +221,9 @@ public void
testTryCompleteWhenMinBytesNotSatisfiedOnFirstFetch() {
// We are testing the case when the share partition is getting fetched
for the first time, so for the first time
// the fetchOffsetMetadata will return empty. Post the first
readFromLog call, the fetchOffsetMetadata will be
Review Comment:
this comment needs update - fetchOffsetMetadata is no longer populated for
the share partition
##########
core/src/test/java/kafka/server/share/DelayedShareFetchTest.java:
##########
@@ -327,6 +322,8 @@ public void
testTryCompleteWhenMinBytesNotSatisfiedOnSubsequentFetch() {
.withFetchId(fetchId)
.build());
+ // The share partition has been fetched before, hence the fetch offset
metadata is already cached.
Review Comment:
similarly for this comment, it needs an update
##########
core/src/test/java/kafka/server/share/SharePartitionManagerTest.java:
##########
@@ -3268,14 +3265,30 @@ static Seq<Tuple2<TopicIdPartition, LogReadResult>>
buildLogReadResult(List<Topi
Compression.NONE, new SimpleRecord("test-key".getBytes(),
"test-value".getBytes())));
}
+ static Seq<Tuple2<TopicIdPartition, LogReadResult>> buildLogReadResult(
+ List<TopicIdPartition> topicIdPartitions,
+ Map<TopicIdPartition, LogOffsetMetadata> fetchOffsetMetadata
+ ) {
+ return buildLogReadResult(topicIdPartitions, MemoryRecords.withRecords(
+ Compression.NONE, new SimpleRecord("test-key".getBytes(),
"test-value".getBytes())), fetchOffsetMetadata);
+ }
+
static Seq<Tuple2<TopicIdPartition, LogReadResult>>
buildEmptyLogReadResult(List<TopicIdPartition> topicIdPartitions) {
return buildLogReadResult(topicIdPartitions, MemoryRecords.EMPTY);
}
static Seq<Tuple2<TopicIdPartition, LogReadResult>>
buildLogReadResult(List<TopicIdPartition> topicIdPartitions, Records records) {
+ return buildLogReadResult(topicIdPartitions, records, Map.of());
+ }
+
+ static Seq<Tuple2<TopicIdPartition, LogReadResult>> buildLogReadResult(
+ List<TopicIdPartition> topicIdPartitions,
+ Records records,
+ Map<TopicIdPartition, LogOffsetMetadata> fetchOffsetMetadata
+ ) {
List<Tuple2<TopicIdPartition, LogReadResult>> logReadResults = new
ArrayList<>();
topicIdPartitions.forEach(topicIdPartition -> logReadResults.add(new
Tuple2<>(topicIdPartition, new LogReadResult(
- new FetchDataInfo(new LogOffsetMetadata(0, 0, 0), records),
+ new
FetchDataInfo(fetchOffsetMetadata.getOrDefault(topicIdPartition, new
LogOffsetMetadata(0, 0, 0)), records),
Review Comment:
This seems to be a problem. The PR quietly stopped testing the branch of
`isMinBytesSatisfied` that actually counts bytes against fetch.min.bytes.
The three branches ->
For each partition,
[isMinBytesSatisfied](https://github.com/apache/kafka/blob/ae962680085cd125f52ef7b0d466b176c4048a0a/core/src/main/java/kafka/server/share/DelayedShareFetch.java#L594-L638)
compares the cached fetch-offset metadata F with the end-offset (HWM) metadata
E:
1. F is past E: complete immediately.
2. F is on an older segment than E: complete immediately, without counting
bytes. This is the “older-segment shortcut”.
3. F and E on the same segment: add min(E.position − F.position,
partitionMaxBytes) to a running total. At the end, complete only if total >=
minBytes.
Branch 3 is what implements fetch.min.bytes.
The PR changed where F comes from:
trunk: F came from a mocked `sp.fetchOffsetMetadata(...)`, returning (offset
0, segment 1, position 0). The helper
mockTopicIdPartitionToReturnDataEqualToMinBytes put E at (offset 1, segment 1,
position minBytes). Same segment, so the tests counted exactly minBytes bytes
and completed right at the threshold, which is what the helper is named for.
PR: that mock is gone, and F now comes from the log read result.
buildLogReadResult’s default is [(0, segment 0,
0)](https://github.com/apache/kafka/blob/ae962680085cd125f52ef7b0d466b176c4048a0a/core/src/test/java/kafka/server/share/SharePartitionManagerTest.java#L3291).
Segment 0 is older than segment 1, so isMinBytesSatisfied returns true through
branch 2 before it counts anything.
Seven tests changed path this way. None is named for minBytes, which is how
the loss went unnoticed:
DelayedShareFetchTest:
testDelayedShareFetchTryCompleteReturnsTrue
both testPartitionMaxBytesFromUniformStrategy
testTryCompleteAsyncCompletion
SharePartitionManagerTest:
testAcknowledgeCompletesDelayedShareFetchRequest
testReleaseSessionCompletesDelayedShareFetchRequest
testSharePartitionPartialInitializationFailure
On trunk they were the suite’s only positive tests of branch 3. They still
pass on the PR, just through branch 2.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
