AndrewJSchofield commented on code in PR #20253:
URL: https://github.com/apache/kafka/pull/20253#discussion_r2237803023
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -1510,6 +1510,150 @@ public void
testAcquireBatchSkipWithBatchSizeAndEndOffsetLargerThanFirstBatch()
assertTrue(sharePartition.cachedState().containsKey(12L));
}
+ @Test
+ public void testAcquireWithMaxInFlightMessagesAndTryAcquireNewBatch() {
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withSharePartitionMetrics(sharePartitionMetrics)
+ .withMaxInflightMessages(20)
+ .build();
+
+ // Acquire records, should be acquired till maxInFlightMessages i.e.
20 records till 29 offset.
Review Comment:
I don't understand the significance of the "20 records till 29 offset" in
the comment. I understand the test.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1449,6 +1461,46 @@ private void maybeUpdateReadGapFetchOffset(long offset) {
}
}
+ private LastOffsetAndMaxRecords lastOffsetAndMaxRecordsToAcquire(long
fetchOffset, int maxFetchRecords, long lastOffset) {
+ // There can always be records fetched exceeding the max in-flight
messages limit. Hence,
+ // we need to check if the share partition has reached the max
in-flight messages limit
+ // and only acquire limited records.
+ int maxRecordsToAcquire;
+ long lastOffsetToAcquire = lastOffset;
+ lock.readLock().lock();
+ try {
+ int inFlightRecordsCount = numInFlightRecords();
+ // Take minimum of maxFetchRecords and remaining capacity to fill
max in-flight messages limit.
+ maxRecordsToAcquire = Math.min(maxFetchRecords,
maxInFlightMessages - inFlightRecordsCount);
Review Comment:
nit: Generally, we use "records" but it's "maxInFlightMessages". Maybe
should rename to "maxInFlightRecords".
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -1510,6 +1510,150 @@ public void
testAcquireBatchSkipWithBatchSizeAndEndOffsetLargerThanFirstBatch()
assertTrue(sharePartition.cachedState().containsKey(12L));
}
+ @Test
+ public void testAcquireWithMaxInFlightMessagesAndTryAcquireNewBatch() {
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withSharePartitionMetrics(sharePartitionMetrics)
+ .withMaxInflightMessages(20)
+ .build();
+
+ // Acquire records, should be acquired till maxInFlightMessages i.e.
20 records till 29 offset.
+ List<AcquiredRecords> acquiredRecordsList =
fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ BATCH_SIZE,
+ 500 /* Max fetch records */,
+ DEFAULT_FETCH_OFFSET,
+ fetchPartitionData(memoryRecords(10, 0), 0),
+ FETCH_ISOLATION_HWM),
+ 10);
+ // Validate all 10 records will be acquired as the maxInFlightMessages
is 20.
+ assertArrayEquals(expectedAcquiredRecord(0, 9, 1).toArray(),
acquiredRecordsList.toArray());
+ assertEquals(10, sharePartition.nextFetchOffset());
+
+ // Create 4 batches of records.
+ ByteBuffer buffer = ByteBuffer.allocate(4096);
+ memoryRecordsBuilder(buffer, 5, 10).close();
+ memoryRecordsBuilder(buffer, 10, 15).close();
+ memoryRecordsBuilder(buffer, 5, 25).close();
+ memoryRecordsBuilder(buffer, 2, 30).close();
+
+ buffer.flip();
+
+ MemoryRecords records = MemoryRecords.readableRecords(buffer);
+
+ // Acquire records, should be acquired till maxInFlightMessages i.e.
20 records. As second batch
+ // is ending at 24 offset, hence additional 15 records will be
acquired.
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ BATCH_SIZE,
+ 500 /* Max fetch records */,
+ DEFAULT_FETCH_OFFSET,
+ fetchPartitionData(records, 0),
+ FETCH_ISOLATION_HWM),
+ 15);
+
+ // Validate 2 batches are fetched one with 5 records and other till
end of batch, third batch
+ // should be skipped.
+ assertArrayEquals(expectedAcquiredRecord(10, 24, 1).toArray(),
acquiredRecordsList.toArray());
+ assertEquals(25, sharePartition.nextFetchOffset());
+
+ // Should not acquire any records as the share partition is at
capacity and fetch offset is beyond
+ // the end offset.
+ fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ BATCH_SIZE,
+ 500 /* Max fetch records */,
+ 25 /* Fetch Offset */,
+ fetchPartitionData(memoryRecords(10, 25), 10),
+ FETCH_ISOLATION_HWM),
+ 0);
+
+ assertEquals(25, sharePartition.nextFetchOffset());
+ }
+
+ @Test
+ public void testAcquireWithMaxInFlightMessagesAndReleaseLastOffset() {
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withSharePartitionMetrics(sharePartitionMetrics)
+ .withMaxInflightMessages(20)
+ .build();
+
+ // Create 3 batches of records.
Review Comment:
nit: I'd say that's 4 batces.
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -1510,6 +1510,150 @@ public void
testAcquireBatchSkipWithBatchSizeAndEndOffsetLargerThanFirstBatch()
assertTrue(sharePartition.cachedState().containsKey(12L));
}
+ @Test
+ public void testAcquireWithMaxInFlightMessagesAndTryAcquireNewBatch() {
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withSharePartitionMetrics(sharePartitionMetrics)
+ .withMaxInflightMessages(20)
+ .build();
+
+ // Acquire records, should be acquired till maxInFlightMessages i.e.
20 records till 29 offset.
+ List<AcquiredRecords> acquiredRecordsList =
fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ BATCH_SIZE,
+ 500 /* Max fetch records */,
+ DEFAULT_FETCH_OFFSET,
+ fetchPartitionData(memoryRecords(10, 0), 0),
+ FETCH_ISOLATION_HWM),
+ 10);
+ // Validate all 10 records will be acquired as the maxInFlightMessages
is 20.
+ assertArrayEquals(expectedAcquiredRecord(0, 9, 1).toArray(),
acquiredRecordsList.toArray());
+ assertEquals(10, sharePartition.nextFetchOffset());
+
+ // Create 4 batches of records.
+ ByteBuffer buffer = ByteBuffer.allocate(4096);
+ memoryRecordsBuilder(buffer, 5, 10).close();
+ memoryRecordsBuilder(buffer, 10, 15).close();
+ memoryRecordsBuilder(buffer, 5, 25).close();
+ memoryRecordsBuilder(buffer, 2, 30).close();
+
+ buffer.flip();
+
+ MemoryRecords records = MemoryRecords.readableRecords(buffer);
+
+ // Acquire records, should be acquired till maxInFlightMessages i.e.
20 records. As second batch
+ // is ending at 24 offset, hence additional 15 records will be
acquired.
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ BATCH_SIZE,
+ 500 /* Max fetch records */,
+ DEFAULT_FETCH_OFFSET,
+ fetchPartitionData(records, 0),
+ FETCH_ISOLATION_HWM),
+ 15);
+
+ // Validate 2 batches are fetched one with 5 records and other till
end of batch, third batch
+ // should be skipped.
+ assertArrayEquals(expectedAcquiredRecord(10, 24, 1).toArray(),
acquiredRecordsList.toArray());
+ assertEquals(25, sharePartition.nextFetchOffset());
+
+ // Should not acquire any records as the share partition is at
capacity and fetch offset is beyond
+ // the end offset.
+ fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ BATCH_SIZE,
+ 500 /* Max fetch records */,
+ 25 /* Fetch Offset */,
+ fetchPartitionData(memoryRecords(10, 25), 10),
+ FETCH_ISOLATION_HWM),
+ 0);
+
+ assertEquals(25, sharePartition.nextFetchOffset());
+ }
+
+ @Test
+ public void testAcquireWithMaxInFlightMessagesAndReleaseLastOffset() {
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withState(SharePartitionState.ACTIVE)
+ .withSharePartitionMetrics(sharePartitionMetrics)
+ .withMaxInflightMessages(20)
+ .build();
+
+ // Create 3 batches of records.
+ ByteBuffer buffer = ByteBuffer.allocate(4096);
+ memoryRecordsBuilder(buffer, 5, 10).close();
+ memoryRecordsBuilder(buffer, 10, 15).close();
+ memoryRecordsBuilder(buffer, 5, 25).close();
+ memoryRecordsBuilder(buffer, 3, 30).close();
+
+ buffer.flip();
+
+ MemoryRecords records = MemoryRecords.readableRecords(buffer);
+ // Acquire records, should be acquired till maxInFlightMessages i.e.
20 records till 29 offset.
+ List<AcquiredRecords> acquiredRecordsList =
fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ BATCH_SIZE,
+ 500 /* Max fetch records */,
+ DEFAULT_FETCH_OFFSET,
+ fetchPartitionData(records, 10),
+ FETCH_ISOLATION_HWM),
+ 20);
+
+ // Validate 2 batches are fetched one with 5 records and other till
end of batch, third batch
Review Comment:
I think 3 batches are fetched. On line 1608, you release the "middle" batch.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
