AndrewJSchofield commented on code in PR #17414:
URL: https://github.com/apache/kafka/pull/17414#discussion_r1812442065
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchCollectorTest.java:
##########
@@ -620,10 +650,61 @@ public void
testCollectFetchInitializationOffsetOutOfRangeErrorWithOffsetReset()
final Fetch<String, String> fetch =
fetchCollector.collectFetch(fetchBuffer);
assertTrue(fetch.isEmpty());
+ assertEquals(0, fetch.nextOffsets().size());
verify(subscriptions).requestOffsetResetIfPartitionAssigned(topicPartition0);
verify(fetchBuffer).setNextInLineFetch(null);
}
+ @Test
+ public void testReadCommittedWithAbortedTransaction() {
+ buildDependencies(IsolationLevel.READ_COMMITTED);
+ int recordCount = 20;
+ assignAndSeek(topicAPartition0);
+
+ /* The first CompletedFetch object */
+ Records rawRecords =
createTransactionalRecords(ControlRecordType.ABORT, true, 0, recordCount);
+ FetchResponseData.PartitionData partitionData = new
FetchResponseData.PartitionData()
+ .setRecords(rawRecords)
+ .setAbortedTransactions(createAbortedTransactions());
+ CompletedFetch completedFetch1 = completedFetchBuilder
+ .partitionData(partitionData)
+ .build();
+ fetchBuffer.add(completedFetch1);
+ Fetch<String, String> fetch = fetchCollector.collectFetch(fetchBuffer);
+
+ // The Fetch object contains no data record but the offset is moved
forward
+ assertFalse(fetch.isEmpty());
+ assertEquals(0, fetch.numRecords());
+ assertEquals(1, fetch.nextOffsets().size());
+ assertEquals(new OffsetAndMetadata(recordCount + 1, Optional.of(0),
""), fetch.nextOffsets().get(topicAPartition0));
+
+ /* The second CompletedFetch object */
+ int startOffset = recordCount + 1;
+ rawRecords = createTransactionalRecords(ControlRecordType.ABORT,
false, startOffset, recordCount);
+ partitionData = new FetchResponseData.PartitionData()
+ .setRecords(rawRecords)
+ .setAbortedTransactions(createAbortedTransactions());
+ CompletedFetch completedFetch2 = completedFetchBuilder
+ .partitionData(partitionData)
+ .fetchOffset(startOffset)
+ .build();
+ fetchBuffer.add(completedFetch2);
+ fetch = fetchCollector.collectFetch(fetchBuffer);
+
+ // The Fetch object contains both data records and the advanced offset
+ assertFalse(fetch.isEmpty());
+ assertEquals(recordCount, fetch.numRecords());
+ assertEquals(1, fetch.nextOffsets().size());
+ assertEquals(new OffsetAndMetadata(startOffset + recordCount + 1,
Optional.of(0), ""), fetch.nextOffsets().get(topicAPartition0));
+ }
+
+ private List<FetchResponseData.AbortedTransaction>
createAbortedTransactions() {
Review Comment:
I think `createAbortedTransaction` (singular) would be better. Also, I think
you need to supply the `firstOffset` as a parameter which should be the offset
of the first record of the transaction which is being aborted.
--
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]
