You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "apoorvmittal10 (via GitHub)" <gi...@apache.org> on 2025/07/28 19:00:47 UTC
[PR] KAFKA-19555: Restrict records acquisition post max in-flight limit [kafka]
apoorvmittal10 opened a new pull request, #20253:
URL: https://github.com/apache/kafka/pull/20253
The PR restricts the records being acquired post max-inflight limit. Previously the max in-flight limit was only enforced while considering the share partition for further fetches i.e. once the limit was reached the share partition was not considered for further fetches. However, when the records are actively released then there might be some records being acquired post max-inflight limit. This is evident with higher number of consumers reading from same share partition and releasing the records.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19555: Restrict records acquisition post max in-flight limit [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20253:
URL: https://github.com/apache/kafka/pull/20253#discussion_r2239137888
##########
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, all 10 records should be acquired as within maxInflightMessages limit.
+ 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),
Review Comment:
nit: 0 - > 10
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19555: Restrict records acquisition post max in-flight limit [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20253:
URL: https://github.com/apache/kafka/pull/20253#discussion_r2239135797
##########
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, all 10 records should be acquired as within maxInflightMessages limit.
+ 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),
Review Comment:
Thanks for the patch, LGTM.
nit: 0 -> 10
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19555: Restrict records acquisition post max in-flight limit [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #20253:
URL: https://github.com/apache/kafka/pull/20253#discussion_r2237892481
##########
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:
I have noted this. I wll send a separate patch for this. As I have to change a lot of comments and it will be a big refactor. I am avoiding it in this PR, else the main change of the PR will be hard to determine.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19555: Restrict records acquisition post max in-flight limit [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 merged PR #20253:
URL: https://github.com/apache/kafka/pull/20253
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19555: Restrict records acquisition post max in-flight limit [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19555: Restrict records acquisition post max in-flight limit [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #20253:
URL: https://github.com/apache/kafka/pull/20253#discussion_r2239169994
##########
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, all 10 records should be acquired as within maxInflightMessages limit.
+ 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),
Review Comment:
That's log start offset. And the reason I kept it at 0 is because the previous acquire is from 0th offset. However for share partition this log start offset doesn't matter as of now. We update logStartOffset when `OFFSET_OUT_OF_RANGE` exception is encountered, then we fetch the offset for earliest timestamp.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19555: Restrict records acquisition post max in-flight limit [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #20253:
URL: https://github.com/apache/kafka/pull/20253#discussion_r2237889909
##########
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:
My bad, I didn't correct the comments when later changed the tests. Done.
##########
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:
My bad, I didn't correct the comments when later changed the tests. Done.
##########
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:
My bad, I didn't correct the comments when later changed the tests. Done.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19555: Restrict records acquisition post max in-flight limit [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #20253:
URL: https://github.com/apache/kafka/pull/20253#discussion_r2239135797
##########
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, all 10 records should be acquired as within maxInflightMessages limit.
+ 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),
Review Comment:
Thanks for the patch, LGTM.
nit: 0 -> 10
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org