lucasbru commented on code in PR #17414:
URL: https://github.com/apache/kafka/pull/17414#discussion_r1810540186
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchCollectorTest.java:
##########
@@ -620,10 +646,49 @@ 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);
+
+
+ // Initial CompletedFetch
+ CompletedFetch initialCompletedFetch = completedFetchBuilder
Review Comment:
I think you may want to also make the first completedFetch a transactional
completed fetch. See `CompletedFetchTest` for an example. It probably does not
matter whether we include a commit marker or not.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchCollectorTest.java:
##########
@@ -620,10 +646,49 @@ 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);
+
+
+ // Initial CompletedFetch
+ CompletedFetch initialCompletedFetch = completedFetchBuilder
+ .recordCount(recordCount)
+ .build();
+ fetchBuffer.add(initialCompletedFetch);
+ Fetch<String, String> fetch = fetchCollector.collectFetch(fetchBuffer);
+
+ assertFalse(fetch.isEmpty());
+ assertEquals(recordCount, fetch.numRecords());
+ assertEquals(1, fetch.nextOffsets().size());
+ assertEquals(new OffsetAndMetadata(fetch.numRecords(),
Optional.empty(), ""), fetch.nextOffsets().get(topicAPartition0));
Review Comment:
nit: `fetch` is the result we are checking, don't use it on the left side of
the assert, that's a bit confusing. Use `recordCount`
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchCollectorTest.java:
##########
@@ -620,10 +646,49 @@ 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);
+
+
+ // Initial CompletedFetch
+ CompletedFetch initialCompletedFetch = completedFetchBuilder
+ .recordCount(recordCount)
+ .build();
+ fetchBuffer.add(initialCompletedFetch);
+ Fetch<String, String> fetch = fetchCollector.collectFetch(fetchBuffer);
+
+ assertFalse(fetch.isEmpty());
+ assertEquals(recordCount, fetch.numRecords());
+ assertEquals(1, fetch.nextOffsets().size());
+ assertEquals(new OffsetAndMetadata(fetch.numRecords(),
Optional.empty(), ""), fetch.nextOffsets().get(topicAPartition0));
+
+
Review Comment:
nit: double new line (all other tests use single new line)
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchCollectorTest.java:
##########
@@ -620,10 +646,49 @@ 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);
+
+
+ // Initial CompletedFetch
+ CompletedFetch initialCompletedFetch = completedFetchBuilder
+ .recordCount(recordCount)
+ .build();
+ fetchBuffer.add(initialCompletedFetch);
+ Fetch<String, String> fetch = fetchCollector.collectFetch(fetchBuffer);
+
+ assertFalse(fetch.isEmpty());
+ assertEquals(recordCount, fetch.numRecords());
+ assertEquals(1, fetch.nextOffsets().size());
+ assertEquals(new OffsetAndMetadata(fetch.numRecords(),
Optional.empty(), ""), fetch.nextOffsets().get(topicAPartition0));
+
+
+ // Subsequent CompletedFetch with abort marker
+ Records rawRecords =
createTranscactionalRecords(ControlRecordType.ABORT, recordCount);
+ FetchResponseData.PartitionData partitionData = new
FetchResponseData.PartitionData()
+ .setRecords(rawRecords);
Review Comment:
I think for the `FetchResponseData` to be correctly formed, we alos have to
`setAbortedTransactions`. See, for example, `CompletedFetchTest`.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchCollectorTest.java:
##########
@@ -840,7 +931,22 @@ private Records createRecords(final int recordCount) {
for (int i = 0; i < recordCount; i++)
builder.append(0L, "key".getBytes(), ("value-" +
i).getBytes());
- return builder.build();
+ return builder;
+ }
+ }
+
+ private Records createTranscactionalRecords(ControlRecordType
controlRecordType, int recordCount) {
Review Comment:
nit: `Transactional` not `Trasncactional`
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchCollectorTest.java:
##########
@@ -620,10 +646,49 @@ 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);
+
+
Review Comment:
nit: double new line (all other tests use single new line)
--
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]
