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/01/09 11:24:00 UTC
[PR] KAFKA-18452: Implemented batch size in acquired records [kafka]
apoorvmittal10 opened a new pull request, #18459:
URL: https://github.com/apache/kafka/pull/18459
The PR implements a way to divide acquired batches into batch size as desirable by client.
The BatchSize is the soft limit and should allign the batches in response and cached state in broker at the log batch boundaries i.e.
```
say produce log batch is from 0-50, 51-100
for batch size from client - 60: then response will be 0-100
for batch size from client - 40: then response will be 0-50, 51-100
```
This means that broker will not break the batches rather align them in response as per desirable batch size from client so complete batches are acknowledged which shall avoid offset based tracking on broker.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-18452: Implemented batch size in acquired records [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1913284564
##########
core/src/test/java/kafka/server/share/SharePartitionManagerTest.java:
##########
@@ -146,6 +146,7 @@ public class SharePartitionManagerTest {
private static final short MAX_FETCH_RECORDS = 500;
private static final int DELAYED_SHARE_FETCH_MAX_WAIT_MS = 2000;
private static final int DELAYED_SHARE_FETCH_TIMEOUT_MS = 3000;
+ private static final int BATCH_SIZE = 500;
Review Comment:
I don't think it's necessary to do that in tests, it could be for individual classes itself.
--
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-18452: Implemented batch size in acquired records [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao merged PR #18459:
URL: https://github.com/apache/kafka/pull/18459
--
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-18452: Implemented batch size in acquired records [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1915206149
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1176,11 +1176,12 @@ private boolean initializedOrThrowException() {
};
}
- private AcquiredRecords acquireNewBatchRecords(
+ private ShareAcquiredRecords acquireNewBatchRecords(
String memberId,
Iterable<? extends RecordBatch> batches,
long firstOffset,
long lastOffset,
+ int batchSize,
Review Comment:
And not splitting batches on broker, as partial acknowledgement for a large batch results in offset based tracking for InflightBatch, so alligning them as per max.poll.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-18452: Implemented batch size in acquired records [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1915367062
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -1104,6 +1116,120 @@ public void testAcquireWithEmptyFetchRecords() {
assertEquals(0, sharePartition.nextFetchOffset());
}
+ @Test
+ public void testAcquireWithBatchSizeAndSingleBatch() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Single batch has more records than batch size. Hence, only a single batch exceeding the batch size
+ // should be acquired.
+ MemoryRecords records = memoryRecords(5);
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 2 /* Batch size */,
+ 10,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
+
+ assertArrayEquals(expectedAcquiredRecord(0, 4, 1).toArray(), acquiredRecordsList.toArray());
+ assertEquals(5, sharePartition.nextFetchOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ assertEquals(0, sharePartition.cachedState().get(0L).firstOffset());
+ assertEquals(4, sharePartition.cachedState().get(0L).lastOffset());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(0L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(0L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeAndMultipleBatches() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Create 3 batches of records.
+ ByteBuffer buffer = ByteBuffer.allocate(4096);
+ memoryRecordsBuilder(buffer, 5, 2).close();
+ memoryRecordsBuilder(buffer, 5, 10).close();
+ memoryRecordsBuilder(buffer, 7, 15).close();
+ memoryRecordsBuilder(buffer, 6, 22).close();
+ buffer.flip();
+ MemoryRecords records = MemoryRecords.readableRecords(buffer);
+
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 5 /* Batch size */,
+ 100,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 26 /* Gap of 3 records will also be added to first batch */);
+
+ // Fetch expected records from 4 batches, but change the first expected record to include gap offsets.
+ List<AcquiredRecords> expectedAcquiredRecords = expectedAcquiredRecords(records, 1);
+ expectedAcquiredRecords.remove(0);
+ expectedAcquiredRecords.addAll(0, expectedAcquiredRecord(2, 9, 1));
+
+ assertArrayEquals(expectedAcquiredRecords.toArray(), acquiredRecordsList.toArray());
+ assertEquals(28, sharePartition.nextFetchOffset());
+ assertEquals(4, sharePartition.cachedState().size());
+ assertTrue(sharePartition.cachedState().containsKey(2L));
+ assertTrue(sharePartition.cachedState().containsKey(10L));
+ assertTrue(sharePartition.cachedState().containsKey(15L));
+ assertTrue(sharePartition.cachedState().containsKey(22L));
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(2L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(10L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(15L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(22L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(2L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(10L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(15L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(22L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(2L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(10L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(15L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(22L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(2L).offsetState());
+ assertNull(sharePartition.cachedState().get(10L).offsetState());
+ assertNull(sharePartition.cachedState().get(15L).offsetState());
+ assertNull(sharePartition.cachedState().get(22L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeAndMaxFetchRecords() {
Review Comment:
We don't have a test that combines multiple record batches into a single acquired inFlightBatch. Should we add one?
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -580,6 +580,7 @@ public long nextFetchOffset() {
* fetched from the leader.
*
* @param memberId The member id of the client that is fetching the record.
+ * @param batchSize The batch size of the individual acquired records batch.
Review Comment:
Could we add that the unit of batch is record?
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -1104,6 +1116,120 @@ public void testAcquireWithEmptyFetchRecords() {
assertEquals(0, sharePartition.nextFetchOffset());
}
+ @Test
+ public void testAcquireWithBatchSizeAndSingleBatch() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Single batch has more records than batch size. Hence, only a single batch exceeding the batch size
+ // should be acquired.
+ MemoryRecords records = memoryRecords(5);
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 2 /* Batch size */,
+ 10,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
+
+ assertArrayEquals(expectedAcquiredRecord(0, 4, 1).toArray(), acquiredRecordsList.toArray());
+ assertEquals(5, sharePartition.nextFetchOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ assertEquals(0, sharePartition.cachedState().get(0L).firstOffset());
+ assertEquals(4, sharePartition.cachedState().get(0L).lastOffset());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(0L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(0L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeAndMultipleBatches() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Create 3 batches of records.
Review Comment:
Created 4 batches instead of 3.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1204,25 +1206,78 @@ private AcquiredRecords acquireNewBatchRecords(
lastAcquiredOffset = lastOffsetFromBatchWithRequestOffset(batches, firstAcquiredOffset + maxFetchRecords - 1);
}
- // Schedule acquisition lock timeout for the batch.
- AcquisitionLockTimerTask timerTask = scheduleAcquisitionLockTimeout(memberId, firstAcquiredOffset, lastAcquiredOffset);
- // Add the new batch to the in-flight records along with the acquisition lock timeout task for the batch.
- cachedState.put(firstAcquiredOffset, new InFlightBatch(
- memberId,
- firstAcquiredOffset,
- lastAcquiredOffset,
- RecordState.ACQUIRED,
- 1,
- timerTask));
+ // Create batches of acquired records.
+ List<AcquiredRecords> acquiredRecords = createBatches(memberId, batches, firstAcquiredOffset, lastAcquiredOffset, batchSize);
// if the cachedState was empty before acquiring the new batches then startOffset needs to be updated
if (cachedState.firstKey() == firstAcquiredOffset) {
startOffset = firstAcquiredOffset;
}
endOffset = lastAcquiredOffset;
- return new AcquiredRecords()
- .setFirstOffset(firstAcquiredOffset)
+ return new ShareAcquiredRecords(acquiredRecords, (int) (lastAcquiredOffset - firstAcquiredOffset + 1));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private List<AcquiredRecords> createBatches(
+ String memberId,
+ Iterable<? extends RecordBatch> batches,
+ long firstAcquiredOffset,
+ long lastAcquiredOffset,
+ int batchSize
+ ) {
+ lock.writeLock().lock();
+ try {
+ List<AcquiredRecords> result = new ArrayList<>();
+ long currentFirstOffset = firstAcquiredOffset;
+ // No split of batches is required if the batch size is greater than records which
+ // can be acquired, else split the batch into multiple batches.
+ if (lastAcquiredOffset - firstAcquiredOffset + 1 > batchSize) {
+ // The batch is split into multiple batches considering batch size.
+ // Note: Try reading only the baseOffset of the batch and avoid reading the lastOffset
+ // as lastOffset call of RecordBatch is expensive (loads headers).
+ for (RecordBatch batch : batches) {
+ long batchBaseOffset = batch.baseOffset();
+ if (batchBaseOffset < firstAcquiredOffset) {
Review Comment:
It seems that we could eliminate this statement.
--
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-18452: Implemented batch size in acquired records [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1914678819
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1204,25 +1205,91 @@ private AcquiredRecords acquireNewBatchRecords(
lastAcquiredOffset = lastOffsetFromBatchWithRequestOffset(batches, firstAcquiredOffset + maxFetchRecords - 1);
}
- // Schedule acquisition lock timeout for the batch.
- AcquisitionLockTimerTask timerTask = scheduleAcquisitionLockTimeout(memberId, firstAcquiredOffset, lastAcquiredOffset);
- // Add the new batch to the in-flight records along with the acquisition lock timeout task for the batch.
- cachedState.put(firstAcquiredOffset, new InFlightBatch(
- memberId,
- firstAcquiredOffset,
- lastAcquiredOffset,
- RecordState.ACQUIRED,
- 1,
- timerTask));
+ // Create batches of acquired records.
+ List<AcquiredRecords> acquiredRecords = createBatches(batches, firstAcquiredOffset, lastAcquiredOffset, memberId, batchSize);
// if the cachedState was empty before acquiring the new batches then startOffset needs to be updated
if (cachedState.firstKey() == firstAcquiredOffset) {
startOffset = firstAcquiredOffset;
}
endOffset = lastAcquiredOffset;
- return new AcquiredRecords()
- .setFirstOffset(firstAcquiredOffset)
- .setLastOffset(lastAcquiredOffset)
- .setDeliveryCount((short) 1);
+ return new ShareAcquiredRecords(acquiredRecords, (int) (lastAcquiredOffset - firstAcquiredOffset + 1));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private List<AcquiredRecords> createBatches(
+ Iterable<? extends RecordBatch> batches,
+ long firstAcquiredOffset,
+ long lastAcquiredOffset,
+ String memberId,
+ int batchSize
+ ) {
+ lock.writeLock().lock();
+ try {
+ List<AcquiredRecords> result = new ArrayList<>();
+ if (lastAcquiredOffset - firstAcquiredOffset + 1 <= batchSize) {
+ // No split of batches is required as the batch size is greater than records which
+ // can be acquired.
+ result.add(new AcquiredRecords()
+ .setFirstOffset(firstAcquiredOffset)
+ .setLastOffset(lastAcquiredOffset)
+ .setDeliveryCount((short) 1));
+ } else {
+ // The batch is split into multiple batches considering batch size.
+ long currentFirstOffset = firstAcquiredOffset;
+ RecordBatch lastBatch = null;
+ // Try reading only the baseOffset of the batch and avoid reading the lastOffset as
+ // lastOffset call of RecordBatch is expensive (loads headers).
+ for (RecordBatch batch : batches) {
+ long batchBaseOffset = batch.baseOffset();
+ if (batchBaseOffset < firstAcquiredOffset) {
+ continue;
+ }
+
+ // Track last batch for filling the final batch, if required.
+ lastBatch = batch;
Review Comment:
nit: I'd add a blank line here for legibility.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1204,25 +1205,91 @@ private AcquiredRecords acquireNewBatchRecords(
lastAcquiredOffset = lastOffsetFromBatchWithRequestOffset(batches, firstAcquiredOffset + maxFetchRecords - 1);
}
- // Schedule acquisition lock timeout for the batch.
- AcquisitionLockTimerTask timerTask = scheduleAcquisitionLockTimeout(memberId, firstAcquiredOffset, lastAcquiredOffset);
- // Add the new batch to the in-flight records along with the acquisition lock timeout task for the batch.
- cachedState.put(firstAcquiredOffset, new InFlightBatch(
- memberId,
- firstAcquiredOffset,
- lastAcquiredOffset,
- RecordState.ACQUIRED,
- 1,
- timerTask));
+ // Create batches of acquired records.
+ List<AcquiredRecords> acquiredRecords = createBatches(batches, firstAcquiredOffset, lastAcquiredOffset, memberId, batchSize);
// if the cachedState was empty before acquiring the new batches then startOffset needs to be updated
if (cachedState.firstKey() == firstAcquiredOffset) {
startOffset = firstAcquiredOffset;
}
endOffset = lastAcquiredOffset;
- return new AcquiredRecords()
- .setFirstOffset(firstAcquiredOffset)
- .setLastOffset(lastAcquiredOffset)
- .setDeliveryCount((short) 1);
+ return new ShareAcquiredRecords(acquiredRecords, (int) (lastAcquiredOffset - firstAcquiredOffset + 1));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private List<AcquiredRecords> createBatches(
+ Iterable<? extends RecordBatch> batches,
+ long firstAcquiredOffset,
+ long lastAcquiredOffset,
+ String memberId,
+ int batchSize
+ ) {
+ lock.writeLock().lock();
+ try {
+ List<AcquiredRecords> result = new ArrayList<>();
+ if (lastAcquiredOffset - firstAcquiredOffset + 1 <= batchSize) {
+ // No split of batches is required as the batch size is greater than records which
+ // can be acquired.
+ result.add(new AcquiredRecords()
+ .setFirstOffset(firstAcquiredOffset)
+ .setLastOffset(lastAcquiredOffset)
+ .setDeliveryCount((short) 1));
+ } else {
+ // The batch is split into multiple batches considering batch size.
+ long currentFirstOffset = firstAcquiredOffset;
+ RecordBatch lastBatch = null;
+ // Try reading only the baseOffset of the batch and avoid reading the lastOffset as
+ // lastOffset call of RecordBatch is expensive (loads headers).
+ for (RecordBatch batch : batches) {
+ long batchBaseOffset = batch.baseOffset();
+ if (batchBaseOffset < firstAcquiredOffset) {
+ continue;
+ }
+
+ // Track last batch for filling the final batch, if required.
+ lastBatch = batch;
+ // Check if the batch is already past the last acquired offset then break.
+ if (batchBaseOffset > lastAcquiredOffset) {
+ result.add(new AcquiredRecords()
+ .setFirstOffset(currentFirstOffset)
+ .setLastOffset(lastAcquiredOffset)
+ .setDeliveryCount((short) 1));
+ lastBatch = null;
+ break;
+ }
+
+ // Create new batch once the batch size is reached.
+ if (batchBaseOffset - currentFirstOffset >= batchSize) {
+ result.add(new AcquiredRecords()
+ .setFirstOffset(currentFirstOffset)
+ .setLastOffset(batchBaseOffset - 1)
+ .setDeliveryCount((short) 1));
+ currentFirstOffset = batchBaseOffset;
+ }
+ }
+ // Add the last batch if required.
+ if (lastBatch != null) {
+ result.add(new AcquiredRecords()
+ .setFirstOffset(currentFirstOffset)
+ .setLastOffset(lastAcquiredOffset)
+ .setDeliveryCount((short) 1));
+ }
+ }
+
+ result.forEach(record -> {
Review Comment:
The variable `record` is really an instance of `AcquiredRecords`. I suggest renaming it to `acquiredRecords` because `record.firstOffset()` and `record.lastOffset()` doesn't make sense for something is currently named as if it was a single record.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1204,25 +1205,91 @@ private AcquiredRecords acquireNewBatchRecords(
lastAcquiredOffset = lastOffsetFromBatchWithRequestOffset(batches, firstAcquiredOffset + maxFetchRecords - 1);
}
- // Schedule acquisition lock timeout for the batch.
- AcquisitionLockTimerTask timerTask = scheduleAcquisitionLockTimeout(memberId, firstAcquiredOffset, lastAcquiredOffset);
- // Add the new batch to the in-flight records along with the acquisition lock timeout task for the batch.
- cachedState.put(firstAcquiredOffset, new InFlightBatch(
- memberId,
- firstAcquiredOffset,
- lastAcquiredOffset,
- RecordState.ACQUIRED,
- 1,
- timerTask));
+ // Create batches of acquired records.
+ List<AcquiredRecords> acquiredRecords = createBatches(batches, firstAcquiredOffset, lastAcquiredOffset, memberId, batchSize);
// if the cachedState was empty before acquiring the new batches then startOffset needs to be updated
if (cachedState.firstKey() == firstAcquiredOffset) {
startOffset = firstAcquiredOffset;
}
endOffset = lastAcquiredOffset;
- return new AcquiredRecords()
- .setFirstOffset(firstAcquiredOffset)
- .setLastOffset(lastAcquiredOffset)
- .setDeliveryCount((short) 1);
+ return new ShareAcquiredRecords(acquiredRecords, (int) (lastAcquiredOffset - firstAcquiredOffset + 1));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private List<AcquiredRecords> createBatches(
+ Iterable<? extends RecordBatch> batches,
+ long firstAcquiredOffset,
+ long lastAcquiredOffset,
+ String memberId,
Review Comment:
nit: Almost always, `memberId` is the first argument. I suggest making it the first argument here too.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1204,25 +1205,91 @@ private AcquiredRecords acquireNewBatchRecords(
lastAcquiredOffset = lastOffsetFromBatchWithRequestOffset(batches, firstAcquiredOffset + maxFetchRecords - 1);
}
- // Schedule acquisition lock timeout for the batch.
- AcquisitionLockTimerTask timerTask = scheduleAcquisitionLockTimeout(memberId, firstAcquiredOffset, lastAcquiredOffset);
- // Add the new batch to the in-flight records along with the acquisition lock timeout task for the batch.
- cachedState.put(firstAcquiredOffset, new InFlightBatch(
- memberId,
- firstAcquiredOffset,
- lastAcquiredOffset,
- RecordState.ACQUIRED,
- 1,
- timerTask));
+ // Create batches of acquired records.
+ List<AcquiredRecords> acquiredRecords = createBatches(batches, firstAcquiredOffset, lastAcquiredOffset, memberId, batchSize);
// if the cachedState was empty before acquiring the new batches then startOffset needs to be updated
if (cachedState.firstKey() == firstAcquiredOffset) {
startOffset = firstAcquiredOffset;
}
endOffset = lastAcquiredOffset;
- return new AcquiredRecords()
- .setFirstOffset(firstAcquiredOffset)
- .setLastOffset(lastAcquiredOffset)
- .setDeliveryCount((short) 1);
+ return new ShareAcquiredRecords(acquiredRecords, (int) (lastAcquiredOffset - firstAcquiredOffset + 1));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private List<AcquiredRecords> createBatches(
+ Iterable<? extends RecordBatch> batches,
+ long firstAcquiredOffset,
+ long lastAcquiredOffset,
+ String memberId,
+ int batchSize
+ ) {
+ lock.writeLock().lock();
+ try {
+ List<AcquiredRecords> result = new ArrayList<>();
+ if (lastAcquiredOffset - firstAcquiredOffset + 1 <= batchSize) {
+ // No split of batches is required as the batch size is greater than records which
+ // can be acquired.
+ result.add(new AcquiredRecords()
+ .setFirstOffset(firstAcquiredOffset)
+ .setLastOffset(lastAcquiredOffset)
+ .setDeliveryCount((short) 1));
+ } else {
+ // The batch is split into multiple batches considering batch size.
+ long currentFirstOffset = firstAcquiredOffset;
Review Comment:
I don't think that's an improvement, I'm afraid. I find the code easier to follow with the current name.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1204,25 +1205,91 @@ private AcquiredRecords acquireNewBatchRecords(
lastAcquiredOffset = lastOffsetFromBatchWithRequestOffset(batches, firstAcquiredOffset + maxFetchRecords - 1);
}
- // Schedule acquisition lock timeout for the batch.
- AcquisitionLockTimerTask timerTask = scheduleAcquisitionLockTimeout(memberId, firstAcquiredOffset, lastAcquiredOffset);
- // Add the new batch to the in-flight records along with the acquisition lock timeout task for the batch.
- cachedState.put(firstAcquiredOffset, new InFlightBatch(
- memberId,
- firstAcquiredOffset,
- lastAcquiredOffset,
- RecordState.ACQUIRED,
- 1,
- timerTask));
+ // Create batches of acquired records.
+ List<AcquiredRecords> acquiredRecords = createBatches(batches, firstAcquiredOffset, lastAcquiredOffset, memberId, batchSize);
// if the cachedState was empty before acquiring the new batches then startOffset needs to be updated
if (cachedState.firstKey() == firstAcquiredOffset) {
startOffset = firstAcquiredOffset;
}
endOffset = lastAcquiredOffset;
- return new AcquiredRecords()
- .setFirstOffset(firstAcquiredOffset)
- .setLastOffset(lastAcquiredOffset)
- .setDeliveryCount((short) 1);
+ return new ShareAcquiredRecords(acquiredRecords, (int) (lastAcquiredOffset - firstAcquiredOffset + 1));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private List<AcquiredRecords> createBatches(
+ Iterable<? extends RecordBatch> batches,
+ long firstAcquiredOffset,
+ long lastAcquiredOffset,
+ String memberId,
+ int batchSize
+ ) {
+ lock.writeLock().lock();
+ try {
+ List<AcquiredRecords> result = new ArrayList<>();
+ if (lastAcquiredOffset - firstAcquiredOffset + 1 <= batchSize) {
+ // No split of batches is required as the batch size is greater than records which
+ // can be acquired.
+ result.add(new AcquiredRecords()
+ .setFirstOffset(firstAcquiredOffset)
+ .setLastOffset(lastAcquiredOffset)
+ .setDeliveryCount((short) 1));
+ } else {
+ // The batch is split into multiple batches considering batch size.
+ long currentFirstOffset = firstAcquiredOffset;
+ RecordBatch lastBatch = null;
+ // Try reading only the baseOffset of the batch and avoid reading the lastOffset as
+ // lastOffset call of RecordBatch is expensive (loads headers).
+ for (RecordBatch batch : batches) {
+ long batchBaseOffset = batch.baseOffset();
+ if (batchBaseOffset < firstAcquiredOffset) {
+ continue;
+ }
+
+ // Track last batch for filling the final batch, if required.
+ lastBatch = batch;
+ // Check if the batch is already past the last acquired offset then break.
+ if (batchBaseOffset > lastAcquiredOffset) {
+ result.add(new AcquiredRecords()
+ .setFirstOffset(currentFirstOffset)
+ .setLastOffset(lastAcquiredOffset)
+ .setDeliveryCount((short) 1));
Review Comment:
Yes. Agreed.
--
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-18452: Implemented batch size in acquired records [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1914658017
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1176,11 +1176,12 @@ private boolean initializedOrThrowException() {
};
}
- private AcquiredRecords acquireNewBatchRecords(
+ private ShareAcquiredRecords acquireNewBatchRecords(
String memberId,
Iterable<? extends RecordBatch> batches,
long firstOffset,
long lastOffset,
+ int batchSize,
Review Comment:
Yes. That's it.
--
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-18452: Implemented batch size in acquired records [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1915210285
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1204,25 +1205,91 @@ private AcquiredRecords acquireNewBatchRecords(
lastAcquiredOffset = lastOffsetFromBatchWithRequestOffset(batches, firstAcquiredOffset + maxFetchRecords - 1);
}
- // Schedule acquisition lock timeout for the batch.
- AcquisitionLockTimerTask timerTask = scheduleAcquisitionLockTimeout(memberId, firstAcquiredOffset, lastAcquiredOffset);
- // Add the new batch to the in-flight records along with the acquisition lock timeout task for the batch.
- cachedState.put(firstAcquiredOffset, new InFlightBatch(
- memberId,
- firstAcquiredOffset,
- lastAcquiredOffset,
- RecordState.ACQUIRED,
- 1,
- timerTask));
+ // Create batches of acquired records.
+ List<AcquiredRecords> acquiredRecords = createBatches(batches, firstAcquiredOffset, lastAcquiredOffset, memberId, batchSize);
// if the cachedState was empty before acquiring the new batches then startOffset needs to be updated
if (cachedState.firstKey() == firstAcquiredOffset) {
startOffset = firstAcquiredOffset;
}
endOffset = lastAcquiredOffset;
- return new AcquiredRecords()
- .setFirstOffset(firstAcquiredOffset)
- .setLastOffset(lastAcquiredOffset)
- .setDeliveryCount((short) 1);
+ return new ShareAcquiredRecords(acquiredRecords, (int) (lastAcquiredOffset - firstAcquiredOffset + 1));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private List<AcquiredRecords> createBatches(
+ Iterable<? extends RecordBatch> batches,
+ long firstAcquiredOffset,
+ long lastAcquiredOffset,
+ String memberId,
+ int batchSize
+ ) {
+ lock.writeLock().lock();
+ try {
+ List<AcquiredRecords> result = new ArrayList<>();
+ if (lastAcquiredOffset - firstAcquiredOffset + 1 <= batchSize) {
+ // No split of batches is required as the batch size is greater than records which
+ // can be acquired.
+ result.add(new AcquiredRecords()
+ .setFirstOffset(firstAcquiredOffset)
+ .setLastOffset(lastAcquiredOffset)
+ .setDeliveryCount((short) 1));
+ } else {
+ // The batch is split into multiple batches considering batch size.
+ long currentFirstOffset = firstAcquiredOffset;
+ RecordBatch lastBatch = null;
+ // Try reading only the baseOffset of the batch and avoid reading the lastOffset as
+ // lastOffset call of RecordBatch is expensive (loads headers).
+ for (RecordBatch batch : batches) {
+ long batchBaseOffset = batch.baseOffset();
+ if (batchBaseOffset < firstAcquiredOffset) {
+ continue;
+ }
+
+ // Track last batch for filling the final batch, if required.
+ lastBatch = batch;
Review Comment:
Refactored this method and lastBatch is not required now.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1204,25 +1205,91 @@ private AcquiredRecords acquireNewBatchRecords(
lastAcquiredOffset = lastOffsetFromBatchWithRequestOffset(batches, firstAcquiredOffset + maxFetchRecords - 1);
}
- // Schedule acquisition lock timeout for the batch.
- AcquisitionLockTimerTask timerTask = scheduleAcquisitionLockTimeout(memberId, firstAcquiredOffset, lastAcquiredOffset);
- // Add the new batch to the in-flight records along with the acquisition lock timeout task for the batch.
- cachedState.put(firstAcquiredOffset, new InFlightBatch(
- memberId,
- firstAcquiredOffset,
- lastAcquiredOffset,
- RecordState.ACQUIRED,
- 1,
- timerTask));
+ // Create batches of acquired records.
+ List<AcquiredRecords> acquiredRecords = createBatches(batches, firstAcquiredOffset, lastAcquiredOffset, memberId, batchSize);
// if the cachedState was empty before acquiring the new batches then startOffset needs to be updated
if (cachedState.firstKey() == firstAcquiredOffset) {
startOffset = firstAcquiredOffset;
}
endOffset = lastAcquiredOffset;
- return new AcquiredRecords()
- .setFirstOffset(firstAcquiredOffset)
- .setLastOffset(lastAcquiredOffset)
- .setDeliveryCount((short) 1);
+ return new ShareAcquiredRecords(acquiredRecords, (int) (lastAcquiredOffset - firstAcquiredOffset + 1));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private List<AcquiredRecords> createBatches(
+ Iterable<? extends RecordBatch> batches,
+ long firstAcquiredOffset,
+ long lastAcquiredOffset,
+ String memberId,
+ int batchSize
+ ) {
+ lock.writeLock().lock();
+ try {
+ List<AcquiredRecords> result = new ArrayList<>();
+ if (lastAcquiredOffset - firstAcquiredOffset + 1 <= batchSize) {
+ // No split of batches is required as the batch size is greater than records which
+ // can be acquired.
+ result.add(new AcquiredRecords()
+ .setFirstOffset(firstAcquiredOffset)
+ .setLastOffset(lastAcquiredOffset)
+ .setDeliveryCount((short) 1));
+ } else {
+ // The batch is split into multiple batches considering batch size.
+ long currentFirstOffset = firstAcquiredOffset;
+ RecordBatch lastBatch = null;
+ // Try reading only the baseOffset of the batch and avoid reading the lastOffset as
+ // lastOffset call of RecordBatch is expensive (loads headers).
+ for (RecordBatch batch : batches) {
+ long batchBaseOffset = batch.baseOffset();
+ if (batchBaseOffset < firstAcquiredOffset) {
+ continue;
+ }
+
+ // Track last batch for filling the final batch, if required.
+ lastBatch = batch;
+ // Check if the batch is already past the last acquired offset then break.
+ if (batchBaseOffset > lastAcquiredOffset) {
+ result.add(new AcquiredRecords()
+ .setFirstOffset(currentFirstOffset)
+ .setLastOffset(lastAcquiredOffset)
+ .setDeliveryCount((short) 1));
+ lastBatch = null;
+ break;
+ }
+
+ // Create new batch once the batch size is reached.
+ if (batchBaseOffset - currentFirstOffset >= batchSize) {
+ result.add(new AcquiredRecords()
+ .setFirstOffset(currentFirstOffset)
+ .setLastOffset(batchBaseOffset - 1)
+ .setDeliveryCount((short) 1));
+ currentFirstOffset = batchBaseOffset;
+ }
+ }
+ // Add the last batch if required.
+ if (lastBatch != null) {
+ result.add(new AcquiredRecords()
+ .setFirstOffset(currentFirstOffset)
+ .setLastOffset(lastAcquiredOffset)
+ .setDeliveryCount((short) 1));
+ }
+ }
+
+ result.forEach(record -> {
Review Comment:
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-18452: Implemented batch size in acquired records [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1910299036
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1204,25 +1205,91 @@ private AcquiredRecords acquireNewBatchRecords(
lastAcquiredOffset = lastOffsetFromBatchWithRequestOffset(batches, firstAcquiredOffset + maxFetchRecords - 1);
}
- // Schedule acquisition lock timeout for the batch.
- AcquisitionLockTimerTask timerTask = scheduleAcquisitionLockTimeout(memberId, firstAcquiredOffset, lastAcquiredOffset);
- // Add the new batch to the in-flight records along with the acquisition lock timeout task for the batch.
- cachedState.put(firstAcquiredOffset, new InFlightBatch(
- memberId,
- firstAcquiredOffset,
- lastAcquiredOffset,
- RecordState.ACQUIRED,
- 1,
- timerTask));
+ // Create batches of acquired records.
+ List<AcquiredRecords> acquiredRecords = createBatches(batches, firstAcquiredOffset, lastAcquiredOffset, memberId, batchSize);
// if the cachedState was empty before acquiring the new batches then startOffset needs to be updated
if (cachedState.firstKey() == firstAcquiredOffset) {
startOffset = firstAcquiredOffset;
}
endOffset = lastAcquiredOffset;
- return new AcquiredRecords()
- .setFirstOffset(firstAcquiredOffset)
- .setLastOffset(lastAcquiredOffset)
- .setDeliveryCount((short) 1);
+ return new ShareAcquiredRecords(acquiredRecords, (int) (lastAcquiredOffset - firstAcquiredOffset + 1));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private List<AcquiredRecords> createBatches(
+ Iterable<? extends RecordBatch> batches,
+ long firstAcquiredOffset,
+ long lastAcquiredOffset,
+ String memberId,
+ int batchSize
+ ) {
+ lock.writeLock().lock();
+ try {
+ List<AcquiredRecords> result = new ArrayList<>();
+ if (lastAcquiredOffset - firstAcquiredOffset + 1 <= batchSize) {
+ // No split of batches is required as the batch size is greater than records which
+ // can be acquired.
+ result.add(new AcquiredRecords()
+ .setFirstOffset(firstAcquiredOffset)
+ .setLastOffset(lastAcquiredOffset)
+ .setDeliveryCount((short) 1));
+ } else {
+ // The batch is split into multiple batches considering batch size.
+ long currentFirstOffset = firstAcquiredOffset;
Review Comment:
shouldn't the variable name for `currentFirstOffset` be `previousBatchFirstOffset`, sounds more intuitive to me. Wdyt?
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -589,6 +589,7 @@ public long nextFetchOffset() {
@SuppressWarnings("cyclomaticcomplexity") // Consider refactoring to avoid suppression
public ShareAcquiredRecords acquire(
String memberId,
+ int batchSize,
Review Comment:
description of param `batchSize` is missing in the javadoc for this function
--
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-18452: Implemented batch size in acquired records [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1915586393
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -1104,6 +1116,201 @@ public void testAcquireWithEmptyFetchRecords() {
assertEquals(0, sharePartition.nextFetchOffset());
}
+ @Test
+ public void testAcquireWithBatchSizeAndSingleBatch() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Single batch has more records than batch size. Hence, only a single batch exceeding the batch size
+ // should be acquired.
+ MemoryRecords records = memoryRecords(5);
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 2 /* Batch size */,
+ 10,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
+
+ assertArrayEquals(expectedAcquiredRecord(0, 4, 1).toArray(), acquiredRecordsList.toArray());
+ assertEquals(5, sharePartition.nextFetchOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ assertEquals(0, sharePartition.cachedState().get(0L).firstOffset());
+ assertEquals(4, sharePartition.cachedState().get(0L).lastOffset());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(0L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(0L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeAndMultipleBatches() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Create 4 batches of records.
+ ByteBuffer buffer = ByteBuffer.allocate(4096);
+ memoryRecordsBuilder(buffer, 5, 2).close();
+ memoryRecordsBuilder(buffer, 5, 10).close();
+ memoryRecordsBuilder(buffer, 7, 15).close();
+ memoryRecordsBuilder(buffer, 6, 22).close();
+ buffer.flip();
+ MemoryRecords records = MemoryRecords.readableRecords(buffer);
+
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 5 /* Batch size */,
+ 100,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 26 /* Gap of 3 records will also be added to first batch */);
+
+ // Fetch expected records from 4 batches, but change the first expected record to include gap offsets.
+ List<AcquiredRecords> expectedAcquiredRecords = expectedAcquiredRecords(records, 1);
+ expectedAcquiredRecords.remove(0);
+ expectedAcquiredRecords.addAll(0, expectedAcquiredRecord(2, 9, 1));
+
+ assertArrayEquals(expectedAcquiredRecords.toArray(), acquiredRecordsList.toArray());
+ assertEquals(28, sharePartition.nextFetchOffset());
+ assertEquals(4, sharePartition.cachedState().size());
+ assertTrue(sharePartition.cachedState().containsKey(2L));
+ assertTrue(sharePartition.cachedState().containsKey(10L));
+ assertTrue(sharePartition.cachedState().containsKey(15L));
+ assertTrue(sharePartition.cachedState().containsKey(22L));
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(2L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(10L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(15L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(22L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(2L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(10L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(15L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(22L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(2L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(10L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(15L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(22L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(2L).offsetState());
+ assertNull(sharePartition.cachedState().get(10L).offsetState());
+ assertNull(sharePartition.cachedState().get(15L).offsetState());
+ assertNull(sharePartition.cachedState().get(22L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeAndMaxFetchRecords() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Create 3 batches of records.
+ ByteBuffer buffer = ByteBuffer.allocate(4096);
+ memoryRecordsBuilder(buffer, 5, 0).close();
+ memoryRecordsBuilder(buffer, 15, 5).close();
+ memoryRecordsBuilder(buffer, 15, 20).close();
+ buffer.flip();
+ MemoryRecords records = MemoryRecords.readableRecords(buffer);
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 2 /* Batch size */,
+ 10,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 20);
+
+ List<AcquiredRecords> expectedAcquiredRecords = expectedAcquiredRecords(records, 1);
+ // The last batch should be ignored as it exceeds the max fetch records.
+ expectedAcquiredRecords.remove(2);
+
+ assertArrayEquals(expectedAcquiredRecords.toArray(), acquiredRecordsList.toArray());
+ assertEquals(20, sharePartition.nextFetchOffset());
+ assertEquals(2, sharePartition.cachedState().size());
+ assertEquals(0, sharePartition.cachedState().get(0L).firstOffset());
+ assertEquals(4, sharePartition.cachedState().get(0L).lastOffset());
+ assertEquals(5, sharePartition.cachedState().get(5L).firstOffset());
+ assertEquals(19, sharePartition.cachedState().get(5L).lastOffset());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(5L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(5L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(0L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(5L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(0L).offsetState());
+ assertNull(sharePartition.cachedState().get(5L).offsetState());
+ }
+
+ @Test
+ public void testAcquireSingleBatchWithBatchSizeAndEndOffsetAheadBatchFirstOffset() {
Review Comment:
Yeah, that's what I wanted to test. When endOffset is ahed of first batch offset then do we acquire the batches/offsets correctly.
--
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-18452: Implemented batch size in acquired records [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1914650239
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1204,25 +1205,91 @@ private AcquiredRecords acquireNewBatchRecords(
lastAcquiredOffset = lastOffsetFromBatchWithRequestOffset(batches, firstAcquiredOffset + maxFetchRecords - 1);
}
- // Schedule acquisition lock timeout for the batch.
- AcquisitionLockTimerTask timerTask = scheduleAcquisitionLockTimeout(memberId, firstAcquiredOffset, lastAcquiredOffset);
- // Add the new batch to the in-flight records along with the acquisition lock timeout task for the batch.
- cachedState.put(firstAcquiredOffset, new InFlightBatch(
- memberId,
- firstAcquiredOffset,
- lastAcquiredOffset,
- RecordState.ACQUIRED,
- 1,
- timerTask));
+ // Create batches of acquired records.
+ List<AcquiredRecords> acquiredRecords = createBatches(batches, firstAcquiredOffset, lastAcquiredOffset, memberId, batchSize);
// if the cachedState was empty before acquiring the new batches then startOffset needs to be updated
if (cachedState.firstKey() == firstAcquiredOffset) {
startOffset = firstAcquiredOffset;
}
endOffset = lastAcquiredOffset;
- return new AcquiredRecords()
- .setFirstOffset(firstAcquiredOffset)
- .setLastOffset(lastAcquiredOffset)
- .setDeliveryCount((short) 1);
+ return new ShareAcquiredRecords(acquiredRecords, (int) (lastAcquiredOffset - firstAcquiredOffset + 1));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private List<AcquiredRecords> createBatches(
+ Iterable<? extends RecordBatch> batches,
+ long firstAcquiredOffset,
+ long lastAcquiredOffset,
+ String memberId,
+ int batchSize
Review Comment:
We've discussed this. I would like to assess the efficacy of the approach of having the batch size information known on client and broker, and then we can decide on the best way to control it. In practice, almost all of the broker code would be the same whether it's a config or part of the request, so not too difficult to switch if we want later.
--
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-18452: Implemented batch size in acquired records [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1915497293
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -1104,6 +1116,120 @@ public void testAcquireWithEmptyFetchRecords() {
assertEquals(0, sharePartition.nextFetchOffset());
}
+ @Test
+ public void testAcquireWithBatchSizeAndSingleBatch() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Single batch has more records than batch size. Hence, only a single batch exceeding the batch size
+ // should be acquired.
+ MemoryRecords records = memoryRecords(5);
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 2 /* Batch size */,
+ 10,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
+
+ assertArrayEquals(expectedAcquiredRecord(0, 4, 1).toArray(), acquiredRecordsList.toArray());
+ assertEquals(5, sharePartition.nextFetchOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ assertEquals(0, sharePartition.cachedState().get(0L).firstOffset());
+ assertEquals(4, sharePartition.cachedState().get(0L).lastOffset());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(0L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(0L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeAndMultipleBatches() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Create 3 batches of records.
+ ByteBuffer buffer = ByteBuffer.allocate(4096);
+ memoryRecordsBuilder(buffer, 5, 2).close();
+ memoryRecordsBuilder(buffer, 5, 10).close();
+ memoryRecordsBuilder(buffer, 7, 15).close();
+ memoryRecordsBuilder(buffer, 6, 22).close();
+ buffer.flip();
+ MemoryRecords records = MemoryRecords.readableRecords(buffer);
+
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 5 /* Batch size */,
+ 100,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 26 /* Gap of 3 records will also be added to first batch */);
+
+ // Fetch expected records from 4 batches, but change the first expected record to include gap offsets.
+ List<AcquiredRecords> expectedAcquiredRecords = expectedAcquiredRecords(records, 1);
+ expectedAcquiredRecords.remove(0);
+ expectedAcquiredRecords.addAll(0, expectedAcquiredRecord(2, 9, 1));
+
+ assertArrayEquals(expectedAcquiredRecords.toArray(), acquiredRecordsList.toArray());
+ assertEquals(28, sharePartition.nextFetchOffset());
+ assertEquals(4, sharePartition.cachedState().size());
+ assertTrue(sharePartition.cachedState().containsKey(2L));
+ assertTrue(sharePartition.cachedState().containsKey(10L));
+ assertTrue(sharePartition.cachedState().containsKey(15L));
+ assertTrue(sharePartition.cachedState().containsKey(22L));
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(2L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(10L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(15L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(22L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(2L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(10L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(15L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(22L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(2L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(10L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(15L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(22L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(2L).offsetState());
+ assertNull(sharePartition.cachedState().get(10L).offsetState());
+ assertNull(sharePartition.cachedState().get(15L).offsetState());
+ assertNull(sharePartition.cachedState().get(22L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeAndMaxFetchRecords() {
Review Comment:
Thanks. This sounds good then.
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -1104,6 +1116,201 @@ public void testAcquireWithEmptyFetchRecords() {
assertEquals(0, sharePartition.nextFetchOffset());
}
+ @Test
+ public void testAcquireWithBatchSizeAndSingleBatch() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Single batch has more records than batch size. Hence, only a single batch exceeding the batch size
+ // should be acquired.
+ MemoryRecords records = memoryRecords(5);
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 2 /* Batch size */,
+ 10,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
+
+ assertArrayEquals(expectedAcquiredRecord(0, 4, 1).toArray(), acquiredRecordsList.toArray());
+ assertEquals(5, sharePartition.nextFetchOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ assertEquals(0, sharePartition.cachedState().get(0L).firstOffset());
+ assertEquals(4, sharePartition.cachedState().get(0L).lastOffset());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(0L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(0L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeAndMultipleBatches() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Create 4 batches of records.
+ ByteBuffer buffer = ByteBuffer.allocate(4096);
+ memoryRecordsBuilder(buffer, 5, 2).close();
+ memoryRecordsBuilder(buffer, 5, 10).close();
+ memoryRecordsBuilder(buffer, 7, 15).close();
+ memoryRecordsBuilder(buffer, 6, 22).close();
+ buffer.flip();
+ MemoryRecords records = MemoryRecords.readableRecords(buffer);
+
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 5 /* Batch size */,
+ 100,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 26 /* Gap of 3 records will also be added to first batch */);
+
+ // Fetch expected records from 4 batches, but change the first expected record to include gap offsets.
+ List<AcquiredRecords> expectedAcquiredRecords = expectedAcquiredRecords(records, 1);
+ expectedAcquiredRecords.remove(0);
+ expectedAcquiredRecords.addAll(0, expectedAcquiredRecord(2, 9, 1));
+
+ assertArrayEquals(expectedAcquiredRecords.toArray(), acquiredRecordsList.toArray());
+ assertEquals(28, sharePartition.nextFetchOffset());
+ assertEquals(4, sharePartition.cachedState().size());
+ assertTrue(sharePartition.cachedState().containsKey(2L));
+ assertTrue(sharePartition.cachedState().containsKey(10L));
+ assertTrue(sharePartition.cachedState().containsKey(15L));
+ assertTrue(sharePartition.cachedState().containsKey(22L));
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(2L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(10L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(15L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(22L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(2L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(10L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(15L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(22L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(2L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(10L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(15L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(22L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(2L).offsetState());
+ assertNull(sharePartition.cachedState().get(10L).offsetState());
+ assertNull(sharePartition.cachedState().get(15L).offsetState());
+ assertNull(sharePartition.cachedState().get(22L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeAndMaxFetchRecords() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Create 3 batches of records.
+ ByteBuffer buffer = ByteBuffer.allocate(4096);
+ memoryRecordsBuilder(buffer, 5, 0).close();
+ memoryRecordsBuilder(buffer, 15, 5).close();
+ memoryRecordsBuilder(buffer, 15, 20).close();
+ buffer.flip();
+ MemoryRecords records = MemoryRecords.readableRecords(buffer);
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 2 /* Batch size */,
+ 10,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 20);
+
+ List<AcquiredRecords> expectedAcquiredRecords = expectedAcquiredRecords(records, 1);
+ // The last batch should be ignored as it exceeds the max fetch records.
+ expectedAcquiredRecords.remove(2);
+
+ assertArrayEquals(expectedAcquiredRecords.toArray(), acquiredRecordsList.toArray());
+ assertEquals(20, sharePartition.nextFetchOffset());
+ assertEquals(2, sharePartition.cachedState().size());
+ assertEquals(0, sharePartition.cachedState().get(0L).firstOffset());
+ assertEquals(4, sharePartition.cachedState().get(0L).lastOffset());
+ assertEquals(5, sharePartition.cachedState().get(5L).firstOffset());
+ assertEquals(19, sharePartition.cachedState().get(5L).lastOffset());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(5L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(5L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(0L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(5L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(0L).offsetState());
+ assertNull(sharePartition.cachedState().get(5L).offsetState());
+ }
+
+ @Test
+ public void testAcquireSingleBatchWithBatchSizeAndEndOffsetAheadBatchFirstOffset() {
Review Comment:
The endOffset is actually after the firstOffset of the first batch. Ditto in the two other tests below.
--
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-18452: Implemented batch size in acquired records [kafka]
Posted by "adixitconfluent (via GitHub)" <gi...@apache.org>.
adixitconfluent commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1910309874
##########
core/src/test/java/kafka/server/share/SharePartitionManagerTest.java:
##########
@@ -146,6 +146,7 @@ public class SharePartitionManagerTest {
private static final short MAX_FETCH_RECORDS = 500;
private static final int DELAYED_SHARE_FETCH_MAX_WAIT_MS = 2000;
private static final int DELAYED_SHARE_FETCH_TIMEOUT_MS = 3000;
+ private static final int BATCH_SIZE = 500;
Review Comment:
We have defined the same BATCH_SIZE = 500 in SharePartitionTest, DelayedShareFetchTest, ShareFetchUtilsTest and ShareFetchTest, can we reuse it?
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -1104,6 +1116,109 @@ public void testAcquireWithEmptyFetchRecords() {
assertEquals(0, sharePartition.nextFetchOffset());
}
+ @Test
+ public void testAcquireWithBatchSizeAndSingleBatch() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Single batch has more records than batch size. Hence, only a single batch exceeding the batch size
+ // should be acquired.
+ MemoryRecords records = memoryRecords(5);
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 2 /* Batch size */,
+ 10,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
+
+ assertArrayEquals(expectedAcquiredRecord(0, 4, 1).toArray(), acquiredRecordsList.toArray());
+ assertEquals(5, sharePartition.nextFetchOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ assertEquals(0, sharePartition.cachedState().get(0L).firstOffset());
+ assertEquals(4, sharePartition.cachedState().get(0L).lastOffset());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(0L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(0L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeAndMultipleBatches() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Create 3 batches of records.
+ ByteBuffer buffer = ByteBuffer.allocate(4096);
+ memoryRecordsBuilder(buffer, 5, 2).close();
+ memoryRecordsBuilder(buffer, 5, 10).close();
+ memoryRecordsBuilder(buffer, 7, 15).close();
+ memoryRecordsBuilder(buffer, 5, 22).close();
+ buffer.flip();
+ MemoryRecords records = MemoryRecords.readableRecords(buffer);
+
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 5 /* Batch size */,
+ 100,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 25 /* Gap of 3 records will also be added to first batch */);
+
+ // Fetch expected records from 3 batches, but change the first expected record to include gap offsets.
+ List<AcquiredRecords> expectedAcquiredRecords = expectedAcquiredRecords(records, 1);
+ expectedAcquiredRecords.remove(0);
+ expectedAcquiredRecords.addAll(0, expectedAcquiredRecord(2, 9, 1));
+
+ assertArrayEquals(expectedAcquiredRecords.toArray(), acquiredRecordsList.toArray());
+ assertEquals(27, sharePartition.nextFetchOffset());
+ assertEquals(4, sharePartition.cachedState().size());
+ assertTrue(sharePartition.cachedState().containsKey(2L));
+ assertTrue(sharePartition.cachedState().containsKey(10L));
+ assertTrue(sharePartition.cachedState().containsKey(15L));
+ assertTrue(sharePartition.cachedState().containsKey(22L));
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(2L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(10L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(15L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(22L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(2L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(10L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(15L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(22L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(2L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(10L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(15L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(22L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(2L).offsetState());
+ assertNull(sharePartition.cachedState().get(10L).offsetState());
+ assertNull(sharePartition.cachedState().get(15L).offsetState());
+ assertNull(sharePartition.cachedState().get(22L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeWithBatchSize() {
Review Comment:
the test name seems incorrect
--
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-18452: Implemented batch size in acquired records [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1915209614
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -1104,6 +1116,109 @@ public void testAcquireWithEmptyFetchRecords() {
assertEquals(0, sharePartition.nextFetchOffset());
}
+ @Test
+ public void testAcquireWithBatchSizeAndSingleBatch() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Single batch has more records than batch size. Hence, only a single batch exceeding the batch size
+ // should be acquired.
+ MemoryRecords records = memoryRecords(5);
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 2 /* Batch size */,
+ 10,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
+
+ assertArrayEquals(expectedAcquiredRecord(0, 4, 1).toArray(), acquiredRecordsList.toArray());
+ assertEquals(5, sharePartition.nextFetchOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ assertEquals(0, sharePartition.cachedState().get(0L).firstOffset());
+ assertEquals(4, sharePartition.cachedState().get(0L).lastOffset());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(0L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(0L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeAndMultipleBatches() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Create 3 batches of records.
+ ByteBuffer buffer = ByteBuffer.allocate(4096);
+ memoryRecordsBuilder(buffer, 5, 2).close();
+ memoryRecordsBuilder(buffer, 5, 10).close();
+ memoryRecordsBuilder(buffer, 7, 15).close();
+ memoryRecordsBuilder(buffer, 5, 22).close();
+ buffer.flip();
+ MemoryRecords records = MemoryRecords.readableRecords(buffer);
+
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 5 /* Batch size */,
+ 100,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 25 /* Gap of 3 records will also be added to first batch */);
+
+ // Fetch expected records from 3 batches, but change the first expected record to include gap offsets.
Review Comment:
Thanks for pointing out, 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-18452: Implemented batch size in acquired records [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1913952687
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -1104,6 +1116,109 @@ public void testAcquireWithEmptyFetchRecords() {
assertEquals(0, sharePartition.nextFetchOffset());
}
+ @Test
+ public void testAcquireWithBatchSizeAndSingleBatch() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Single batch has more records than batch size. Hence, only a single batch exceeding the batch size
+ // should be acquired.
+ MemoryRecords records = memoryRecords(5);
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 2 /* Batch size */,
+ 10,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
+
+ assertArrayEquals(expectedAcquiredRecord(0, 4, 1).toArray(), acquiredRecordsList.toArray());
+ assertEquals(5, sharePartition.nextFetchOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ assertEquals(0, sharePartition.cachedState().get(0L).firstOffset());
+ assertEquals(4, sharePartition.cachedState().get(0L).lastOffset());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(0L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(0L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeAndMultipleBatches() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Create 3 batches of records.
+ ByteBuffer buffer = ByteBuffer.allocate(4096);
+ memoryRecordsBuilder(buffer, 5, 2).close();
+ memoryRecordsBuilder(buffer, 5, 10).close();
+ memoryRecordsBuilder(buffer, 7, 15).close();
+ memoryRecordsBuilder(buffer, 5, 22).close();
+ buffer.flip();
+ MemoryRecords records = MemoryRecords.readableRecords(buffer);
+
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 5 /* Batch size */,
+ 100,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 25 /* Gap of 3 records will also be added to first batch */);
+
+ // Fetch expected records from 3 batches, but change the first expected record to include gap offsets.
Review Comment:
There are 4 batches, instead of 3.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1176,11 +1176,12 @@ private boolean initializedOrThrowException() {
};
}
- private AcquiredRecords acquireNewBatchRecords(
+ private ShareAcquiredRecords acquireNewBatchRecords(
String memberId,
Iterable<? extends RecordBatch> batches,
long firstOffset,
long lastOffset,
+ int batchSize,
Review Comment:
Currently, maxFetchRecords is the maxPollRecords. What benefits do we get by setting batchSize smaller than maxFetchRecords? To allow acks to be sent quicker?
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -589,6 +589,7 @@ public long nextFetchOffset() {
@SuppressWarnings("cyclomaticcomplexity") // Consider refactoring to avoid suppression
public ShareAcquiredRecords acquire(
String memberId,
+ int batchSize,
Review Comment:
The javadoc for this new param is still missing.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -627,9 +628,8 @@ public ShareAcquiredRecords acquire(
if (subMap.isEmpty()) {
log.trace("No cached data exists for the share partition for requested fetch batch: {}-{}",
groupId, topicIdPartition);
- AcquiredRecords acquiredRecords = acquireNewBatchRecords(memberId, fetchPartitionData.records.batches(),
- firstBatch.baseOffset(), lastBatch.lastOffset(), maxFetchRecords);
- return ShareAcquiredRecords.fromAcquiredRecords(acquiredRecords);
+ return acquireNewBatchRecords(memberId, fetchPartitionData.records.batches(),
Review Comment:
This is an existing issue. In shareFetchResponse, we always set records to all fetch batches. However, some of the batches could be excluded during acquisition. So, it's wasteful to return excluded batches in shareFetchResponse.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1204,25 +1205,91 @@ private AcquiredRecords acquireNewBatchRecords(
lastAcquiredOffset = lastOffsetFromBatchWithRequestOffset(batches, firstAcquiredOffset + maxFetchRecords - 1);
}
- // Schedule acquisition lock timeout for the batch.
- AcquisitionLockTimerTask timerTask = scheduleAcquisitionLockTimeout(memberId, firstAcquiredOffset, lastAcquiredOffset);
- // Add the new batch to the in-flight records along with the acquisition lock timeout task for the batch.
- cachedState.put(firstAcquiredOffset, new InFlightBatch(
- memberId,
- firstAcquiredOffset,
- lastAcquiredOffset,
- RecordState.ACQUIRED,
- 1,
- timerTask));
+ // Create batches of acquired records.
+ List<AcquiredRecords> acquiredRecords = createBatches(batches, firstAcquiredOffset, lastAcquiredOffset, memberId, batchSize);
// if the cachedState was empty before acquiring the new batches then startOffset needs to be updated
if (cachedState.firstKey() == firstAcquiredOffset) {
startOffset = firstAcquiredOffset;
}
endOffset = lastAcquiredOffset;
- return new AcquiredRecords()
- .setFirstOffset(firstAcquiredOffset)
- .setLastOffset(lastAcquiredOffset)
- .setDeliveryCount((short) 1);
+ return new ShareAcquiredRecords(acquiredRecords, (int) (lastAcquiredOffset - firstAcquiredOffset + 1));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private List<AcquiredRecords> createBatches(
+ Iterable<? extends RecordBatch> batches,
+ long firstAcquiredOffset,
+ long lastAcquiredOffset,
+ String memberId,
+ int batchSize
+ ) {
+ lock.writeLock().lock();
+ try {
+ List<AcquiredRecords> result = new ArrayList<>();
+ if (lastAcquiredOffset - firstAcquiredOffset + 1 <= batchSize) {
+ // No split of batches is required as the batch size is greater than records which
+ // can be acquired.
+ result.add(new AcquiredRecords()
+ .setFirstOffset(firstAcquiredOffset)
+ .setLastOffset(lastAcquiredOffset)
+ .setDeliveryCount((short) 1));
+ } else {
+ // The batch is split into multiple batches considering batch size.
+ long currentFirstOffset = firstAcquiredOffset;
+ RecordBatch lastBatch = null;
+ // Try reading only the baseOffset of the batch and avoid reading the lastOffset as
+ // lastOffset call of RecordBatch is expensive (loads headers).
+ for (RecordBatch batch : batches) {
+ long batchBaseOffset = batch.baseOffset();
+ if (batchBaseOffset < firstAcquiredOffset) {
+ continue;
+ }
+
+ // Track last batch for filling the final batch, if required.
+ lastBatch = batch;
+ // Check if the batch is already past the last acquired offset then break.
+ if (batchBaseOffset > lastAcquiredOffset) {
+ result.add(new AcquiredRecords()
+ .setFirstOffset(currentFirstOffset)
+ .setLastOffset(lastAcquiredOffset)
+ .setDeliveryCount((short) 1));
Review Comment:
Could we just break here since the code outside the loop does the same thing?
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -244,13 +244,14 @@ public CompletableFuture<Map<TopicIdPartition, PartitionData>> fetchMessages(
String groupId,
String memberId,
FetchParams fetchParams,
+ int batchSize,
Review Comment:
missing javadoc for the new param
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1204,25 +1205,91 @@ private AcquiredRecords acquireNewBatchRecords(
lastAcquiredOffset = lastOffsetFromBatchWithRequestOffset(batches, firstAcquiredOffset + maxFetchRecords - 1);
}
- // Schedule acquisition lock timeout for the batch.
- AcquisitionLockTimerTask timerTask = scheduleAcquisitionLockTimeout(memberId, firstAcquiredOffset, lastAcquiredOffset);
- // Add the new batch to the in-flight records along with the acquisition lock timeout task for the batch.
- cachedState.put(firstAcquiredOffset, new InFlightBatch(
- memberId,
- firstAcquiredOffset,
- lastAcquiredOffset,
- RecordState.ACQUIRED,
- 1,
- timerTask));
+ // Create batches of acquired records.
+ List<AcquiredRecords> acquiredRecords = createBatches(batches, firstAcquiredOffset, lastAcquiredOffset, memberId, batchSize);
// if the cachedState was empty before acquiring the new batches then startOffset needs to be updated
if (cachedState.firstKey() == firstAcquiredOffset) {
startOffset = firstAcquiredOffset;
}
endOffset = lastAcquiredOffset;
- return new AcquiredRecords()
- .setFirstOffset(firstAcquiredOffset)
- .setLastOffset(lastAcquiredOffset)
- .setDeliveryCount((short) 1);
+ return new ShareAcquiredRecords(acquiredRecords, (int) (lastAcquiredOffset - firstAcquiredOffset + 1));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private List<AcquiredRecords> createBatches(
+ Iterable<? extends RecordBatch> batches,
+ long firstAcquiredOffset,
+ long lastAcquiredOffset,
+ String memberId,
+ int batchSize
Review Comment:
This is an existing issue. Will it be better for BatchSize to be a group level config instead of in each share fetch request?
--
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-18452: Implemented batch size in acquired records [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1915462216
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1204,25 +1206,78 @@ private AcquiredRecords acquireNewBatchRecords(
lastAcquiredOffset = lastOffsetFromBatchWithRequestOffset(batches, firstAcquiredOffset + maxFetchRecords - 1);
}
- // Schedule acquisition lock timeout for the batch.
- AcquisitionLockTimerTask timerTask = scheduleAcquisitionLockTimeout(memberId, firstAcquiredOffset, lastAcquiredOffset);
- // Add the new batch to the in-flight records along with the acquisition lock timeout task for the batch.
- cachedState.put(firstAcquiredOffset, new InFlightBatch(
- memberId,
- firstAcquiredOffset,
- lastAcquiredOffset,
- RecordState.ACQUIRED,
- 1,
- timerTask));
+ // Create batches of acquired records.
+ List<AcquiredRecords> acquiredRecords = createBatches(memberId, batches, firstAcquiredOffset, lastAcquiredOffset, batchSize);
// if the cachedState was empty before acquiring the new batches then startOffset needs to be updated
if (cachedState.firstKey() == firstAcquiredOffset) {
startOffset = firstAcquiredOffset;
}
endOffset = lastAcquiredOffset;
- return new AcquiredRecords()
- .setFirstOffset(firstAcquiredOffset)
+ return new ShareAcquiredRecords(acquiredRecords, (int) (lastAcquiredOffset - firstAcquiredOffset + 1));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private List<AcquiredRecords> createBatches(
+ String memberId,
+ Iterable<? extends RecordBatch> batches,
+ long firstAcquiredOffset,
+ long lastAcquiredOffset,
+ int batchSize
+ ) {
+ lock.writeLock().lock();
+ try {
+ List<AcquiredRecords> result = new ArrayList<>();
+ long currentFirstOffset = firstAcquiredOffset;
+ // No split of batches is required if the batch size is greater than records which
+ // can be acquired, else split the batch into multiple batches.
+ if (lastAcquiredOffset - firstAcquiredOffset + 1 > batchSize) {
+ // The batch is split into multiple batches considering batch size.
+ // Note: Try reading only the baseOffset of the batch and avoid reading the lastOffset
+ // as lastOffset call of RecordBatch is expensive (loads headers).
+ for (RecordBatch batch : batches) {
+ long batchBaseOffset = batch.baseOffset();
+ if (batchBaseOffset < firstAcquiredOffset) {
Review Comment:
That is so right, I have added multiple test to validate skipping in different scenarios.
--
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-18452: Implemented batch size in acquired records [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1915587034
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -1104,6 +1116,201 @@ public void testAcquireWithEmptyFetchRecords() {
assertEquals(0, sharePartition.nextFetchOffset());
}
+ @Test
+ public void testAcquireWithBatchSizeAndSingleBatch() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Single batch has more records than batch size. Hence, only a single batch exceeding the batch size
+ // should be acquired.
+ MemoryRecords records = memoryRecords(5);
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 2 /* Batch size */,
+ 10,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
+
+ assertArrayEquals(expectedAcquiredRecord(0, 4, 1).toArray(), acquiredRecordsList.toArray());
+ assertEquals(5, sharePartition.nextFetchOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ assertEquals(0, sharePartition.cachedState().get(0L).firstOffset());
+ assertEquals(4, sharePartition.cachedState().get(0L).lastOffset());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(0L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(0L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeAndMultipleBatches() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Create 4 batches of records.
+ ByteBuffer buffer = ByteBuffer.allocate(4096);
+ memoryRecordsBuilder(buffer, 5, 2).close();
+ memoryRecordsBuilder(buffer, 5, 10).close();
+ memoryRecordsBuilder(buffer, 7, 15).close();
+ memoryRecordsBuilder(buffer, 6, 22).close();
+ buffer.flip();
+ MemoryRecords records = MemoryRecords.readableRecords(buffer);
+
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 5 /* Batch size */,
+ 100,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 26 /* Gap of 3 records will also be added to first batch */);
+
+ // Fetch expected records from 4 batches, but change the first expected record to include gap offsets.
+ List<AcquiredRecords> expectedAcquiredRecords = expectedAcquiredRecords(records, 1);
+ expectedAcquiredRecords.remove(0);
+ expectedAcquiredRecords.addAll(0, expectedAcquiredRecord(2, 9, 1));
+
+ assertArrayEquals(expectedAcquiredRecords.toArray(), acquiredRecordsList.toArray());
+ assertEquals(28, sharePartition.nextFetchOffset());
+ assertEquals(4, sharePartition.cachedState().size());
+ assertTrue(sharePartition.cachedState().containsKey(2L));
+ assertTrue(sharePartition.cachedState().containsKey(10L));
+ assertTrue(sharePartition.cachedState().containsKey(15L));
+ assertTrue(sharePartition.cachedState().containsKey(22L));
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(2L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(10L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(15L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(22L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(2L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(10L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(15L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(22L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(2L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(10L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(15L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(22L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(2L).offsetState());
+ assertNull(sharePartition.cachedState().get(10L).offsetState());
+ assertNull(sharePartition.cachedState().get(15L).offsetState());
+ assertNull(sharePartition.cachedState().get(22L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeAndMaxFetchRecords() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Create 3 batches of records.
+ ByteBuffer buffer = ByteBuffer.allocate(4096);
+ memoryRecordsBuilder(buffer, 5, 0).close();
+ memoryRecordsBuilder(buffer, 15, 5).close();
+ memoryRecordsBuilder(buffer, 15, 20).close();
+ buffer.flip();
+ MemoryRecords records = MemoryRecords.readableRecords(buffer);
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 2 /* Batch size */,
+ 10,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 20);
+
+ List<AcquiredRecords> expectedAcquiredRecords = expectedAcquiredRecords(records, 1);
+ // The last batch should be ignored as it exceeds the max fetch records.
+ expectedAcquiredRecords.remove(2);
+
+ assertArrayEquals(expectedAcquiredRecords.toArray(), acquiredRecordsList.toArray());
+ assertEquals(20, sharePartition.nextFetchOffset());
+ assertEquals(2, sharePartition.cachedState().size());
+ assertEquals(0, sharePartition.cachedState().get(0L).firstOffset());
+ assertEquals(4, sharePartition.cachedState().get(0L).lastOffset());
+ assertEquals(5, sharePartition.cachedState().get(5L).firstOffset());
+ assertEquals(19, sharePartition.cachedState().get(5L).lastOffset());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(5L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(5L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(0L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(5L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(0L).offsetState());
+ assertNull(sharePartition.cachedState().get(5L).offsetState());
+ }
+
+ @Test
+ public void testAcquireSingleBatchWithBatchSizeAndEndOffsetAheadBatchFirstOffset() {
Review Comment:
Do I need to have better method name so it's not confusing?
--
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-18452: Implemented batch size in acquired records [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1913288932
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -1104,6 +1116,109 @@ public void testAcquireWithEmptyFetchRecords() {
assertEquals(0, sharePartition.nextFetchOffset());
}
+ @Test
+ public void testAcquireWithBatchSizeAndSingleBatch() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Single batch has more records than batch size. Hence, only a single batch exceeding the batch size
+ // should be acquired.
+ MemoryRecords records = memoryRecords(5);
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 2 /* Batch size */,
+ 10,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
+
+ assertArrayEquals(expectedAcquiredRecord(0, 4, 1).toArray(), acquiredRecordsList.toArray());
+ assertEquals(5, sharePartition.nextFetchOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ assertEquals(0, sharePartition.cachedState().get(0L).firstOffset());
+ assertEquals(4, sharePartition.cachedState().get(0L).lastOffset());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(0L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(0L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeAndMultipleBatches() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Create 3 batches of records.
+ ByteBuffer buffer = ByteBuffer.allocate(4096);
+ memoryRecordsBuilder(buffer, 5, 2).close();
+ memoryRecordsBuilder(buffer, 5, 10).close();
+ memoryRecordsBuilder(buffer, 7, 15).close();
+ memoryRecordsBuilder(buffer, 5, 22).close();
+ buffer.flip();
+ MemoryRecords records = MemoryRecords.readableRecords(buffer);
+
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 5 /* Batch size */,
+ 100,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 25 /* Gap of 3 records will also be added to first batch */);
+
+ // Fetch expected records from 3 batches, but change the first expected record to include gap offsets.
+ List<AcquiredRecords> expectedAcquiredRecords = expectedAcquiredRecords(records, 1);
+ expectedAcquiredRecords.remove(0);
+ expectedAcquiredRecords.addAll(0, expectedAcquiredRecord(2, 9, 1));
+
+ assertArrayEquals(expectedAcquiredRecords.toArray(), acquiredRecordsList.toArray());
+ assertEquals(27, sharePartition.nextFetchOffset());
+ assertEquals(4, sharePartition.cachedState().size());
+ assertTrue(sharePartition.cachedState().containsKey(2L));
+ assertTrue(sharePartition.cachedState().containsKey(10L));
+ assertTrue(sharePartition.cachedState().containsKey(15L));
+ assertTrue(sharePartition.cachedState().containsKey(22L));
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(2L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(10L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(15L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(22L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(2L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(10L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(15L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(22L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(2L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(10L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(15L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(22L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(2L).offsetState());
+ assertNull(sharePartition.cachedState().get(10L).offsetState());
+ assertNull(sharePartition.cachedState().get(15L).offsetState());
+ assertNull(sharePartition.cachedState().get(22L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeWithBatchSize() {
Review Comment:
My bad, corrected it.
--
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-18452: Implemented batch size in acquired records [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1915209071
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -627,9 +628,8 @@ public ShareAcquiredRecords acquire(
if (subMap.isEmpty()) {
log.trace("No cached data exists for the share partition for requested fetch batch: {}-{}",
groupId, topicIdPartition);
- AcquiredRecords acquiredRecords = acquireNewBatchRecords(memberId, fetchPartitionData.records.batches(),
- firstBatch.baseOffset(), lastBatch.lastOffset(), maxFetchRecords);
- return ShareAcquiredRecords.fromAcquiredRecords(acquiredRecords);
+ return acquireNewBatchRecords(memberId, fetchPartitionData.records.batches(),
Review Comment:
Created: https://issues.apache.org/jira/browse/KAFKA-18522. I ll adjust the fetch records batches in response accordingly.
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -244,13 +244,14 @@ public CompletableFuture<Map<TopicIdPartition, PartitionData>> fetchMessages(
String groupId,
String memberId,
FetchParams fetchParams,
+ int batchSize,
Review Comment:
My bad, 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-18452: Implemented batch size in acquired records [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1915460848
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -1104,6 +1116,120 @@ public void testAcquireWithEmptyFetchRecords() {
assertEquals(0, sharePartition.nextFetchOffset());
}
+ @Test
+ public void testAcquireWithBatchSizeAndSingleBatch() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Single batch has more records than batch size. Hence, only a single batch exceeding the batch size
+ // should be acquired.
+ MemoryRecords records = memoryRecords(5);
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 2 /* Batch size */,
+ 10,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
+
+ assertArrayEquals(expectedAcquiredRecord(0, 4, 1).toArray(), acquiredRecordsList.toArray());
+ assertEquals(5, sharePartition.nextFetchOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ assertEquals(0, sharePartition.cachedState().get(0L).firstOffset());
+ assertEquals(4, sharePartition.cachedState().get(0L).lastOffset());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(0L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(0L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeAndMultipleBatches() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Create 3 batches of records.
+ ByteBuffer buffer = ByteBuffer.allocate(4096);
+ memoryRecordsBuilder(buffer, 5, 2).close();
+ memoryRecordsBuilder(buffer, 5, 10).close();
+ memoryRecordsBuilder(buffer, 7, 15).close();
+ memoryRecordsBuilder(buffer, 6, 22).close();
+ buffer.flip();
+ MemoryRecords records = MemoryRecords.readableRecords(buffer);
+
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 5 /* Batch size */,
+ 100,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 26 /* Gap of 3 records will also be added to first batch */);
+
+ // Fetch expected records from 4 batches, but change the first expected record to include gap offsets.
+ List<AcquiredRecords> expectedAcquiredRecords = expectedAcquiredRecords(records, 1);
+ expectedAcquiredRecords.remove(0);
+ expectedAcquiredRecords.addAll(0, expectedAcquiredRecord(2, 9, 1));
+
+ assertArrayEquals(expectedAcquiredRecords.toArray(), acquiredRecordsList.toArray());
+ assertEquals(28, sharePartition.nextFetchOffset());
+ assertEquals(4, sharePartition.cachedState().size());
+ assertTrue(sharePartition.cachedState().containsKey(2L));
+ assertTrue(sharePartition.cachedState().containsKey(10L));
+ assertTrue(sharePartition.cachedState().containsKey(15L));
+ assertTrue(sharePartition.cachedState().containsKey(22L));
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(2L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(10L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(15L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(22L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(2L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(10L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(15L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(22L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(2L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(10L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(15L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(22L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(2L).offsetState());
+ assertNull(sharePartition.cachedState().get(10L).offsetState());
+ assertNull(sharePartition.cachedState().get(15L).offsetState());
+ assertNull(sharePartition.cachedState().get(22L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeAndMaxFetchRecords() {
Review Comment:
I think that's being validated in `testAcquireWithMultipleBatchesAndMaxFetchRecords` where we provide 3 batches and 1 single response by combining first 2 batches arrive in response. Am I missing something here?
--
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-18452: Implemented batch size in acquired records [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1915461874
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -1104,6 +1116,120 @@ public void testAcquireWithEmptyFetchRecords() {
assertEquals(0, sharePartition.nextFetchOffset());
}
+ @Test
+ public void testAcquireWithBatchSizeAndSingleBatch() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Single batch has more records than batch size. Hence, only a single batch exceeding the batch size
+ // should be acquired.
+ MemoryRecords records = memoryRecords(5);
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 2 /* Batch size */,
+ 10,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
+
+ assertArrayEquals(expectedAcquiredRecord(0, 4, 1).toArray(), acquiredRecordsList.toArray());
+ assertEquals(5, sharePartition.nextFetchOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ assertEquals(0, sharePartition.cachedState().get(0L).firstOffset());
+ assertEquals(4, sharePartition.cachedState().get(0L).lastOffset());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(0L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(0L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeAndMultipleBatches() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Create 3 batches of records.
Review Comment:
Sorry, fixed.
--
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-18452: Implemented batch size in acquired records [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #18459:
URL: https://github.com/apache/kafka/pull/18459#issuecomment-2592932011
> @apoorvmittal10 : Thanks for the updated PR. It seems the following tests are quarantined?
>
> ```
> SharePartitionTest > testAcquireSingleBatchWithBatchSizeAndEndOffsetLargerThanBatchFirstOffset()
> SharePartitionTest > testAcquireWithBatchSizeAndMaxFetchRecords()
> SharePartitionTest > testAcquireWithBatchSizeAndMultipleBatches()
> SharePartitionTest > testAcquireBatchSkipWithBatchSizeAndEndOffsetLargerThanFirstBatch()
> SharePartitionTest > testAcquireWithBatchSizeAndEndOffsetLargerThanBatchFirstOffset()
> SharePartitionTest > testAcquireWithBatchSizeAndSingleBatch()
> ```
@junrao I spoke to @mumrah and he mentioned that any new tests will be auto quarantined for 7 days. As these tests are added in this PR hence marked as quarantined.
--
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-18452: Implemented batch size in acquired records [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1915182706
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1204,25 +1205,91 @@ private AcquiredRecords acquireNewBatchRecords(
lastAcquiredOffset = lastOffsetFromBatchWithRequestOffset(batches, firstAcquiredOffset + maxFetchRecords - 1);
}
- // Schedule acquisition lock timeout for the batch.
- AcquisitionLockTimerTask timerTask = scheduleAcquisitionLockTimeout(memberId, firstAcquiredOffset, lastAcquiredOffset);
- // Add the new batch to the in-flight records along with the acquisition lock timeout task for the batch.
- cachedState.put(firstAcquiredOffset, new InFlightBatch(
- memberId,
- firstAcquiredOffset,
- lastAcquiredOffset,
- RecordState.ACQUIRED,
- 1,
- timerTask));
+ // Create batches of acquired records.
+ List<AcquiredRecords> acquiredRecords = createBatches(batches, firstAcquiredOffset, lastAcquiredOffset, memberId, batchSize);
// if the cachedState was empty before acquiring the new batches then startOffset needs to be updated
if (cachedState.firstKey() == firstAcquiredOffset) {
startOffset = firstAcquiredOffset;
}
endOffset = lastAcquiredOffset;
- return new AcquiredRecords()
- .setFirstOffset(firstAcquiredOffset)
- .setLastOffset(lastAcquiredOffset)
- .setDeliveryCount((short) 1);
+ return new ShareAcquiredRecords(acquiredRecords, (int) (lastAcquiredOffset - firstAcquiredOffset + 1));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private List<AcquiredRecords> createBatches(
+ Iterable<? extends RecordBatch> batches,
+ long firstAcquiredOffset,
+ long lastAcquiredOffset,
+ String memberId,
+ int batchSize
+ ) {
+ lock.writeLock().lock();
+ try {
+ List<AcquiredRecords> result = new ArrayList<>();
+ if (lastAcquiredOffset - firstAcquiredOffset + 1 <= batchSize) {
+ // No split of batches is required as the batch size is greater than records which
+ // can be acquired.
+ result.add(new AcquiredRecords()
+ .setFirstOffset(firstAcquiredOffset)
+ .setLastOffset(lastAcquiredOffset)
+ .setDeliveryCount((short) 1));
+ } else {
+ // The batch is split into multiple batches considering batch size.
+ long currentFirstOffset = firstAcquiredOffset;
+ RecordBatch lastBatch = null;
+ // Try reading only the baseOffset of the batch and avoid reading the lastOffset as
+ // lastOffset call of RecordBatch is expensive (loads headers).
+ for (RecordBatch batch : batches) {
+ long batchBaseOffset = batch.baseOffset();
+ if (batchBaseOffset < firstAcquiredOffset) {
+ continue;
+ }
+
+ // Track last batch for filling the final batch, if required.
+ lastBatch = batch;
+ // Check if the batch is already past the last acquired offset then break.
+ if (batchBaseOffset > lastAcquiredOffset) {
+ result.add(new AcquiredRecords()
+ .setFirstOffset(currentFirstOffset)
+ .setLastOffset(lastAcquiredOffset)
+ .setDeliveryCount((short) 1));
Review Comment:
Yeah, agree. I have simplified the `if/else` statement as well to make common insertion in array.
--
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-18452: Implemented batch size in acquired records [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1915210540
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1204,25 +1205,91 @@ private AcquiredRecords acquireNewBatchRecords(
lastAcquiredOffset = lastOffsetFromBatchWithRequestOffset(batches, firstAcquiredOffset + maxFetchRecords - 1);
}
- // Schedule acquisition lock timeout for the batch.
- AcquisitionLockTimerTask timerTask = scheduleAcquisitionLockTimeout(memberId, firstAcquiredOffset, lastAcquiredOffset);
- // Add the new batch to the in-flight records along with the acquisition lock timeout task for the batch.
- cachedState.put(firstAcquiredOffset, new InFlightBatch(
- memberId,
- firstAcquiredOffset,
- lastAcquiredOffset,
- RecordState.ACQUIRED,
- 1,
- timerTask));
+ // Create batches of acquired records.
+ List<AcquiredRecords> acquiredRecords = createBatches(batches, firstAcquiredOffset, lastAcquiredOffset, memberId, batchSize);
// if the cachedState was empty before acquiring the new batches then startOffset needs to be updated
if (cachedState.firstKey() == firstAcquiredOffset) {
startOffset = firstAcquiredOffset;
}
endOffset = lastAcquiredOffset;
- return new AcquiredRecords()
- .setFirstOffset(firstAcquiredOffset)
- .setLastOffset(lastAcquiredOffset)
- .setDeliveryCount((short) 1);
+ return new ShareAcquiredRecords(acquiredRecords, (int) (lastAcquiredOffset - firstAcquiredOffset + 1));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private List<AcquiredRecords> createBatches(
+ Iterable<? extends RecordBatch> batches,
+ long firstAcquiredOffset,
+ long lastAcquiredOffset,
+ String memberId,
Review Comment:
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-18452: Implemented batch size in acquired records [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #18459:
URL: https://github.com/apache/kafka/pull/18459#issuecomment-2592063122
> @apoorvmittal10 : Thanks for the updated PR. It seems the following tests are quarantined?
>
> ```
> SharePartitionTest > testAcquireSingleBatchWithBatchSizeAndEndOffsetLargerThanBatchFirstOffset()
> SharePartitionTest > testAcquireWithBatchSizeAndMaxFetchRecords()
> SharePartitionTest > testAcquireWithBatchSizeAndMultipleBatches()
> SharePartitionTest > testAcquireBatchSkipWithBatchSizeAndEndOffsetLargerThanFirstBatch()
> SharePartitionTest > testAcquireWithBatchSizeAndEndOffsetLargerThanBatchFirstOffset()
> SharePartitionTest > testAcquireWithBatchSizeAndSingleBatch()
> ```
Ohh strange, looking into.
--
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-18452: Implemented batch size in acquired records [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1915179761
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -589,6 +589,7 @@ public long nextFetchOffset() {
@SuppressWarnings("cyclomaticcomplexity") // Consider refactoring to avoid suppression
public ShareAcquiredRecords acquire(
String memberId,
+ int batchSize,
Review Comment:
Missed it, 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-18452: Implemented batch size in acquired records [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1915699553
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -1104,6 +1116,201 @@ public void testAcquireWithEmptyFetchRecords() {
assertEquals(0, sharePartition.nextFetchOffset());
}
+ @Test
+ public void testAcquireWithBatchSizeAndSingleBatch() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Single batch has more records than batch size. Hence, only a single batch exceeding the batch size
+ // should be acquired.
+ MemoryRecords records = memoryRecords(5);
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 2 /* Batch size */,
+ 10,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
+
+ assertArrayEquals(expectedAcquiredRecord(0, 4, 1).toArray(), acquiredRecordsList.toArray());
+ assertEquals(5, sharePartition.nextFetchOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ assertEquals(0, sharePartition.cachedState().get(0L).firstOffset());
+ assertEquals(4, sharePartition.cachedState().get(0L).lastOffset());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(0L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(0L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeAndMultipleBatches() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Create 4 batches of records.
+ ByteBuffer buffer = ByteBuffer.allocate(4096);
+ memoryRecordsBuilder(buffer, 5, 2).close();
+ memoryRecordsBuilder(buffer, 5, 10).close();
+ memoryRecordsBuilder(buffer, 7, 15).close();
+ memoryRecordsBuilder(buffer, 6, 22).close();
+ buffer.flip();
+ MemoryRecords records = MemoryRecords.readableRecords(buffer);
+
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 5 /* Batch size */,
+ 100,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 26 /* Gap of 3 records will also be added to first batch */);
+
+ // Fetch expected records from 4 batches, but change the first expected record to include gap offsets.
+ List<AcquiredRecords> expectedAcquiredRecords = expectedAcquiredRecords(records, 1);
+ expectedAcquiredRecords.remove(0);
+ expectedAcquiredRecords.addAll(0, expectedAcquiredRecord(2, 9, 1));
+
+ assertArrayEquals(expectedAcquiredRecords.toArray(), acquiredRecordsList.toArray());
+ assertEquals(28, sharePartition.nextFetchOffset());
+ assertEquals(4, sharePartition.cachedState().size());
+ assertTrue(sharePartition.cachedState().containsKey(2L));
+ assertTrue(sharePartition.cachedState().containsKey(10L));
+ assertTrue(sharePartition.cachedState().containsKey(15L));
+ assertTrue(sharePartition.cachedState().containsKey(22L));
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(2L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(10L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(15L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(22L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(2L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(10L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(15L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(22L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(2L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(10L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(15L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(22L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(2L).offsetState());
+ assertNull(sharePartition.cachedState().get(10L).offsetState());
+ assertNull(sharePartition.cachedState().get(15L).offsetState());
+ assertNull(sharePartition.cachedState().get(22L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeAndMaxFetchRecords() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Create 3 batches of records.
+ ByteBuffer buffer = ByteBuffer.allocate(4096);
+ memoryRecordsBuilder(buffer, 5, 0).close();
+ memoryRecordsBuilder(buffer, 15, 5).close();
+ memoryRecordsBuilder(buffer, 15, 20).close();
+ buffer.flip();
+ MemoryRecords records = MemoryRecords.readableRecords(buffer);
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 2 /* Batch size */,
+ 10,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 20);
+
+ List<AcquiredRecords> expectedAcquiredRecords = expectedAcquiredRecords(records, 1);
+ // The last batch should be ignored as it exceeds the max fetch records.
+ expectedAcquiredRecords.remove(2);
+
+ assertArrayEquals(expectedAcquiredRecords.toArray(), acquiredRecordsList.toArray());
+ assertEquals(20, sharePartition.nextFetchOffset());
+ assertEquals(2, sharePartition.cachedState().size());
+ assertEquals(0, sharePartition.cachedState().get(0L).firstOffset());
+ assertEquals(4, sharePartition.cachedState().get(0L).lastOffset());
+ assertEquals(5, sharePartition.cachedState().get(5L).firstOffset());
+ assertEquals(19, sharePartition.cachedState().get(5L).lastOffset());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(5L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(5L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(0L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(5L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(0L).offsetState());
+ assertNull(sharePartition.cachedState().get(5L).offsetState());
+ }
+
+ @Test
+ public void testAcquireSingleBatchWithBatchSizeAndEndOffsetAheadBatchFirstOffset() {
Review Comment:
OK. Perhaps use larger since ahead can be ambiguous?
--
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-18452: Implemented batch size in acquired records [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1915702844
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -1104,6 +1116,201 @@ public void testAcquireWithEmptyFetchRecords() {
assertEquals(0, sharePartition.nextFetchOffset());
}
+ @Test
+ public void testAcquireWithBatchSizeAndSingleBatch() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Single batch has more records than batch size. Hence, only a single batch exceeding the batch size
+ // should be acquired.
+ MemoryRecords records = memoryRecords(5);
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 2 /* Batch size */,
+ 10,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
+
+ assertArrayEquals(expectedAcquiredRecord(0, 4, 1).toArray(), acquiredRecordsList.toArray());
+ assertEquals(5, sharePartition.nextFetchOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ assertEquals(0, sharePartition.cachedState().get(0L).firstOffset());
+ assertEquals(4, sharePartition.cachedState().get(0L).lastOffset());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(0L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(0L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeAndMultipleBatches() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Create 4 batches of records.
+ ByteBuffer buffer = ByteBuffer.allocate(4096);
+ memoryRecordsBuilder(buffer, 5, 2).close();
+ memoryRecordsBuilder(buffer, 5, 10).close();
+ memoryRecordsBuilder(buffer, 7, 15).close();
+ memoryRecordsBuilder(buffer, 6, 22).close();
+ buffer.flip();
+ MemoryRecords records = MemoryRecords.readableRecords(buffer);
+
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 5 /* Batch size */,
+ 100,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 26 /* Gap of 3 records will also be added to first batch */);
+
+ // Fetch expected records from 4 batches, but change the first expected record to include gap offsets.
+ List<AcquiredRecords> expectedAcquiredRecords = expectedAcquiredRecords(records, 1);
+ expectedAcquiredRecords.remove(0);
+ expectedAcquiredRecords.addAll(0, expectedAcquiredRecord(2, 9, 1));
+
+ assertArrayEquals(expectedAcquiredRecords.toArray(), acquiredRecordsList.toArray());
+ assertEquals(28, sharePartition.nextFetchOffset());
+ assertEquals(4, sharePartition.cachedState().size());
+ assertTrue(sharePartition.cachedState().containsKey(2L));
+ assertTrue(sharePartition.cachedState().containsKey(10L));
+ assertTrue(sharePartition.cachedState().containsKey(15L));
+ assertTrue(sharePartition.cachedState().containsKey(22L));
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(2L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(10L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(15L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(22L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(2L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(10L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(15L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(22L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(2L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(10L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(15L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(22L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(2L).offsetState());
+ assertNull(sharePartition.cachedState().get(10L).offsetState());
+ assertNull(sharePartition.cachedState().get(15L).offsetState());
+ assertNull(sharePartition.cachedState().get(22L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithBatchSizeAndMaxFetchRecords() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ // Create 3 batches of records.
+ ByteBuffer buffer = ByteBuffer.allocate(4096);
+ memoryRecordsBuilder(buffer, 5, 0).close();
+ memoryRecordsBuilder(buffer, 15, 5).close();
+ memoryRecordsBuilder(buffer, 15, 20).close();
+ buffer.flip();
+ MemoryRecords records = MemoryRecords.readableRecords(buffer);
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 2 /* Batch size */,
+ 10,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 20);
+
+ List<AcquiredRecords> expectedAcquiredRecords = expectedAcquiredRecords(records, 1);
+ // The last batch should be ignored as it exceeds the max fetch records.
+ expectedAcquiredRecords.remove(2);
+
+ assertArrayEquals(expectedAcquiredRecords.toArray(), acquiredRecordsList.toArray());
+ assertEquals(20, sharePartition.nextFetchOffset());
+ assertEquals(2, sharePartition.cachedState().size());
+ assertEquals(0, sharePartition.cachedState().get(0L).firstOffset());
+ assertEquals(4, sharePartition.cachedState().get(0L).lastOffset());
+ assertEquals(5, sharePartition.cachedState().get(5L).firstOffset());
+ assertEquals(19, sharePartition.cachedState().get(5L).lastOffset());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(0L).batchState());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(5L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(0L).batchMemberId());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(5L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(0L).batchDeliveryCount());
+ assertEquals(1, sharePartition.cachedState().get(5L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(0L).offsetState());
+ assertNull(sharePartition.cachedState().get(5L).offsetState());
+ }
+
+ @Test
+ public void testAcquireSingleBatchWithBatchSizeAndEndOffsetAheadBatchFirstOffset() {
Review Comment:
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-18452: Implemented batch size in acquired records [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18459:
URL: https://github.com/apache/kafka/pull/18459#discussion_r1915462709
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -580,6 +580,7 @@ public long nextFetchOffset() {
* fetched from the leader.
*
* @param memberId The member id of the client that is fetching the record.
+ * @param batchSize The batch size of the individual acquired records batch.
Review Comment:
Done, also in SharePartitionManager.
--
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