You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ju...@apache.org on 2024/10/23 20:21:39 UTC
(kafka) branch trunk updated: KAFKA-17774: Adding capability to handle max fetch records in Share Fetch (KIP-932) (#17322)
This is an automated email from the ASF dual-hosted git repository.
junrao pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 0d44415bac1 KAFKA-17774: Adding capability to handle max fetch records in Share Fetch (KIP-932) (#17322)
0d44415bac1 is described below
commit 0d44415bac1c33aea637ee98d220626e05e63350
Author: Apoorv Mittal <ap...@gmail.com>
AuthorDate: Wed Oct 23 21:21:32 2024 +0100
KAFKA-17774: Adding capability to handle max fetch records in Share Fetch (KIP-932) (#17322)
The PR adds capability to restrict the messages in Share Fetch. The max fetch records will be an additional way to limit the number of records sent from broker to client.
In Share Fetch, with min and mx bytes, there exists 3 problems:
1. The max.poll.records client config sends the max number of records defined to application but might have fetched extra becuase of higher max bytes. But the timeout for the sent records has started on the broker.
2. As the application processes records as per max.poll.records, hence those number of records are sent in every acknowledgement. This causes the cache data to be tracked per offset as the batch is broken.
3. The client has to sent the partial acknoledgment batch and cannot piggyback on fetch requests.
To handle the above scenario max fetch records has been added. Once this PR is merged and we define the right methodolgy then KIP will be updated to have max fetch records in share fetch RPC rather as broker config.
Reviewers: Abhinav Dixit <ad...@confluent.io>, Andrew Schofield <as...@confluent.io>, Chia-Ping Tsai <ch...@gmail.com>, Jun Rao <ju...@gmail.com>
---
.../java/kafka/server/share/ShareFetchUtils.java | 29 +-
.../java/kafka/server/share/SharePartition.java | 117 +++-
.../kafka/server/share/SharePartitionManager.java | 13 +-
.../src/main/scala/kafka/server/BrokerServer.scala | 1 +
.../kafka/server/share/DelayedShareFetchTest.java | 32 +-
.../kafka/server/share/ShareFetchUtilsTest.java | 142 ++++-
.../server/share/SharePartitionManagerTest.java | 14 +-
.../kafka/server/share/SharePartitionTest.java | 693 +++++++++++++--------
.../scala/unit/kafka/server/KafkaConfigTest.scala | 1 +
.../group/modern/share/ShareGroupConfig.java | 15 +-
.../server/share/fetch/ShareAcquiredRecords.java | 72 +++
.../kafka/server/share/fetch/ShareFetchData.java | 9 +-
12 files changed, 798 insertions(+), 340 deletions(-)
diff --git a/core/src/main/java/kafka/server/share/ShareFetchUtils.java b/core/src/main/java/kafka/server/share/ShareFetchUtils.java
index 10fea91c00a..b517f2f9e06 100644
--- a/core/src/main/java/kafka/server/share/ShareFetchUtils.java
+++ b/core/src/main/java/kafka/server/share/ShareFetchUtils.java
@@ -20,10 +20,10 @@ import kafka.server.ReplicaManager;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.message.ShareFetchResponseData;
-import org.apache.kafka.common.message.ShareFetchResponseData.AcquiredRecords;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.record.FileRecords;
import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.server.share.fetch.ShareAcquiredRecords;
import org.apache.kafka.server.share.fetch.ShareFetchData;
import org.apache.kafka.server.storage.log.FetchPartitionData;
@@ -32,7 +32,6 @@ import org.slf4j.LoggerFactory;
import java.util.Collections;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -55,12 +54,17 @@ public class ShareFetchUtils {
ReplicaManager replicaManager
) {
Map<TopicIdPartition, ShareFetchResponseData.PartitionData> response = new HashMap<>();
- responseData.forEach((topicIdPartition, fetchPartitionData) -> {
+
+ // Acquired records count for the share fetch request.
+ int acquiredRecordsCount = 0;
+ for (Map.Entry<TopicIdPartition, FetchPartitionData> entry : responseData.entrySet()) {
+ TopicIdPartition topicIdPartition = entry.getKey();
+ FetchPartitionData fetchPartitionData = entry.getValue();
SharePartition sharePartition = sharePartitionManager.sharePartition(shareFetchData.groupId(), topicIdPartition);
if (sharePartition == null) {
log.error("Encountered null share partition for groupId={}, topicIdPartition={}. Skipping it.", shareFetchData.groupId(), topicIdPartition);
- return;
+ continue;
}
ShareFetchResponseData.PartitionData partitionData = new ShareFetchResponseData.PartitionData()
.setPartitionIndex(topicIdPartition.partition());
@@ -84,24 +88,31 @@ public class ShareFetchUtils {
partitionData.setErrorMessage(Errors.NONE.message());
}
} else {
- List<AcquiredRecords> acquiredRecords = sharePartition.acquire(shareFetchData.memberId(), fetchPartitionData);
+ ShareAcquiredRecords shareAcquiredRecords = sharePartition.acquire(shareFetchData.memberId(), shareFetchData.maxFetchRecords() - acquiredRecordsCount, fetchPartitionData);
log.trace("Acquired records for topicIdPartition: {} with share fetch data: {}, records: {}",
- topicIdPartition, shareFetchData, acquiredRecords);
+ topicIdPartition, shareFetchData, shareAcquiredRecords);
// Maybe, in the future, check if no records are acquired, and we want to retry
// replica manager fetch. Depends on the share partition manager implementation,
// if we want parallel requests for the same share partition or not.
- if (acquiredRecords.isEmpty()) {
+ if (shareAcquiredRecords.acquiredRecords().isEmpty()) {
partitionData
.setRecords(null)
.setAcquiredRecords(Collections.emptyList());
} else {
partitionData
+ // We set the records to the fetchPartitionData records. We do not alter the records
+ // fetched from the replica manager as they follow zero copy buffer. The acquired records
+ // might be a subset of the records fetched from the replica manager, depending
+ // on the max fetch records or available records in the share partition. The client
+ // sends the max bytes in request which should limit the bytes sent to the client
+ // in the response.
.setRecords(fetchPartitionData.records)
- .setAcquiredRecords(acquiredRecords);
+ .setAcquiredRecords(shareAcquiredRecords.acquiredRecords());
+ acquiredRecordsCount += shareAcquiredRecords.count();
}
}
response.put(topicIdPartition, partitionData);
- });
+ }
return response;
}
diff --git a/core/src/main/java/kafka/server/share/SharePartition.java b/core/src/main/java/kafka/server/share/SharePartition.java
index f9aa2fb528f..67a87ba1438 100644
--- a/core/src/main/java/kafka/server/share/SharePartition.java
+++ b/core/src/main/java/kafka/server/share/SharePartition.java
@@ -34,6 +34,7 @@ import org.apache.kafka.common.record.RecordBatch;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.coordinator.group.GroupConfigManager;
import org.apache.kafka.server.share.acknowledge.ShareAcknowledgementBatch;
+import org.apache.kafka.server.share.fetch.ShareAcquiredRecords;
import org.apache.kafka.server.share.persister.GroupTopicPartitionData;
import org.apache.kafka.server.share.persister.PartitionAllData;
import org.apache.kafka.server.share.persister.PartitionErrorData;
@@ -513,18 +514,27 @@ public class SharePartition {
* fetched from the leader.
*
* @param memberId The member id of the client that is fetching the record.
+ * @param maxFetchRecords The maximum number of records that should be acquired, this is a soft
+ * limit and the method might acquire more records than the maxFetchRecords,
+ * if the records are already part of the same fetch batch.
* @param fetchPartitionData The fetched records for the share partition.
* @return The acquired records for the share partition.
*/
- public List<AcquiredRecords> acquire(
+ public ShareAcquiredRecords acquire(
String memberId,
+ int maxFetchRecords,
FetchPartitionData fetchPartitionData
) {
log.trace("Received acquire request for share partition: {}-{} memberId: {}", groupId, topicIdPartition, memberId);
+ if (maxFetchRecords <= 0) {
+ // Nothing to acquire.
+ return ShareAcquiredRecords.empty();
+ }
+
RecordBatch lastBatch = fetchPartitionData.records.lastBatch().orElse(null);
if (lastBatch == null) {
// Nothing to acquire.
- return Collections.emptyList();
+ return ShareAcquiredRecords.empty();
}
// We require the first batch of records to get the base offset. Stop parsing further
@@ -550,17 +560,25 @@ public class SharePartition {
if (subMap.isEmpty()) {
log.trace("No cached data exists for the share partition for requested fetch batch: {}-{}",
groupId, topicIdPartition);
- return Collections.singletonList(
- acquireNewBatchRecords(memberId, firstBatch.baseOffset(), lastBatch.lastOffset()));
+ AcquiredRecords acquiredRecords = acquireNewBatchRecords(memberId, fetchPartitionData.records.batches(),
+ firstBatch.baseOffset(), lastBatch.lastOffset(), maxFetchRecords);
+ return ShareAcquiredRecords.fromAcquiredRecords(acquiredRecords);
}
log.trace("Overlap exists with in-flight records. Acquire the records if available for"
+ " the share partition: {}-{}", groupId, topicIdPartition);
List<AcquiredRecords> result = new ArrayList<>();
+ // The acquired count is used to track the number of records acquired for the request.
+ int acquiredCount = 0;
// The fetched records are already part of the in-flight records. The records might
// be available for re-delivery hence try acquiring same. The request batches could
// be an exact match, subset or span over multiple already fetched batches.
for (Map.Entry<Long, InFlightBatch> entry : subMap.entrySet()) {
+ // If the acquired count is equal to the max fetch records then break the loop.
+ if (acquiredCount >= maxFetchRecords) {
+ break;
+ }
+
InFlightBatch inFlightBatch = entry.getValue();
// Compute if the batch is a full match.
boolean fullMatch = checkForFullMatch(inFlightBatch, firstBatch.baseOffset(), lastBatch.lastOffset());
@@ -586,7 +604,11 @@ public class SharePartition {
// the offsets state in the in-flight batch.
inFlightBatch.maybeInitializeOffsetStateUpdate();
}
- acquireSubsetBatchRecords(memberId, firstBatch.baseOffset(), lastBatch.lastOffset(), inFlightBatch, result);
+ // Do not send max fetch records to acquireSubsetBatchRecords as we want to acquire
+ // all the records from the batch as the batch will anyway be part of the file-records
+ // response batch.
+ int acquiredSubsetCount = acquireSubsetBatchRecords(memberId, firstBatch.baseOffset(), lastBatch.lastOffset(), inFlightBatch, result);
+ acquiredCount += acquiredSubsetCount;
continue;
}
@@ -612,16 +634,20 @@ public class SharePartition {
.setFirstOffset(inFlightBatch.firstOffset())
.setLastOffset(inFlightBatch.lastOffset())
.setDeliveryCount((short) inFlightBatch.batchDeliveryCount()));
+ acquiredCount += (int) (inFlightBatch.lastOffset() - inFlightBatch.firstOffset() + 1);
}
// Some of the request offsets are not found in the fetched batches. Acquire the
// missing records as well.
- if (subMap.lastEntry().getValue().lastOffset() < lastBatch.lastOffset()) {
+ if (acquiredCount < maxFetchRecords && subMap.lastEntry().getValue().lastOffset() < lastBatch.lastOffset()) {
log.trace("There exists another batch which needs to be acquired as well");
- result.add(acquireNewBatchRecords(memberId, subMap.lastEntry().getValue().lastOffset() + 1,
- lastBatch.lastOffset()));
+ AcquiredRecords acquiredRecords = acquireNewBatchRecords(memberId, fetchPartitionData.records.batches(),
+ subMap.lastEntry().getValue().lastOffset() + 1,
+ lastBatch.lastOffset(), maxFetchRecords - acquiredCount);
+ result.add(acquiredRecords);
+ acquiredCount += (int) (acquiredRecords.lastOffset() - acquiredRecords.firstOffset() + 1);
}
- return result;
+ return new ShareAcquiredRecords(result, acquiredCount);
} finally {
lock.writeLock().unlock();
}
@@ -1059,36 +1085,57 @@ public class SharePartition {
private AcquiredRecords acquireNewBatchRecords(
String memberId,
+ Iterable<? extends RecordBatch> batches,
long firstOffset,
- long lastOffset
+ long lastOffset,
+ int maxFetchRecords
) {
lock.writeLock().lock();
try {
+ // If same batch is fetched and previous batch is removed from the cache then we need to
+ // update the batch first offset to endOffset, only if endOffset passed the firstOffset.
+ // For an initial start of the share fetch from a topic partition the endOffset will be initialized
+ // to 0 but firstOffset can be higher than 0.
+ long firstAcquiredOffset = firstOffset;
+ if (cachedState.isEmpty() && endOffset > firstAcquiredOffset) {
+ firstAcquiredOffset = endOffset;
+ }
+
+ // Check how many messages can be acquired from the batch.
+ long lastAcquiredOffset = lastOffset;
+ if (maxFetchRecords < lastAcquiredOffset - firstAcquiredOffset + 1) {
+ // The max messages to acquire is less than the complete available batches hence
+ // limit the acquired records. The last offset shall be the batches last offset
+ // which falls under the max messages limit. As the max fetch records is the soft
+ // limit, the last offset can be higher than the max messages.
+ lastAcquiredOffset = lastOffsetFromBatchWithRequestOffset(batches, firstAcquiredOffset + maxFetchRecords - 1);
+ }
+
// Schedule acquisition lock timeout for the batch.
- AcquisitionLockTimerTask timerTask = scheduleAcquisitionLockTimeout(memberId, firstOffset, lastOffset);
+ 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(firstOffset, new InFlightBatch(
+ cachedState.put(firstAcquiredOffset, new InFlightBatch(
memberId,
- firstOffset,
- lastOffset,
+ firstAcquiredOffset,
+ lastAcquiredOffset,
RecordState.ACQUIRED,
1,
timerTask));
// if the cachedState was empty before acquiring the new batches then startOffset needs to be updated
- if (cachedState.firstKey() == firstOffset) {
- startOffset = firstOffset;
+ if (cachedState.firstKey() == firstAcquiredOffset) {
+ startOffset = firstAcquiredOffset;
}
- endOffset = lastOffset;
+ endOffset = lastAcquiredOffset;
return new AcquiredRecords()
- .setFirstOffset(firstOffset)
- .setLastOffset(lastOffset)
+ .setFirstOffset(firstAcquiredOffset)
+ .setLastOffset(lastAcquiredOffset)
.setDeliveryCount((short) 1);
} finally {
lock.writeLock().unlock();
}
}
- private void acquireSubsetBatchRecords(
+ private int acquireSubsetBatchRecords(
String memberId,
long requestFirstOffset,
long requestLastOffset,
@@ -1096,6 +1143,7 @@ public class SharePartition {
List<AcquiredRecords> result
) {
lock.writeLock().lock();
+ int acquiredCount = 0;
try {
for (Map.Entry<Long, InFlightState> offsetState : inFlightBatch.offsetState.entrySet()) {
// For the first batch which might have offsets prior to the request base
@@ -1134,10 +1182,12 @@ public class SharePartition {
.setFirstOffset(offsetState.getKey())
.setLastOffset(offsetState.getKey())
.setDeliveryCount((short) offsetState.getValue().deliveryCount));
+ acquiredCount++;
}
} finally {
lock.writeLock().unlock();
}
+ return acquiredCount;
}
/**
@@ -1658,6 +1708,33 @@ public class SharePartition {
return lastOffsetAcknowledged;
}
+ /**
+ * Find the last offset from the batch which contains the request offset. If found, return the last offset
+ * of the batch, otherwise return the request offset.
+ *
+ * @param batches The batches to search for the request offset.
+ * @param offset The request offset to find.
+ * @return The last offset of the batch which contains the request offset, otherwise the request offset.
+ */
+ private long lastOffsetFromBatchWithRequestOffset(
+ Iterable<? extends RecordBatch> batches,
+ long offset
+ ) {
+ // Fetch the last batch which might contains the request offset. Avoid calling lastOffset()
+ // on the batches as it loads the header in memory.
+ RecordBatch previousBatch = null;
+ for (RecordBatch batch : batches) {
+ if (offset >= batch.baseOffset()) {
+ previousBatch = batch;
+ continue;
+ }
+ break;
+ }
+ if (previousBatch != null && offset <= previousBatch.lastOffset())
+ return previousBatch.lastOffset();
+ return offset;
+ }
+
// Visible for testing
CompletableFuture<Void> writeShareGroupState(List<PersisterStateBatch> stateBatches) {
CompletableFuture<Void> future = new CompletableFuture<>();
diff --git a/core/src/main/java/kafka/server/share/SharePartitionManager.java b/core/src/main/java/kafka/server/share/SharePartitionManager.java
index 2387a364820..19b4143c8d4 100644
--- a/core/src/main/java/kafka/server/share/SharePartitionManager.java
+++ b/core/src/main/java/kafka/server/share/SharePartitionManager.java
@@ -136,6 +136,11 @@ public class SharePartitionManager implements AutoCloseable {
*/
private final ShareGroupMetrics shareGroupMetrics;
+ /**
+ * The max fetch records is the maximum number of records that can be fetched by a share fetch request.
+ */
+ private final int maxFetchRecords;
+
public SharePartitionManager(
ReplicaManager replicaManager,
Time time,
@@ -143,6 +148,7 @@ public class SharePartitionManager implements AutoCloseable {
int defaultRecordLockDurationMs,
int maxDeliveryCount,
int maxInFlightMessages,
+ int maxFetchRecords,
Persister persister,
GroupConfigManager groupConfigManager,
Metrics metrics
@@ -154,6 +160,7 @@ public class SharePartitionManager implements AutoCloseable {
defaultRecordLockDurationMs,
maxDeliveryCount,
maxInFlightMessages,
+ maxFetchRecords,
persister,
groupConfigManager,
metrics
@@ -168,6 +175,7 @@ public class SharePartitionManager implements AutoCloseable {
int defaultRecordLockDurationMs,
int maxDeliveryCount,
int maxInFlightMessages,
+ int maxFetchRecords,
Persister persister,
GroupConfigManager groupConfigManager,
Metrics metrics
@@ -181,6 +189,7 @@ public class SharePartitionManager implements AutoCloseable {
new SystemTimer("share-group-lock-timeout")),
maxDeliveryCount,
maxInFlightMessages,
+ maxFetchRecords,
persister,
groupConfigManager,
metrics
@@ -197,6 +206,7 @@ public class SharePartitionManager implements AutoCloseable {
Timer timer,
int maxDeliveryCount,
int maxInFlightMessages,
+ int maxFetchRecords,
Persister persister,
GroupConfigManager groupConfigManager,
Metrics metrics
@@ -212,6 +222,7 @@ public class SharePartitionManager implements AutoCloseable {
this.persister = persister;
this.groupConfigManager = groupConfigManager;
this.shareGroupMetrics = new ShareGroupMetrics(Objects.requireNonNull(metrics), time);
+ this.maxFetchRecords = maxFetchRecords;
}
/**
@@ -235,7 +246,7 @@ public class SharePartitionManager implements AutoCloseable {
partitionMaxBytes.keySet(), groupId, fetchParams);
CompletableFuture<Map<TopicIdPartition, PartitionData>> future = new CompletableFuture<>();
- processShareFetch(new ShareFetchData(fetchParams, groupId, memberId, future, partitionMaxBytes));
+ processShareFetch(new ShareFetchData(fetchParams, groupId, memberId, future, partitionMaxBytes, maxFetchRecords));
return future;
}
diff --git a/core/src/main/scala/kafka/server/BrokerServer.scala b/core/src/main/scala/kafka/server/BrokerServer.scala
index f11b90a6974..bdd6cc62b69 100644
--- a/core/src/main/scala/kafka/server/BrokerServer.scala
+++ b/core/src/main/scala/kafka/server/BrokerServer.scala
@@ -432,6 +432,7 @@ class BrokerServer(
config.shareGroupConfig.shareGroupRecordLockDurationMs,
config.shareGroupConfig.shareGroupDeliveryCountLimit,
config.shareGroupConfig.shareGroupPartitionMaxRecordLocks,
+ config.shareGroupConfig.shareFetchMaxFetchRecords,
persister,
groupConfigManager,
new Metrics()
diff --git a/core/src/test/java/kafka/server/share/DelayedShareFetchTest.java b/core/src/test/java/kafka/server/share/DelayedShareFetchTest.java
index 31f2975d349..655cd50eef5 100644
--- a/core/src/test/java/kafka/server/share/DelayedShareFetchTest.java
+++ b/core/src/test/java/kafka/server/share/DelayedShareFetchTest.java
@@ -27,9 +27,11 @@ import org.apache.kafka.common.message.ShareFetchResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.requests.FetchRequest;
import org.apache.kafka.server.share.SharePartitionKey;
+import org.apache.kafka.server.share.fetch.ShareAcquiredRecords;
import org.apache.kafka.server.share.fetch.ShareFetchData;
import org.apache.kafka.server.storage.log.FetchIsolation;
import org.apache.kafka.server.storage.log.FetchParams;
+import org.apache.kafka.server.storage.log.FetchPartitionData;
import org.apache.kafka.server.util.timer.SystemTimer;
import org.apache.kafka.server.util.timer.SystemTimerReaper;
import org.apache.kafka.server.util.timer.Timer;
@@ -59,6 +61,8 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
@@ -68,6 +72,8 @@ import static org.mockito.Mockito.when;
public class DelayedShareFetchTest {
private static final int MAX_WAIT_MS = 5000;
+ private static final int MAX_FETCH_RECORDS = 100;
+
private static Timer mockTimer;
@BeforeEach
@@ -104,7 +110,7 @@ public class DelayedShareFetchTest {
ShareFetchData shareFetchData = new ShareFetchData(
new FetchParams(ApiKeys.SHARE_FETCH.latestVersion(), FetchRequest.ORDINARY_CONSUMER_ID, -1, MAX_WAIT_MS,
1, 1024 * 1024, FetchIsolation.HIGH_WATERMARK, Optional.empty()), groupId, Uuid.randomUuid().toString(),
- new CompletableFuture<>(), partitionMaxBytes);
+ new CompletableFuture<>(), partitionMaxBytes, MAX_FETCH_RECORDS);
when(sp0.canAcquireRecords()).thenReturn(false);
when(sp1.canAcquireRecords()).thenReturn(false);
@@ -142,12 +148,12 @@ public class DelayedShareFetchTest {
ShareFetchData shareFetchData = new ShareFetchData(
new FetchParams(ApiKeys.SHARE_FETCH.latestVersion(), FetchRequest.ORDINARY_CONSUMER_ID, -1, MAX_WAIT_MS,
1, 1024 * 1024, FetchIsolation.HIGH_WATERMARK, Optional.empty()), groupId, Uuid.randomUuid().toString(),
- new CompletableFuture<>(), partitionMaxBytes);
+ new CompletableFuture<>(), partitionMaxBytes, MAX_FETCH_RECORDS);
when(sp0.canAcquireRecords()).thenReturn(true);
when(sp1.canAcquireRecords()).thenReturn(false);
- when(sp0.acquire(any(), any())).thenReturn(
- Collections.singletonList(new ShareFetchResponseData.AcquiredRecords().setFirstOffset(0).setLastOffset(3).setDeliveryCount((short) 1)));
+ when(sp0.acquire(anyString(), anyInt(), any(FetchPartitionData.class))).thenReturn(
+ ShareAcquiredRecords.fromAcquiredRecords(new ShareFetchResponseData.AcquiredRecords().setFirstOffset(0).setLastOffset(3).setDeliveryCount((short) 1)));
doAnswer(invocation -> buildLogReadResult(Collections.singleton(tp0))).when(replicaManager).readFromLog(any(), any(), any(ReplicaQuota.class), anyBoolean());
DelayedShareFetch delayedShareFetch = DelayedShareFetchBuilder.builder()
@@ -186,7 +192,7 @@ public class DelayedShareFetchTest {
ShareFetchData shareFetchData = new ShareFetchData(
new FetchParams(ApiKeys.SHARE_FETCH.latestVersion(), FetchRequest.ORDINARY_CONSUMER_ID, -1, MAX_WAIT_MS,
1, 1024 * 1024, FetchIsolation.HIGH_WATERMARK, Optional.empty()), groupId, Uuid.randomUuid().toString(),
- new CompletableFuture<>(), partitionMaxBytes);
+ new CompletableFuture<>(), partitionMaxBytes, MAX_FETCH_RECORDS);
when(sp0.canAcquireRecords()).thenReturn(false);
when(sp1.canAcquireRecords()).thenReturn(false);
@@ -229,12 +235,12 @@ public class DelayedShareFetchTest {
ShareFetchData shareFetchData = new ShareFetchData(
new FetchParams(ApiKeys.SHARE_FETCH.latestVersion(), FetchRequest.ORDINARY_CONSUMER_ID, -1, MAX_WAIT_MS,
1, 1024 * 1024, FetchIsolation.HIGH_WATERMARK, Optional.empty()), groupId, Uuid.randomUuid().toString(),
- new CompletableFuture<>(), partitionMaxBytes);
+ new CompletableFuture<>(), partitionMaxBytes, MAX_FETCH_RECORDS);
when(sp0.canAcquireRecords()).thenReturn(true);
when(sp1.canAcquireRecords()).thenReturn(false);
- when(sp0.acquire(any(), any())).thenReturn(
- Collections.singletonList(new ShareFetchResponseData.AcquiredRecords().setFirstOffset(0).setLastOffset(3).setDeliveryCount((short) 1)));
+ when(sp0.acquire(anyString(), anyInt(), any(FetchPartitionData.class))).thenReturn(
+ ShareAcquiredRecords.fromAcquiredRecords(new ShareFetchResponseData.AcquiredRecords().setFirstOffset(0).setLastOffset(3).setDeliveryCount((short) 1)));
doAnswer(invocation -> buildLogReadResult(Collections.singleton(tp0))).when(replicaManager).readFromLog(any(), any(), any(ReplicaQuota.class), anyBoolean());
DelayedShareFetch delayedShareFetch = DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetchData)
@@ -270,7 +276,7 @@ public class DelayedShareFetchTest {
ShareFetchData shareFetchData = new ShareFetchData(
new FetchParams(ApiKeys.SHARE_FETCH.latestVersion(), FetchRequest.ORDINARY_CONSUMER_ID, -1, MAX_WAIT_MS,
1, 1024 * 1024, FetchIsolation.HIGH_WATERMARK, Optional.empty()), groupId, Uuid.randomUuid().toString(),
- future, partitionMaxBytes);
+ future, partitionMaxBytes, MAX_FETCH_RECORDS);
when(sp0.maybeAcquireFetchLock()).thenReturn(true);
when(sp0.canAcquireRecords()).thenReturn(false);
@@ -325,7 +331,7 @@ public class DelayedShareFetchTest {
ShareFetchData shareFetchData1 = new ShareFetchData(
new FetchParams(ApiKeys.SHARE_FETCH.latestVersion(), FetchRequest.ORDINARY_CONSUMER_ID, -1, MAX_WAIT_MS,
1, 1024 * 1024, FetchIsolation.HIGH_WATERMARK, Optional.empty()), groupId, Uuid.randomUuid().toString(),
- new CompletableFuture<>(), partitionMaxBytes1);
+ new CompletableFuture<>(), partitionMaxBytes1, MAX_FETCH_RECORDS);
DelayedOperationPurgatory<DelayedShareFetch> delayedShareFetchPurgatory = new DelayedOperationPurgatory<>(
"TestShareFetch", mockTimer, replicaManager.localBrokerId(),
@@ -354,7 +360,7 @@ public class DelayedShareFetchTest {
ShareFetchData shareFetchData2 = new ShareFetchData(
new FetchParams(ApiKeys.SHARE_FETCH.latestVersion(), FetchRequest.ORDINARY_CONSUMER_ID, -1, MAX_WAIT_MS,
1, 1024 * 1024, FetchIsolation.HIGH_WATERMARK, Optional.empty()), groupId, Uuid.randomUuid().toString(),
- new CompletableFuture<>(), partitionMaxBytes2);
+ new CompletableFuture<>(), partitionMaxBytes2, MAX_FETCH_RECORDS);
doAnswer(invocation -> buildLogReadResult(Collections.singleton(tp1))).when(replicaManager).readFromLog(any(), any(), any(ReplicaQuota.class), anyBoolean());
@@ -376,8 +382,8 @@ public class DelayedShareFetchTest {
// sp1 can be acquired now
when(sp1.maybeAcquireFetchLock()).thenReturn(true);
when(sp1.canAcquireRecords()).thenReturn(true);
- when(sp1.acquire(any(), any())).thenReturn(
- Collections.singletonList(new ShareFetchResponseData.AcquiredRecords().setFirstOffset(0).setLastOffset(3).setDeliveryCount((short) 1)));
+ when(sp1.acquire(anyString(), anyInt(), any(FetchPartitionData.class))).thenReturn(
+ ShareAcquiredRecords.fromAcquiredRecords(new ShareFetchResponseData.AcquiredRecords().setFirstOffset(0).setLastOffset(3).setDeliveryCount((short) 1)));
// when forceComplete is called for delayedShareFetch2, since tp1 is common in between delayed share fetch
// requests, it should add a "check and complete" action for request key tp1 on the purgatory.
diff --git a/core/src/test/java/kafka/server/share/ShareFetchUtilsTest.java b/core/src/test/java/kafka/server/share/ShareFetchUtilsTest.java
index f2757e0f754..13248fb7110 100644
--- a/core/src/test/java/kafka/server/share/ShareFetchUtilsTest.java
+++ b/core/src/test/java/kafka/server/share/ShareFetchUtilsTest.java
@@ -30,6 +30,7 @@ import org.apache.kafka.common.record.FileRecords;
import org.apache.kafka.common.record.MemoryRecords;
import org.apache.kafka.common.record.SimpleRecord;
import org.apache.kafka.common.requests.FetchRequest;
+import org.apache.kafka.server.share.fetch.ShareAcquiredRecords;
import org.apache.kafka.server.share.fetch.ShareFetchData;
import org.apache.kafka.server.storage.log.FetchIsolation;
import org.apache.kafka.server.storage.log.FetchParams;
@@ -40,6 +41,7 @@ import org.mockito.Mockito;
import java.util.Collections;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Optional;
import java.util.OptionalInt;
@@ -54,6 +56,7 @@ import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doNothing;
@@ -68,9 +71,8 @@ public class ShareFetchUtilsTest {
public void testProcessFetchResponse() {
String groupId = "grp";
String memberId = Uuid.randomUuid().toString();
- Uuid topicId = Uuid.randomUuid();
- TopicIdPartition tp0 = new TopicIdPartition(topicId, new TopicPartition("foo", 0));
- TopicIdPartition tp1 = new TopicIdPartition(topicId, new TopicPartition("foo", 1));
+ TopicIdPartition tp0 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 0));
+ TopicIdPartition tp1 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 1));
Map<TopicIdPartition, Integer> partitionMaxBytes = new HashMap<>();
partitionMaxBytes.put(tp0, PARTITION_MAX_BYTES);
partitionMaxBytes.put(tp1, PARTITION_MAX_BYTES);
@@ -81,12 +83,12 @@ public class ShareFetchUtilsTest {
when(sp0.nextFetchOffset()).thenReturn((long) 3);
when(sp1.nextFetchOffset()).thenReturn((long) 3);
- when(sp0.acquire(any(), any())).thenReturn(
- Collections.singletonList(new ShareFetchResponseData.AcquiredRecords()
- .setFirstOffset(0).setLastOffset(3).setDeliveryCount((short) 1)));
- when(sp1.acquire(any(), any())).thenReturn(
- Collections.singletonList(new ShareFetchResponseData.AcquiredRecords()
- .setFirstOffset(100).setLastOffset(103).setDeliveryCount((short) 1)));
+ when(sp0.acquire(anyString(), anyInt(), any(FetchPartitionData.class))).thenReturn(
+ ShareAcquiredRecords.fromAcquiredRecords(new ShareFetchResponseData.AcquiredRecords()
+ .setFirstOffset(0).setLastOffset(3).setDeliveryCount((short) 1)));
+ when(sp1.acquire(anyString(), anyInt(), any(FetchPartitionData.class))).thenReturn(
+ ShareAcquiredRecords.fromAcquiredRecords(new ShareFetchResponseData.AcquiredRecords()
+ .setFirstOffset(100).setLastOffset(103).setDeliveryCount((short) 1)));
doNothing().when(sp1).updateCacheAndOffsets(any(Long.class));
doNothing().when(sp0).updateCacheAndOffsets(any(Long.class));
@@ -98,7 +100,7 @@ public class ShareFetchUtilsTest {
ShareFetchData shareFetchData = new ShareFetchData(
new FetchParams(ApiKeys.SHARE_FETCH.latestVersion(), FetchRequest.ORDINARY_CONSUMER_ID, -1, 0,
1, 1024 * 1024, FetchIsolation.HIGH_WATERMARK, Optional.empty()), groupId, memberId,
- new CompletableFuture<>(), partitionMaxBytes);
+ new CompletableFuture<>(), partitionMaxBytes, 100);
MemoryRecords records = MemoryRecords.withRecords(Compression.NONE,
new SimpleRecord("0".getBytes(), "v".getBytes()),
@@ -141,9 +143,8 @@ public class ShareFetchUtilsTest {
public void testProcessFetchResponseWithEmptyRecords() {
String groupId = "grp";
String memberId = Uuid.randomUuid().toString();
- Uuid topicId = Uuid.randomUuid();
- TopicIdPartition tp0 = new TopicIdPartition(topicId, new TopicPartition("foo", 0));
- TopicIdPartition tp1 = new TopicIdPartition(topicId, new TopicPartition("foo", 1));
+ TopicIdPartition tp0 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 0));
+ TopicIdPartition tp1 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 1));
Map<TopicIdPartition, Integer> partitionMaxBytes = new HashMap<>();
partitionMaxBytes.put(tp0, PARTITION_MAX_BYTES);
partitionMaxBytes.put(tp1, PARTITION_MAX_BYTES);
@@ -154,8 +155,8 @@ public class ShareFetchUtilsTest {
when(sp0.nextFetchOffset()).thenReturn((long) 3);
when(sp1.nextFetchOffset()).thenReturn((long) 3);
- when(sp0.acquire(any(), any())).thenReturn(Collections.emptyList());
- when(sp1.acquire(any(), any())).thenReturn(Collections.emptyList());
+ when(sp0.acquire(anyString(), anyInt(), any(FetchPartitionData.class))).thenReturn(ShareAcquiredRecords.empty());
+ when(sp1.acquire(anyString(), anyInt(), any(FetchPartitionData.class))).thenReturn(ShareAcquiredRecords.empty());
doNothing().when(sp1).updateCacheAndOffsets(any(Long.class));
doNothing().when(sp0).updateCacheAndOffsets(any(Long.class));
@@ -167,7 +168,7 @@ public class ShareFetchUtilsTest {
ShareFetchData shareFetchData = new ShareFetchData(
new FetchParams(ApiKeys.SHARE_FETCH.latestVersion(), FetchRequest.ORDINARY_CONSUMER_ID, -1, 0,
1, 1024 * 1024, FetchIsolation.HIGH_WATERMARK, Optional.empty()), groupId, memberId,
- new CompletableFuture<>(), partitionMaxBytes);
+ new CompletableFuture<>(), partitionMaxBytes, 100);
Map<TopicIdPartition, FetchPartitionData> responseData = new HashMap<>();
responseData.put(tp0, new FetchPartitionData(Errors.NONE, 0L, 0L,
@@ -193,9 +194,9 @@ public class ShareFetchUtilsTest {
@Test
public void testProcessFetchResponseWithLsoMovementForTopicPartition() {
String groupId = "grp";
- Uuid fooId = Uuid.randomUuid();
- TopicIdPartition tp0 = new TopicIdPartition(fooId, new TopicPartition("foo", 0));
- TopicIdPartition tp1 = new TopicIdPartition(fooId, new TopicPartition("foo", 1));
+
+ TopicIdPartition tp0 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 0));
+ TopicIdPartition tp1 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 1));
Map<TopicIdPartition, Integer> partitionMaxBytes = new HashMap<>();
partitionMaxBytes.put(tp0, PARTITION_MAX_BYTES);
@@ -211,7 +212,7 @@ public class ShareFetchUtilsTest {
ShareFetchData shareFetchData = new ShareFetchData(
new FetchParams(ApiKeys.SHARE_FETCH.latestVersion(), FetchRequest.ORDINARY_CONSUMER_ID, -1, 0,
1, 1024 * 1024, FetchIsolation.HIGH_WATERMARK, Optional.empty()),
- groupId, Uuid.randomUuid().toString(), new CompletableFuture<>(), partitionMaxBytes);
+ groupId, Uuid.randomUuid().toString(), new CompletableFuture<>(), partitionMaxBytes, 100);
ReplicaManager replicaManager = mock(ReplicaManager.class);
@@ -221,15 +222,14 @@ public class ShareFetchUtilsTest {
when(sp0.nextFetchOffset()).thenReturn((long) 0, (long) 5);
when(sp1.nextFetchOffset()).thenReturn((long) 4, (long) 4);
-
- when(sp0.acquire(anyString(), any(FetchPartitionData.class))).thenReturn(
- Collections.emptyList(),
- Collections.singletonList(new ShareFetchResponseData.AcquiredRecords()
- .setFirstOffset(0).setLastOffset(3).setDeliveryCount((short) 1)));
- when(sp1.acquire(anyString(), any(FetchPartitionData.class))).thenReturn(
- Collections.singletonList(new ShareFetchResponseData.AcquiredRecords()
- .setFirstOffset(100).setLastOffset(103).setDeliveryCount((short) 1)),
- Collections.emptyList());
+ when(sp0.acquire(anyString(), anyInt(), any(FetchPartitionData.class))).thenReturn(
+ ShareAcquiredRecords.empty(),
+ ShareAcquiredRecords.fromAcquiredRecords(new ShareFetchResponseData.AcquiredRecords()
+ .setFirstOffset(0).setLastOffset(3).setDeliveryCount((short) 1)));
+ when(sp1.acquire(anyString(), anyInt(), any(FetchPartitionData.class))).thenReturn(
+ ShareAcquiredRecords.fromAcquiredRecords(new ShareFetchResponseData.AcquiredRecords()
+ .setFirstOffset(100).setLastOffset(103).setDeliveryCount((short) 1)),
+ ShareAcquiredRecords.empty());
doNothing().when(sp1).updateCacheAndOffsets(any(Long.class));
doNothing().when(sp0).updateCacheAndOffsets(any(Long.class));
@@ -296,9 +296,8 @@ public class ShareFetchUtilsTest {
@Test
public void testProcessFetchResponseWhenNoRecordsAreAcquired() {
String groupId = "grp";
- Uuid fooId = Uuid.randomUuid();
- TopicIdPartition tp0 = new TopicIdPartition(fooId, new TopicPartition("foo", 0));
+ TopicIdPartition tp0 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 0));
Map<TopicIdPartition, Integer> partitionMaxBytes = Collections.singletonMap(tp0, PARTITION_MAX_BYTES);
SharePartition sp0 = Mockito.mock(SharePartition.class);
@@ -308,15 +307,14 @@ public class ShareFetchUtilsTest {
ShareFetchData shareFetchData = new ShareFetchData(
new FetchParams(ApiKeys.SHARE_FETCH.latestVersion(), FetchRequest.ORDINARY_CONSUMER_ID, -1, 0,
1, 1024 * 1024, FetchIsolation.HIGH_WATERMARK, Optional.empty()),
- groupId, Uuid.randomUuid().toString(), new CompletableFuture<>(), partitionMaxBytes);
+ groupId, Uuid.randomUuid().toString(), new CompletableFuture<>(), partitionMaxBytes, 100);
ReplicaManager replicaManager = mock(ReplicaManager.class);
// Mock the replicaManager.fetchOffsetForTimestamp method to return a timestamp and offset for the topic partition.
FileRecords.TimestampAndOffset timestampAndOffset = new FileRecords.TimestampAndOffset(100L, 1L, Optional.empty());
doReturn(new OffsetResultHolder(Option.apply(timestampAndOffset), Option.empty())).when(replicaManager).fetchOffsetForTimestamp(any(TopicPartition.class), anyLong(), any(), any(), anyBoolean());
-
- when(sp0.acquire(anyString(), any(FetchPartitionData.class))).thenReturn(Collections.emptyList());
+ when(sp0.acquire(anyString(), anyInt(), any(FetchPartitionData.class))).thenReturn(ShareAcquiredRecords.empty());
doNothing().when(sp0).updateCacheAndOffsets(any(Long.class));
MemoryRecords records = MemoryRecords.withRecords(Compression.NONE,
@@ -358,4 +356,80 @@ public class ShareFetchUtilsTest {
Mockito.verify(sp0, times(1)).updateCacheAndOffsets(1L);
}
+
+ @Test
+ public void testProcessFetchResponseWithMaxFetchRecords() {
+ String groupId = "grp";
+ TopicIdPartition tp0 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 0));
+ TopicIdPartition tp1 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 1));
+
+ Map<TopicIdPartition, Integer> partitionMaxBytes = new HashMap<>();
+ partitionMaxBytes.put(tp0, PARTITION_MAX_BYTES);
+ partitionMaxBytes.put(tp1, PARTITION_MAX_BYTES);
+
+ SharePartition sp0 = Mockito.mock(SharePartition.class);
+ SharePartition sp1 = Mockito.mock(SharePartition.class);
+
+ SharePartitionManager sharePartitionManager = mock(SharePartitionManager.class);
+ when(sharePartitionManager.sharePartition(groupId, tp0)).thenReturn(sp0);
+ when(sharePartitionManager.sharePartition(groupId, tp1)).thenReturn(sp1);
+
+ when(sp0.nextFetchOffset()).thenReturn((long) 0, (long) 5);
+ when(sp1.nextFetchOffset()).thenReturn((long) 4, (long) 4);
+
+ Uuid memberId = Uuid.randomUuid();
+ // Set max fetch records to 10
+ ShareFetchData shareFetchData = new ShareFetchData(
+ new FetchParams(ApiKeys.SHARE_FETCH.latestVersion(), FetchRequest.ORDINARY_CONSUMER_ID, -1, 0,
+ 1, 1024 * 1024, FetchIsolation.HIGH_WATERMARK, Optional.empty()),
+ groupId, memberId.toString(), new CompletableFuture<>(), partitionMaxBytes, 10);
+
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+ // Mock the replicaManager.fetchOffsetForTimestamp method to return a timestamp and offset for the topic partition.
+ FileRecords.TimestampAndOffset timestampAndOffset = new FileRecords.TimestampAndOffset(100L, 1L, Optional.empty());
+ doReturn(new OffsetResultHolder(Option.apply(timestampAndOffset), Option.empty())).when(replicaManager).fetchOffsetForTimestamp(any(TopicPartition.class), anyLong(), any(), any(), anyBoolean());
+
+ MemoryRecords records1 = MemoryRecords.withRecords(Compression.NONE,
+ new SimpleRecord("0".getBytes(), "v".getBytes()),
+ new SimpleRecord("1".getBytes(), "v".getBytes()),
+ new SimpleRecord("2".getBytes(), "v".getBytes()),
+ new SimpleRecord(null, "value".getBytes()));
+
+ FetchPartitionData fetchPartitionData1 = new FetchPartitionData(Errors.NONE, 0L, 0L,
+ records1, Optional.empty(), OptionalLong.empty(), Optional.empty(),
+ OptionalInt.empty(), false);
+ FetchPartitionData fetchPartitionData2 = new FetchPartitionData(Errors.NONE, 0L, 0L,
+ records1, Optional.empty(), OptionalLong.empty(), Optional.empty(),
+ OptionalInt.empty(), false);
+
+ when(sp0.acquire(memberId.toString(), 10, fetchPartitionData1)).thenReturn(
+ ShareAcquiredRecords.fromAcquiredRecords(new ShareFetchResponseData.AcquiredRecords()
+ .setFirstOffset(0).setLastOffset(1).setDeliveryCount((short) 1)));
+ when(sp1.acquire(memberId.toString(), 8, fetchPartitionData2)).thenReturn(
+ ShareAcquiredRecords.fromAcquiredRecords(new ShareFetchResponseData.AcquiredRecords()
+ .setFirstOffset(100).setLastOffset(103).setDeliveryCount((short) 1)));
+
+ // Send the topic partitions in order so can validate if correct mock is called, accounting
+ // the offset count for the acquired records from the previous share partition acquire.
+ Map<TopicIdPartition, FetchPartitionData> responseData1 = new LinkedHashMap<>();
+ responseData1.put(tp0, fetchPartitionData1);
+ responseData1.put(tp1, fetchPartitionData2);
+
+ Map<TopicIdPartition, ShareFetchResponseData.PartitionData> resultData1 =
+ ShareFetchUtils.processFetchResponse(shareFetchData, responseData1, sharePartitionManager, replicaManager);
+
+ assertEquals(2, resultData1.size());
+ assertTrue(resultData1.containsKey(tp0));
+ assertTrue(resultData1.containsKey(tp1));
+ assertEquals(0, resultData1.get(tp0).partitionIndex());
+ assertEquals(1, resultData1.get(tp1).partitionIndex());
+ assertEquals(Errors.NONE.code(), resultData1.get(tp0).errorCode());
+ assertEquals(Errors.NONE.code(), resultData1.get(tp1).errorCode());
+ assertEquals(1, resultData1.get(tp0).acquiredRecords().size());
+ assertEquals(0, resultData1.get(tp0).acquiredRecords().get(0).firstOffset());
+ assertEquals(1, resultData1.get(tp0).acquiredRecords().get(0).lastOffset());
+ assertEquals(1, resultData1.get(tp1).acquiredRecords().size());
+ assertEquals(100, resultData1.get(tp1).acquiredRecords().get(0).firstOffset());
+ assertEquals(103, resultData1.get(tp1).acquiredRecords().get(0).lastOffset());
+ }
}
diff --git a/core/src/test/java/kafka/server/share/SharePartitionManagerTest.java b/core/src/test/java/kafka/server/share/SharePartitionManagerTest.java
index 915480ed418..0ef58ef8491 100644
--- a/core/src/test/java/kafka/server/share/SharePartitionManagerTest.java
+++ b/core/src/test/java/kafka/server/share/SharePartitionManagerTest.java
@@ -128,6 +128,7 @@ public class SharePartitionManagerTest {
private static final int DEFAULT_RECORD_LOCK_DURATION_MS = 30000;
private static final int MAX_DELIVERY_COUNT = 5;
private static final short MAX_IN_FLIGHT_MESSAGES = 200;
+ 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 FetchParams FETCH_PARAMS = new FetchParams(ApiKeys.SHARE_FETCH.latestVersion(),
@@ -1640,7 +1641,8 @@ public class SharePartitionManagerTest {
groupId,
Uuid.randomUuid().toString(),
new CompletableFuture<>(),
- partitionMaxBytes);
+ partitionMaxBytes,
+ 100);
ReplicaManager replicaManager = mock(ReplicaManager.class);
DelayedOperationPurgatory<DelayedShareFetch> delayedShareFetchPurgatory = new DelayedOperationPurgatory<>(
@@ -1735,7 +1737,8 @@ public class SharePartitionManagerTest {
groupId,
Uuid.randomUuid().toString(),
new CompletableFuture<>(),
- partitionMaxBytes);
+ partitionMaxBytes,
+ 100);
ReplicaManager replicaManager = mock(ReplicaManager.class);
DelayedOperationPurgatory<DelayedShareFetch> delayedShareFetchPurgatory = new DelayedOperationPurgatory<>(
@@ -1827,7 +1830,8 @@ public class SharePartitionManagerTest {
groupId,
Uuid.randomUuid().toString(),
new CompletableFuture<>(),
- partitionMaxBytes);
+ partitionMaxBytes,
+ 100);
ReplicaManager replicaManager = mock(ReplicaManager.class);
DelayedOperationPurgatory<DelayedShareFetch> delayedShareFetchPurgatory = new DelayedOperationPurgatory<>(
@@ -1926,7 +1930,8 @@ public class SharePartitionManagerTest {
groupId,
Uuid.randomUuid().toString(),
new CompletableFuture<>(),
- partitionMaxBytes);
+ partitionMaxBytes,
+ 100);
ReplicaManager replicaManager = mock(ReplicaManager.class);
DelayedOperationPurgatory<DelayedShareFetch> delayedShareFetchPurgatory = new DelayedOperationPurgatory<>(
@@ -2317,6 +2322,7 @@ public class SharePartitionManagerTest {
timer,
MAX_DELIVERY_COUNT,
MAX_IN_FLIGHT_MESSAGES,
+ MAX_FETCH_RECORDS,
persister,
mock(GroupConfigManager.class),
metrics);
diff --git a/core/src/test/java/kafka/server/share/SharePartitionTest.java b/core/src/test/java/kafka/server/share/SharePartitionTest.java
index 0d20a828081..1ef17b44eba 100644
--- a/core/src/test/java/kafka/server/share/SharePartitionTest.java
+++ b/core/src/test/java/kafka/server/share/SharePartitionTest.java
@@ -40,6 +40,7 @@ import org.apache.kafka.common.utils.Time;
import org.apache.kafka.coordinator.group.GroupConfig;
import org.apache.kafka.coordinator.group.GroupConfigManager;
import org.apache.kafka.server.share.acknowledge.ShareAcknowledgementBatch;
+import org.apache.kafka.server.share.fetch.ShareAcquiredRecords;
import org.apache.kafka.server.share.persister.NoOpShareStatePersister;
import org.apache.kafka.server.share.persister.PartitionFactory;
import org.apache.kafka.server.share.persister.Persister;
@@ -97,6 +98,7 @@ public class SharePartitionTest {
private static final short MAX_IN_FLIGHT_MESSAGES = 200;
private static final int ACQUISITION_LOCK_TIMEOUT_MS = 100;
private static final int DEFAULT_MAX_WAIT_ACQUISITION_LOCK_TIMEOUT_MS = 300;
+ private static final int MAX_FETCH_RECORDS = Integer.MAX_VALUE;
@BeforeEach
public void setUp() {
@@ -542,10 +544,12 @@ public class SharePartitionTest {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
MemoryRecords records = memoryRecords(1);
- List<AcquiredRecords> acquiredRecordsList = sharePartition.acquire(
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 3, 0, records,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 1);
assertArrayEquals(expectedAcquiredRecords(records, 1).toArray(), acquiredRecordsList.toArray());
assertEquals(1, sharePartition.nextFetchOffset());
@@ -563,10 +567,12 @@ public class SharePartitionTest {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
MemoryRecords records = memoryRecords(5, 10);
- List<AcquiredRecords> acquiredRecordsList = sharePartition.acquire(
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 3, records,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
assertArrayEquals(expectedAcquiredRecords(records, 1).toArray(), acquiredRecordsList.toArray());
assertEquals(15, sharePartition.nextFetchOffset());
@@ -579,25 +585,107 @@ public class SharePartitionTest {
assertNull(sharePartition.cachedState().get(10L).offsetState());
}
+ @Test
+ public void testAcquireWithMaxFetchRecords() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().build();
+ // Less-number of records than max fetch records.
+ MemoryRecords records = memoryRecords(5);
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 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());
+
+ // More-number of records than max fetch records, but from 0 offset hence previous 10 records
+ // should be ignored and new full batch till end should be acquired.
+ records = memoryRecords(25);
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 10,
+ new FetchPartitionData(Errors.NONE, 20, 0, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 20);
+
+ assertArrayEquals(expectedAcquiredRecord(5, 24, 1).toArray(), acquiredRecordsList.toArray());
+ assertEquals(25, sharePartition.nextFetchOffset());
+ assertEquals(2, sharePartition.cachedState().size());
+ assertEquals(0, sharePartition.cachedState().get(0L).firstOffset());
+ assertEquals(5, sharePartition.cachedState().get(5L).firstOffset());
+ assertEquals(24, sharePartition.cachedState().get(5L).lastOffset());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(5L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(5L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(5L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(5L).offsetState());
+ }
+
+ @Test
+ public void testAcquireWithMultipleBatchesAndMaxFetchRecords() {
+ SharePartition sharePartition = SharePartitionBuilder.builder().build();
+
+ // Create 3 batches of records.
+ ByteBuffer buffer = ByteBuffer.allocate(4096);
+ memoryRecordsBuilder(buffer, 5, 10).close();
+ memoryRecordsBuilder(buffer, 15, 15).close();
+ memoryRecordsBuilder(buffer, 15, 30).close();
+
+ buffer.flip();
+
+ MemoryRecords records = MemoryRecords.readableRecords(buffer);
+ // Acquire 10 records.
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
+ MEMBER_ID,
+ 10,
+ new FetchPartitionData(Errors.NONE, 20, 10, records,
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 20);
+
+ // Validate 2 batches are fetched one with 5 records and other till end of batch, third batch
+ // should be skipped.
+ assertArrayEquals(expectedAcquiredRecord(10, 29, 1).toArray(), acquiredRecordsList.toArray());
+ assertEquals(30, sharePartition.nextFetchOffset());
+ assertEquals(1, sharePartition.cachedState().size());
+ assertEquals(10, sharePartition.cachedState().get(10L).firstOffset());
+ assertEquals(29, sharePartition.cachedState().get(10L).lastOffset());
+ assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(10L).batchState());
+ assertEquals(MEMBER_ID, sharePartition.cachedState().get(10L).batchMemberId());
+ assertEquals(1, sharePartition.cachedState().get(10L).batchDeliveryCount());
+ assertNull(sharePartition.cachedState().get(10L).offsetState());
+ }
+
@Test
public void testAcquireMultipleRecordsWithOverlapAndNewBatch() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
MemoryRecords records = memoryRecords(5, 0);
- List<AcquiredRecords> acquiredRecordsList = sharePartition.acquire(
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 3, records,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
assertArrayEquals(expectedAcquiredRecords(records, 1).toArray(), acquiredRecordsList.toArray());
assertEquals(5, sharePartition.nextFetchOffset());
// Add records from 0-9 offsets, 5-9 should be acquired and 0-4 should be ignored.
records = memoryRecords(10, 0);
- acquiredRecordsList = sharePartition.acquire(
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 3, records,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
assertArrayEquals(expectedAcquiredRecords(memoryRecords(5, 5), 1).toArray(), acquiredRecordsList.toArray());
assertEquals(10, sharePartition.nextFetchOffset());
@@ -609,18 +697,22 @@ public class SharePartitionTest {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
MemoryRecords records = memoryRecords(5, 10);
- List<AcquiredRecords> acquiredRecordsList = sharePartition.acquire(
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 3, records,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
assertArrayEquals(expectedAcquiredRecords(records, 1).toArray(), acquiredRecordsList.toArray());
assertEquals(15, sharePartition.nextFetchOffset());
- acquiredRecordsList = sharePartition.acquire(
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 3, records,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 0);
// No records should be returned as the batch is already acquired.
assertEquals(0, acquiredRecordsList.size());
@@ -628,10 +720,12 @@ public class SharePartitionTest {
// Send subset of the same batch again, no records should be returned.
MemoryRecords subsetRecords = memoryRecords(2, 10);
- acquiredRecordsList = sharePartition.acquire(
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 3, subsetRecords,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 0);
// No records should be returned as the batch is already acquired.
assertEquals(0, acquiredRecordsList.size());
@@ -643,10 +737,12 @@ public class SharePartitionTest {
@Test
public void testAcquireWithEmptyFetchRecords() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- List<AcquiredRecords> acquiredRecordsList = sharePartition.acquire(
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 3, MemoryRecords.EMPTY,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 0);
assertEquals(0, acquiredRecordsList.size());
assertEquals(0, sharePartition.nextFetchOffset());
@@ -663,6 +759,7 @@ public class SharePartitionTest {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 3, memoryRecords(5),
Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
assertEquals(5, sharePartition.nextFetchOffset());
@@ -684,6 +781,7 @@ public class SharePartitionTest {
assertTrue(sharePartition.findNextFetchOffset());
sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 3, memoryRecords(5),
Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
assertEquals(5, sharePartition.nextFetchOffset());
@@ -701,6 +799,7 @@ public class SharePartitionTest {
SharePartition sharePartition = SharePartitionBuilder.builder().withMaxInflightMessages(6).build();
sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 3, memoryRecords(5),
Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
// Limit not reached as only 6 in-flight messages is the limit.
@@ -712,6 +811,7 @@ public class SharePartitionTest {
SharePartition sharePartition = SharePartitionBuilder.builder().withMaxInflightMessages(1).build();
sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 3, memoryRecords(5),
Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
// Limit reached as only one in-flight message is the limit.
@@ -739,17 +839,21 @@ public class SharePartitionTest {
MemoryRecords records2 = memoryRecords(1, 1);
// Another batch is acquired because if there is only 1 batch, and it is acknowledged, the batch will be removed from cachedState
- List<AcquiredRecords> acquiredRecordsList = sharePartition.acquire(
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 10, 0, records1,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 1);
assertEquals(1, acquiredRecordsList.size());
- acquiredRecordsList = sharePartition.acquire(
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 10, 0, records2,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 1);
assertEquals(1, acquiredRecordsList.size());
@@ -771,10 +875,12 @@ public class SharePartitionTest {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
MemoryRecords records = memoryRecords(10, 5);
- List<AcquiredRecords> acquiredRecordsList = sharePartition.acquire(
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 0, records,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 10);
assertEquals(1, acquiredRecordsList.size());
@@ -798,18 +904,22 @@ public class SharePartitionTest {
recordsBuilder.appendWithOffset(18, 0L, TestUtils.randomString(10).getBytes(), TestUtils.randomString(10).getBytes());
MemoryRecords records2 = recordsBuilder.build();
- List<AcquiredRecords> acquiredRecordsList = sharePartition.acquire(
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 30, 0, records1,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 2);
assertArrayEquals(expectedAcquiredRecords(records1, 1).toArray(), acquiredRecordsList.toArray());
assertEquals(7, sharePartition.nextFetchOffset());
- acquiredRecordsList = sharePartition.acquire(
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 30, 0, records2,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 9);
assertArrayEquals(expectedAcquiredRecords(records2, 1).toArray(), acquiredRecordsList.toArray());
assertEquals(19, sharePartition.nextFetchOffset());
@@ -861,18 +971,22 @@ public class SharePartitionTest {
recordsBuilder.appendWithOffset(20, 0L, TestUtils.randomString(10).getBytes(), TestUtils.randomString(10).getBytes());
MemoryRecords records2 = recordsBuilder.build();
- List<AcquiredRecords> acquiredRecordsList = sharePartition.acquire(
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 30, 0, records1,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 2);
assertArrayEquals(expectedAcquiredRecords(records1, 1).toArray(), acquiredRecordsList.toArray());
assertEquals(7, sharePartition.nextFetchOffset());
- acquiredRecordsList = sharePartition.acquire(
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 30, 0, records2,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 11);
assertArrayEquals(expectedAcquiredRecords(records2, 1).toArray(), acquiredRecordsList.toArray());
assertEquals(21, sharePartition.nextFetchOffset());
@@ -926,10 +1040,12 @@ public class SharePartitionTest {
assertFutureThrows(ackResult, InvalidRecordStateException.class);
MemoryRecords records = memoryRecords(5, 5);
- List<AcquiredRecords> acquiredRecordsList = sharePartition.acquire(
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 0, records,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
assertEquals(1, acquiredRecordsList.size());
// Cached data with offset 5-9 should exist.
@@ -949,19 +1065,23 @@ public class SharePartitionTest {
// Create data for the batch with offsets 0-4.
MemoryRecords records = memoryRecords(5, 0);
- List<AcquiredRecords> acquiredRecordsList = sharePartition.acquire(
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 0, records,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
assertEquals(1, acquiredRecordsList.size());
// Create data for the batch with offsets 20-24.
records = memoryRecords(5, 20);
- acquiredRecordsList = sharePartition.acquire(
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 0, records,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
assertEquals(1, acquiredRecordsList.size());
@@ -976,10 +1096,12 @@ public class SharePartitionTest {
// Create data for the batch with offsets 5-10.
records = memoryRecords(6, 5);
- acquiredRecordsList = sharePartition.acquire(
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 0, records,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 6);
assertEquals(1, acquiredRecordsList.size());
@@ -994,10 +1116,12 @@ public class SharePartitionTest {
public void testAcknowledgeWithAnotherMember() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
MemoryRecords records = memoryRecords(5, 5);
- List<AcquiredRecords> acquiredRecordsList = sharePartition.acquire(
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 0, records,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
assertEquals(1, acquiredRecordsList.size());
// Cached data with offset 5-9 should exist.
@@ -1015,10 +1139,12 @@ public class SharePartitionTest {
public void testAcknowledgeWhenOffsetNotAcquired() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
MemoryRecords records = memoryRecords(5, 5);
- List<AcquiredRecords> acquiredRecordsList = sharePartition.acquire(
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 0, records,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
assertEquals(1, acquiredRecordsList.size());
// Cached data with offset 5-9 should exist.
@@ -1039,10 +1165,12 @@ public class SharePartitionTest {
assertFutureThrows(ackResult, InvalidRecordStateException.class);
// Re-acquire the same batch and then acknowledge subset with ACCEPT type.
- acquiredRecordsList = sharePartition.acquire(
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 0, records,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
assertEquals(1, acquiredRecordsList.size());
@@ -1066,24 +1194,30 @@ public class SharePartitionTest {
MemoryRecords records1 = memoryRecords(5, 5);
MemoryRecords records2 = memoryRecords(5, 10);
MemoryRecords records3 = memoryRecords(5, 15);
- List<AcquiredRecords> acquiredRecordsList = sharePartition.acquire(
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 0, records1,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
assertEquals(1, acquiredRecordsList.size());
- acquiredRecordsList = sharePartition.acquire(
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 0, records2,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
assertEquals(1, acquiredRecordsList.size());
- acquiredRecordsList = sharePartition.acquire(
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 0, records3,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
assertEquals(1, acquiredRecordsList.size());
// Cached data with offset 5-19 should exist.
@@ -1113,24 +1247,30 @@ public class SharePartitionTest {
MemoryRecords records1 = memoryRecords(5, 5);
MemoryRecords records2 = memoryRecords(5, 10);
MemoryRecords records3 = memoryRecords(5, 15);
- List<AcquiredRecords> acquiredRecordsList = sharePartition.acquire(
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 0, records1,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
assertEquals(1, acquiredRecordsList.size());
- acquiredRecordsList = sharePartition.acquire(
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 0, records2,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
assertEquals(1, acquiredRecordsList.size());
- acquiredRecordsList = sharePartition.acquire(
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 0, records3,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
assertEquals(1, acquiredRecordsList.size());
// Cached data with offset 5-19 should exist.
@@ -1161,10 +1301,12 @@ public class SharePartitionTest {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
MemoryRecords records = memoryRecords(5, 10);
- List<AcquiredRecords> acquiredRecordsList = sharePartition.acquire(
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 3, records,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
assertArrayEquals(expectedAcquiredRecords(records, 1).toArray(), acquiredRecordsList.toArray());
assertEquals(15, sharePartition.nextFetchOffset());
@@ -1188,10 +1330,12 @@ public class SharePartitionTest {
assertEquals(expectedOffsetStateMap, sharePartition.cachedState().get(10L).offsetState());
// Send the same fetch request batch again but only 2 offsets should come as acquired.
- acquiredRecordsList = sharePartition.acquire(
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 3, records,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 2);
assertArrayEquals(expectedAcquiredRecords(12, 13, 2).toArray(), acquiredRecordsList.toArray());
assertEquals(15, sharePartition.nextFetchOffset());
@@ -1209,34 +1353,42 @@ public class SharePartitionTest {
// Fourth fetch request with 5 records starting from offset 28.
MemoryRecords records4 = memoryRecords(5, 28);
- List<AcquiredRecords> acquiredRecordsList = sharePartition.acquire(
+ List<AcquiredRecords> acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 40, 3, records1,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
assertArrayEquals(expectedAcquiredRecords(records1, 1).toArray(), acquiredRecordsList.toArray());
assertEquals(15, sharePartition.nextFetchOffset());
- acquiredRecordsList = sharePartition.acquire(
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 30, 3, records2,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
assertArrayEquals(expectedAcquiredRecords(records2, 1).toArray(), acquiredRecordsList.toArray());
assertEquals(20, sharePartition.nextFetchOffset());
- acquiredRecordsList = sharePartition.acquire(
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 30, 3, records3,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
assertArrayEquals(expectedAcquiredRecords(records3, 1).toArray(), acquiredRecordsList.toArray());
assertEquals(28, sharePartition.nextFetchOffset());
- acquiredRecordsList = sharePartition.acquire(
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 30, 3, records4,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
assertArrayEquals(expectedAcquiredRecords(records4, 1).toArray(), acquiredRecordsList.toArray());
assertEquals(33, sharePartition.nextFetchOffset());
@@ -1286,30 +1438,36 @@ public class SharePartitionTest {
assertEquals(expectedOffsetStateMap, sharePartition.cachedState().get(28L).offsetState());
// Send next batch from offset 12, only 3 records should be acquired.
- acquiredRecordsList = sharePartition.acquire(
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 40, 3, records1,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 3);
assertArrayEquals(expectedAcquiredRecords(12, 14, 2).toArray(), acquiredRecordsList.toArray());
assertEquals(15, sharePartition.nextFetchOffset());
// Though record2 batch exists to acquire but send batch record3, it should be acquired but
// next fetch offset should not move.
- acquiredRecordsList = sharePartition.acquire(
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 40, 3, records3,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 5);
assertArrayEquals(expectedAcquiredRecords(records3, 2).toArray(), acquiredRecordsList.toArray());
assertEquals(15, sharePartition.nextFetchOffset());
// Acquire partial records from batch 2.
MemoryRecords subsetRecords = memoryRecords(2, 17);
- acquiredRecordsList = sharePartition.acquire(
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 3, subsetRecords,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 2);
assertArrayEquals(expectedAcquiredRecords(17, 18, 2).toArray(), acquiredRecordsList.toArray());
// Next fetch offset should not move.
@@ -1318,10 +1476,12 @@ public class SharePartitionTest {
// Acquire partial records from record 4 to further test if the next fetch offset move
// accordingly once complete record 2 is also acquired.
subsetRecords = memoryRecords(1, 28);
- acquiredRecordsList = sharePartition.acquire(
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 3, subsetRecords,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 1);
assertArrayEquals(expectedAcquiredRecords(28, 28, 2).toArray(), acquiredRecordsList.toArray());
// Next fetch offset should not move.
@@ -1329,10 +1489,12 @@ public class SharePartitionTest {
// Try to acquire complete record 2 though it's already partially acquired, the next fetch
// offset should move.
- acquiredRecordsList = sharePartition.acquire(
+ acquiredRecordsList = fetchAcquiredRecords(sharePartition.acquire(
MEMBER_ID,
+ MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 3, records2,
- Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
+ Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false)),
+ 3);
// Offset 15,16 and 19 should be acquired.
List<AcquiredRecords> expectedAcquiredRecords = expectedAcquiredRecords(15, 16, 2);
@@ -1345,7 +1507,7 @@ public class SharePartitionTest {
@Test
public void testAcquisitionLockForAcquiringSingleRecord() throws InterruptedException {
SharePartition sharePartition = SharePartitionBuilder.builder().withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS).build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 3, 0, memoryRecords(1),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 3, 0, memoryRecords(1),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -1366,7 +1528,7 @@ public class SharePartitionTest {
@Test
public void testAcquisitionLockForAcquiringMultipleRecords() throws InterruptedException {
SharePartition sharePartition = SharePartitionBuilder.builder().withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS).build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 3, memoryRecords(5, 10),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 3, memoryRecords(5, 10),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -1388,7 +1550,7 @@ public class SharePartitionTest {
public void testAcquisitionLockForAcquiringMultipleRecordsWithOverlapAndNewBatch() throws InterruptedException {
SharePartition sharePartition = SharePartitionBuilder.builder().withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS).build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 3, memoryRecords(5, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 3, memoryRecords(5, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -1396,7 +1558,7 @@ public class SharePartitionTest {
assertEquals(1, sharePartition.timer().size());
// Add records from 0-9 offsets, 5-9 should be acquired and 0-4 should be ignored.
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 3, memoryRecords(10, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 3, memoryRecords(10, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -1420,7 +1582,7 @@ public class SharePartitionTest {
public void testAcquisitionLockForAcquiringSameBatchAgain() throws InterruptedException {
SharePartition sharePartition = SharePartitionBuilder.builder().withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS).build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 3, memoryRecords(5, 10),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 3, memoryRecords(5, 10),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -1436,7 +1598,7 @@ public class SharePartitionTest {
() -> ACQUISITION_LOCK_NEVER_GOT_RELEASED);
// Acquire the same batch again.
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 3, memoryRecords(5, 10),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 3, memoryRecords(5, 10),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -1449,7 +1611,7 @@ public class SharePartitionTest {
public void testAcquisitionLockOnAcknowledgingSingleRecordBatch() throws InterruptedException {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 10, 0, memoryRecords(1, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 10, 0, memoryRecords(1, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -1483,7 +1645,7 @@ public class SharePartitionTest {
public void testAcquisitionLockOnAcknowledgingMultipleRecordBatch() throws InterruptedException {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(10, 5),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(10, 5),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -1521,19 +1683,19 @@ public class SharePartitionTest {
MemoryRecords records2 = recordsBuilder.build();
MemoryRecords records3 = memoryRecords(2, 1);
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 30, 0, records3,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 0, records3,
Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
assertNotNull(sharePartition.cachedState().get(1L).batchAcquisitionLockTimeoutTask());
assertEquals(1, sharePartition.timer().size());
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 30, 0, records1,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 0, records1,
Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
assertNotNull(sharePartition.cachedState().get(5L).batchAcquisitionLockTimeoutTask());
assertEquals(2, sharePartition.timer().size());
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 30, 0, records2,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 0, records2,
Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
assertNotNull(sharePartition.cachedState().get(10L).batchAcquisitionLockTimeoutTask());
@@ -1568,7 +1730,7 @@ public class SharePartitionTest {
public void testAcquisitionLockForAcquiringSubsetBatchAgain() throws InterruptedException {
SharePartition sharePartition = SharePartitionBuilder.builder().withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS).build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 3, memoryRecords(8, 10),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 3, memoryRecords(8, 10),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -1586,7 +1748,7 @@ public class SharePartitionTest {
() -> ACQUISITION_LOCK_NEVER_GOT_RELEASED);
// Acquire subset of records again.
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 3, memoryRecords(3, 12),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 3, memoryRecords(3, 12),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -1644,14 +1806,14 @@ public class SharePartitionTest {
recordsBuilder.appendWithOffset(20, 0L, TestUtils.randomString(10).getBytes(), TestUtils.randomString(10).getBytes());
MemoryRecords records2 = recordsBuilder.build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 30, 0, records1,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 0, records1,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
assertNotNull(sharePartition.cachedState().get(5L).batchAcquisitionLockTimeoutTask());
assertEquals(1, sharePartition.timer().size());
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 30, 0, records2,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 0, records2,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -1736,11 +1898,11 @@ public class SharePartitionTest {
// Adding memoryRecords(10, 0) in the sharePartition to make sure that SPSO doesn't move forward when delivery count of records2
// exceed the max delivery count.
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 3, 0, memoryRecords(10, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 3, 0, memoryRecords(10, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 3, 0, memoryRecords(10, 10),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 3, 0, memoryRecords(10, 10),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -1758,7 +1920,7 @@ public class SharePartitionTest {
DEFAULT_MAX_WAIT_ACQUISITION_LOCK_TIMEOUT_MS,
() -> ACQUISITION_LOCK_NEVER_GOT_RELEASED);
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 3, 0, memoryRecords(10, 10),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 3, 0, memoryRecords(10, 10),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -1786,7 +1948,7 @@ public class SharePartitionTest {
.withMaxDeliveryCount(2) // Only 2 delivery attempts will be made before archiving the records
.build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 3, 0, memoryRecords(10, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 3, 0, memoryRecords(10, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -1803,7 +1965,7 @@ public class SharePartitionTest {
DEFAULT_MAX_WAIT_ACQUISITION_LOCK_TIMEOUT_MS,
() -> ACQUISITION_LOCK_NEVER_GOT_RELEASED);
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 3, 0, memoryRecords(5, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 3, 0, memoryRecords(5, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -1862,7 +2024,7 @@ public class SharePartitionTest {
.withMaxDeliveryCount(2) // Only 2 delivery attempts will be made before archiving the records
.build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 3, 0, memoryRecords(10, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 3, 0, memoryRecords(10, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -1878,7 +2040,7 @@ public class SharePartitionTest {
DEFAULT_MAX_WAIT_ACQUISITION_LOCK_TIMEOUT_MS,
() -> ACQUISITION_LOCK_NEVER_GOT_RELEASED);
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 3, 0, memoryRecords(10, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 3, 0, memoryRecords(10, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -1900,7 +2062,7 @@ public class SharePartitionTest {
public void testAcknowledgeAfterAcquisitionLockTimeout() throws InterruptedException {
SharePartition sharePartition = SharePartitionBuilder.builder().withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS).build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 5),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 5),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -1937,7 +2099,7 @@ public class SharePartitionTest {
public void testAcquisitionLockAfterDifferentAcknowledges() throws InterruptedException {
SharePartition sharePartition = SharePartitionBuilder.builder().withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS).build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 5),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 5),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -2003,7 +2165,7 @@ public class SharePartitionTest {
PartitionFactory.newPartitionErrorData(0, Errors.GROUP_ID_NOT_FOUND.code(), Errors.GROUP_ID_NOT_FOUND.message())))));
Mockito.when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(10, 5),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(10, 5),
Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
assertEquals(1, sharePartition.timer().size());
@@ -2035,7 +2197,7 @@ public class SharePartitionTest {
PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
Mockito.when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(6, 5),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(6, 5),
Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
assertEquals(1, sharePartition.timer().size());
@@ -2077,7 +2239,7 @@ public class SharePartitionTest {
public void testReleaseSingleRecordBatch() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 10, 0, memoryRecords(1, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 10, 0, memoryRecords(1, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -2096,7 +2258,7 @@ public class SharePartitionTest {
public void testReleaseMultipleRecordBatch() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(10, 5),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(10, 5),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -2119,15 +2281,15 @@ public class SharePartitionTest {
// Untracked gap of 3 offsets from 7-9.
MemoryRecords records2 = memoryRecords(9, 10);
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 30, 0, records0,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 0, records0,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 30, 0, records1,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 0, records1,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 30, 0, records2,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 0, records2,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -2159,11 +2321,11 @@ public class SharePartitionTest {
recordsBuilder.appendWithOffset(20, 0L, TestUtils.randomString(10).getBytes(), TestUtils.randomString(10).getBytes());
MemoryRecords records2 = recordsBuilder.build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 30, 0, records1,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 0, records1,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 30, 0, records2,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 0, records2,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -2215,10 +2377,10 @@ public class SharePartitionTest {
recordsBuilder.appendWithOffset(20, 0L, TestUtils.randomString(10).getBytes(), TestUtils.randomString(10).getBytes());
MemoryRecords records2 = recordsBuilder.build();
- sharePartition.acquire("member-2", new FetchPartitionData(Errors.NONE, 30, 0, records1,
+ sharePartition.acquire("member-2", MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 0, records1,
Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 30, 0, records2,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 0, records2,
Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
// Acknowledging over subset of second batch with subset of gap offsets.
@@ -2288,10 +2450,10 @@ public class SharePartitionTest {
recordsBuilder.appendWithOffset(20, 0L, TestUtils.randomString(10).getBytes(), TestUtils.randomString(10).getBytes());
MemoryRecords records2 = recordsBuilder.build();
- sharePartition.acquire("member-2", new FetchPartitionData(Errors.NONE, 30, 0, records1,
+ sharePartition.acquire("member-2", MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 0, records1,
Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 30, 0, records2,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 0, records2,
Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
// Acknowledging over subset of second batch with subset of gap offsets.
@@ -2367,7 +2529,7 @@ public class SharePartitionTest {
@Test
public void testReleaseAcquiredRecordsAfterDifferentAcknowledges() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 5),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 5),
Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
sharePartition.acknowledge(MEMBER_ID,
@@ -2394,19 +2556,19 @@ public class SharePartitionTest {
public void testMaxDeliveryCountLimitExceededForRecordsSubsetAfterReleaseAcquiredRecords() {
SharePartition sharePartition = SharePartitionBuilder.builder().withMaxDeliveryCount(2).build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 40, 3, memoryRecords(10, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 40, 3, memoryRecords(10, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
MemoryRecords records2 = memoryRecords(5, 10);
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 40, 3, records2,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 40, 3, records2,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
sharePartition.acknowledge(MEMBER_ID, Collections.singletonList(
new ShareAcknowledgementBatch(10, 14, Collections.singletonList((byte) 2))));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 40, 3, records2,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 40, 3, records2,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -2430,15 +2592,15 @@ public class SharePartitionTest {
// third fetch request with 5 records starting from offset20.
MemoryRecords records3 = memoryRecords(5, 20);
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 40, 3, records1,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 40, 3, records1,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 30, 3, records2,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 3, records2,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 50, 3, records3,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 50, 3, records3,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -2449,16 +2611,16 @@ public class SharePartitionTest {
)));
// Send next batch from offset 13, only 2 records should be acquired.
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 40, 3, records1,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 40, 3, records1,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
// Send next batch from offset 15, only 2 records should be acquired.
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 40, 3, records2,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 40, 3, records2,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 40, 3, records3,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 40, 3, records3,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -2503,15 +2665,15 @@ public class SharePartitionTest {
// Third fetch request with 5 records starting from offset 20.
MemoryRecords records3 = memoryRecords(5, 20);
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 40, 3, records1,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 40, 3, records1,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 30, 3, records2,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 3, records2,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 50, 3, records3,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 50, 3, records3,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -2523,16 +2685,16 @@ public class SharePartitionTest {
)));
// Send next batch from offset 13, only 2 records should be acquired.
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 40, 3, records1,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 40, 3, records1,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
// Send next batch from offset 15, only 2 records should be acquired.
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 40, 3, records2,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 40, 3, records2,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 40, 3, records3,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 40, 3, records3,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -2548,7 +2710,7 @@ public class SharePartitionTest {
public void testReleaseAcquiredRecordsSubsetWithAnotherMember() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 30, 0, memoryRecords(7, 5),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -2585,7 +2747,7 @@ public class SharePartitionTest {
PartitionFactory.newPartitionErrorData(0, Errors.GROUP_ID_NOT_FOUND.code(), Errors.GROUP_ID_NOT_FOUND.message())))));
Mockito.when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(10, 5),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(10, 5),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -2612,7 +2774,7 @@ public class SharePartitionTest {
PartitionFactory.newPartitionErrorData(0, Errors.NONE.code(), Errors.NONE.message())))));
Mockito.when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(6, 5),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(6, 5),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -2650,7 +2812,7 @@ public class SharePartitionTest {
public void testAcquisitionLockOnReleasingMultipleRecordBatch() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(10, 5),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(10, 5),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -2682,11 +2844,11 @@ public class SharePartitionTest {
recordsBuilder.appendWithOffset(20, 0L, TestUtils.randomString(10).getBytes(), TestUtils.randomString(10).getBytes());
MemoryRecords records2 = recordsBuilder.build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 30, 0, records1,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 0, records1,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 30, 0, records2,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 0, records2,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -2763,25 +2925,25 @@ public class SharePartitionTest {
public void testLsoMovementForArchivingBatches() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 7),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 7),
Optional.empty(),
OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 12), Optional.empty(),
OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 17), Optional.empty(),
OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 22), Optional.empty(),
OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 27), Optional.empty(),
OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS,
new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 32), Optional.empty(),
OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
@@ -2831,10 +2993,10 @@ public class SharePartitionTest {
public void testLsoMovementForArchivingOffsets() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 7),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 7),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -2883,10 +3045,10 @@ public class SharePartitionTest {
public void testLsoMovementForArchivingOffsetsWithStartAndEndBatchesNotFullMatches() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 7),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 7),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -2923,10 +3085,10 @@ public class SharePartitionTest {
public void testLsoMovementForArchivingOffsetsWithStartOffsetNotFullMatches() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 7),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 7),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -2963,10 +3125,10 @@ public class SharePartitionTest {
public void testLsoMovementForArchivingOffsetsWithStartOffsetNotFullMatchesPostAcceptAcknowledgement() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 7),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 7),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -3013,9 +3175,9 @@ public class SharePartitionTest {
public void testLsoMovementForArchivingOffsetsWithStartOffsetNotFullMatchesPostReleaseAcknowledgement() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2), Optional.empty(),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2), Optional.empty(),
OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 7), Optional.empty(),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 7), Optional.empty(),
OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
// LSO is at 4.
@@ -3061,9 +3223,9 @@ public class SharePartitionTest {
public void testLsoMovementToEndOffset() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2), Optional.empty(),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2), Optional.empty(),
OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 7), Optional.empty(),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 7), Optional.empty(),
OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
// Acknowledge with RELEASE action.
@@ -3095,9 +3257,9 @@ public class SharePartitionTest {
public void testLsoMovementToEndOffsetWhereEndOffsetIsAvailable() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2), Optional.empty(),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2), Optional.empty(),
OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 7), Optional.empty(),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 7), Optional.empty(),
OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
// Acknowledge with RELEASE action.
@@ -3130,9 +3292,9 @@ public class SharePartitionTest {
public void testLsoMovementAheadOfEndOffsetPostAcknowledgment() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2), Optional.empty(),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2), Optional.empty(),
OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 7), Optional.empty(),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 7), Optional.empty(),
OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
// Acknowledge with RELEASE action.
@@ -3164,9 +3326,9 @@ public class SharePartitionTest {
public void testLsoMovementAheadOfEndOffset() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2), Optional.empty(),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2), Optional.empty(),
OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 7), Optional.empty(),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 7), Optional.empty(),
OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
// LSO is at 14.
@@ -3194,11 +3356,11 @@ public class SharePartitionTest {
// Gap of 15-19.
MemoryRecords records3 = memoryRecords(5, 20);
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, records1, Optional.empty(),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, records1, Optional.empty(),
OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, records2, Optional.empty(),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, records2, Optional.empty(),
OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, records3, Optional.empty(),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, records3, Optional.empty(),
OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
// LSO is at 18.
@@ -3227,9 +3389,9 @@ public class SharePartitionTest {
// Gap of 7-9.
MemoryRecords records2 = memoryRecords(5, 10);
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, records1, Optional.empty(),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, records1, Optional.empty(),
OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, records2, Optional.empty(),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, records2, Optional.empty(),
OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
// Acknowledge with RELEASE action.
@@ -3262,10 +3424,10 @@ public class SharePartitionTest {
recordsBuilder.appendWithOffset(18, 0L, TestUtils.randomString(10).getBytes(), TestUtils.randomString(10).getBytes());
MemoryRecords records2 = recordsBuilder.build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 30, 0, records1, Optional.empty(),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 0, records1, Optional.empty(),
OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 30, 0, records2, Optional.empty(),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 0, records2, Optional.empty(),
OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
sharePartition.acknowledge(MEMBER_ID, Arrays.asList(
@@ -3303,25 +3465,25 @@ public class SharePartitionTest {
public void testReleaseAcquiredRecordsBatchesPostStartOffsetMovement() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 5),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 5),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 10),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 10),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire("member-2", new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 15),
+ sharePartition.acquire("member-2", MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 15),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 20),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 20),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 25),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 25),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 30),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 30),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 35),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 35),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -3390,10 +3552,10 @@ public class SharePartitionTest {
public void testReleaseAcquiredRecordsBatchesPostStartOffsetMovementToStartOfBatch() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 5),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 5),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 10),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 10),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -3421,10 +3583,10 @@ public class SharePartitionTest {
public void testReleaseAcquiredRecordsBatchesPostStartOffsetMovementToMiddleOfBatch() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 5),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 5),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 10),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 10),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -3459,25 +3621,25 @@ public class SharePartitionTest {
SharePartition sharePartition = SharePartitionBuilder.builder()
.withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS).build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 5),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 5),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 10),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 10),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire("member-2", new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 15),
+ sharePartition.acquire("member-2", MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 15),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 20),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 20),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 25),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 25),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 30),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 30),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 35),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 35),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -3542,10 +3704,10 @@ public class SharePartitionTest {
SharePartition sharePartition = SharePartitionBuilder.builder()
.withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS).build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 5),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 5),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 10),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 10),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -3572,10 +3734,10 @@ public class SharePartitionTest {
SharePartition sharePartition = SharePartitionBuilder.builder()
.withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS).build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 5),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 5),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 10),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 10),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -3656,10 +3818,10 @@ public class SharePartitionTest {
public void testAcknowledgeBatchAndOffsetPostLsoMovement() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 10),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 10),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -3715,13 +3877,13 @@ public class SharePartitionTest {
public void testAcknowledgeBatchPostLsoMovement() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 10),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 10),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 20),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 20),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -3775,7 +3937,7 @@ public class SharePartitionTest {
SharePartition sharePartition = SharePartitionBuilder.builder()
.withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS).build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 2),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -3798,7 +3960,7 @@ public class SharePartitionTest {
DEFAULT_MAX_WAIT_ACQUISITION_LOCK_TIMEOUT_MS,
() -> ACQUISITION_LOCK_NEVER_GOT_RELEASED);
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 10),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 10),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -3825,7 +3987,7 @@ public class SharePartitionTest {
public void testLsoMovementThenAcquisitionLockTimeoutThenAcknowledgeBatchLastOffsetAheadOfStartOffsetBatch() throws InterruptedException {
SharePartition sharePartition = SharePartitionBuilder.builder().withDefaultAcquisitionLockTimeoutMs(ACQUISITION_LOCK_TIMEOUT_MS).build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(2, 1),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(2, 1),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -3848,11 +4010,11 @@ public class SharePartitionTest {
DEFAULT_MAX_WAIT_ACQUISITION_LOCK_TIMEOUT_MS,
() -> ACQUISITION_LOCK_NEVER_GOT_RELEASED);
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(2, 3),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(2, 3),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(3, 5),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(3, 5),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4097,7 +4259,7 @@ public class SharePartitionTest {
public void testMaybeUpdateCachedStateWhenAcknowledgementTypeAccept() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(250, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(250, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4119,7 +4281,7 @@ public class SharePartitionTest {
public void testMaybeUpdateCachedStateWhenAcknowledgementTypeReject() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(250, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(250, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4141,7 +4303,7 @@ public class SharePartitionTest {
public void testMaybeUpdateCachedStateWhenAcknowledgementTypeRelease() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(250, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(250, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
assertFalse(sharePartition.canAcquireRecords());
@@ -4164,12 +4326,12 @@ public class SharePartitionTest {
public void testMaybeUpdateCachedStateWhenAcknowledgementsFromBeginningForBatchSubset() {
SharePartition sharePartition = SharePartitionBuilder.builder().withMaxInflightMessages(20).build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(15, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(15, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
assertTrue(sharePartition.canAcquireRecords());
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(15, 15),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(15, 15),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
assertFalse(sharePartition.canAcquireRecords());
@@ -4190,12 +4352,12 @@ public class SharePartitionTest {
@Test
public void testMaybeUpdateCachedStateWhenAcknowledgementsFromBeginningForEntireBatch() {
SharePartition sharePartition = SharePartitionBuilder.builder().withMaxInflightMessages(20).build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(15, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(15, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
assertTrue(sharePartition.canAcquireRecords());
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(15, 15),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(15, 15),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
assertFalse(sharePartition.canAcquireRecords());
@@ -4217,12 +4379,12 @@ public class SharePartitionTest {
public void testMaybeUpdateCachedStateWhenAcknowledgementsInBetween() {
SharePartition sharePartition = SharePartitionBuilder.builder().withMaxInflightMessages(20).build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(15, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(15, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
assertTrue(sharePartition.canAcquireRecords());
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(15, 15),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(15, 15),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
assertFalse(sharePartition.canAcquireRecords());
@@ -4247,12 +4409,12 @@ public class SharePartitionTest {
public void testMaybeUpdateCachedStateWhenAllRecordsInCachedStateAreAcknowledged() {
SharePartition sharePartition = SharePartitionBuilder.builder().withMaxInflightMessages(20).build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(15, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(15, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
assertTrue(sharePartition.canAcquireRecords());
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(15, 15),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(15, 15),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
assertFalse(sharePartition.canAcquireRecords());
@@ -4270,17 +4432,17 @@ public class SharePartitionTest {
public void testMaybeUpdateCachedStateMultipleAcquisitionsAndAcknowledgements() {
SharePartition sharePartition = SharePartitionBuilder.builder().withMaxInflightMessages(100).build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(20, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(20, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
assertTrue(sharePartition.canAcquireRecords());
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(20, 20),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(20, 20),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
assertTrue(sharePartition.canAcquireRecords());
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(20, 40),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(20, 40),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
assertTrue(sharePartition.canAcquireRecords());
@@ -4294,7 +4456,7 @@ public class SharePartitionTest {
assertEquals(59, sharePartition.endOffset());
assertEquals(60, sharePartition.nextFetchOffset());
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(20, 60),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(20, 60),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
assertTrue(sharePartition.canAcquireRecords());
@@ -4310,7 +4472,7 @@ public class SharePartitionTest {
assertEquals(79, sharePartition.endOffset());
assertEquals(80, sharePartition.nextFetchOffset());
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(100, 80),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(100, 80),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
assertFalse(sharePartition.canAcquireRecords());
@@ -4325,7 +4487,7 @@ public class SharePartitionTest {
assertEquals(180, sharePartition.endOffset());
assertEquals(180, sharePartition.nextFetchOffset());
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(20, 180),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(20, 180),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4344,7 +4506,7 @@ public class SharePartitionTest {
assertEquals(0, sharePartition.startOffset());
assertEquals(0, sharePartition.endOffset());
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(150, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(150, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4360,13 +4522,13 @@ public class SharePartitionTest {
assertEquals(0, sharePartition.startOffset());
assertEquals(0, sharePartition.endOffset());
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(150, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(150, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
assertTrue(sharePartition.canAcquireRecords());
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(100, 150),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(100, 150),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4386,7 +4548,7 @@ public class SharePartitionTest {
public void testCanAcquireRecordsAfterReleaseAcknowledgement() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(150, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(150, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4394,7 +4556,7 @@ public class SharePartitionTest {
assertEquals(0, sharePartition.startOffset());
assertEquals(149, sharePartition.endOffset());
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(100, 150),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(100, 150),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4416,7 +4578,7 @@ public class SharePartitionTest {
public void testCanAcquireRecordsAfterArchiveAcknowledgement() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(150, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(150, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4424,7 +4586,7 @@ public class SharePartitionTest {
assertEquals(0, sharePartition.startOffset());
assertEquals(149, sharePartition.endOffset());
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(100, 150),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(100, 150),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4445,7 +4607,7 @@ public class SharePartitionTest {
public void testCanAcquireRecordsAfterAcceptAcknowledgement() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(150, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(150, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4453,7 +4615,7 @@ public class SharePartitionTest {
assertEquals(0, sharePartition.startOffset());
assertEquals(149, sharePartition.endOffset());
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(100, 150),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(100, 150),
Optional.empty(), OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), false));
assertFalse(sharePartition.canAcquireRecords());
@@ -4482,7 +4644,7 @@ public class SharePartitionTest {
PartitionFactory.newPartitionErrorData(0, Errors.UNKNOWN_TOPIC_OR_PARTITION.code(), Errors.UNKNOWN_TOPIC_OR_PARTITION.message())))));
Mockito.when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(10, 5),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(10, 5),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4510,7 +4672,7 @@ public class SharePartitionTest {
PartitionFactory.newPartitionErrorData(0, Errors.GROUP_ID_NOT_FOUND.code(), Errors.GROUP_ID_NOT_FOUND.message())))));
Mockito.when(persister.writeState(Mockito.any())).thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(6, 5),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(6, 5),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4540,7 +4702,7 @@ public class SharePartitionTest {
public void testAcknowledgeSubsetWithAnotherMember() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 30, 0, memoryRecords(7, 5),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 0, memoryRecords(7, 5),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4558,15 +4720,15 @@ public class SharePartitionTest {
public void testAcknowledgeWithAnotherMemberRollbackBatchError() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 5),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 5),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire("member-2", new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 10),
+ sharePartition.acquire("member-2", MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 10),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 15),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 15),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4595,15 +4757,15 @@ public class SharePartitionTest {
public void testAcknowledgeWithAnotherMemberRollbackSubsetError() {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 5),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 5),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 10),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 10),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire("member-2", new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 15),
+ sharePartition.acquire("member-2", MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(5, 15),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4633,14 +4795,14 @@ public class SharePartitionTest {
SharePartition sharePartition = SharePartitionBuilder.builder().withMaxDeliveryCount(2).build();
MemoryRecords records = memoryRecords(10, 5);
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, records,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, records,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
sharePartition.acknowledge(MEMBER_ID, Collections.singletonList(
new ShareAcknowledgementBatch(5, 14, Collections.singletonList((byte) 2))));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 20, 0, records,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, records,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4662,11 +4824,11 @@ public class SharePartitionTest {
// Second fetch request with 5 records starting from offset 15.
MemoryRecords records2 = memoryRecords(5, 15);
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 40, 3, records1,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 40, 3, records1,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 30, 3, records2,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 3, records2,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4676,12 +4838,12 @@ public class SharePartitionTest {
new ShareAcknowledgementBatch(17, 19, Collections.singletonList((byte) 1)))));
// Send next batch from offset 13, only 2 records should be acquired.
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 40, 3, records1,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 40, 3, records1,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
// Send next batch from offset 15, only 2 records should be acquired.
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 40, 3, records2,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 40, 3, records2,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4701,7 +4863,7 @@ public class SharePartitionTest {
// First fetch request with 5 records starting from offset 0.
MemoryRecords records1 = memoryRecords(5, 0);
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 40, 3, records1,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 40, 3, records1,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4709,7 +4871,7 @@ public class SharePartitionTest {
new ShareAcknowledgementBatch(0, 1, Collections.singletonList((byte) 2)))));
// Send next batch from offset 0, only 2 records should be acquired.
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 40, 3, memoryRecords(2, 0),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 40, 3, memoryRecords(2, 0),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4735,14 +4897,14 @@ public class SharePartitionTest {
String memberId1 = "memberId-1";
String memberId2 = "memberId-2";
- sharePartition.acquire(memberId1, new FetchPartitionData(Errors.NONE, 20, 0, records1,
+ sharePartition.acquire(memberId1, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, records1,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
assertFalse(sharePartition.findNextFetchOffset());
assertEquals(10, sharePartition.nextFetchOffset());
- sharePartition.acquire(memberId2, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(10, 10),
+ sharePartition.acquire(memberId2, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(10, 10),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4755,7 +4917,7 @@ public class SharePartitionTest {
assertTrue(sharePartition.findNextFetchOffset());
assertEquals(5, sharePartition.nextFetchOffset());
- sharePartition.acquire(memberId1, new FetchPartitionData(Errors.NONE, 20, 0, records1,
+ sharePartition.acquire(memberId1, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, records1,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4771,7 +4933,7 @@ public class SharePartitionTest {
String memberId1 = MEMBER_ID;
String memberId2 = "member-2";
- sharePartition.acquire(memberId1, new FetchPartitionData(Errors.NONE, 20, 0, records1,
+ sharePartition.acquire(memberId1, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, records1,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
assertEquals(3, sharePartition.nextFetchOffset());
@@ -4780,12 +4942,12 @@ public class SharePartitionTest {
new ShareAcknowledgementBatch(0, 2, Collections.singletonList((byte) 2))));
assertEquals(0, sharePartition.nextFetchOffset());
- sharePartition.acquire(memberId2, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(2, 3),
+ sharePartition.acquire(memberId2, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, memoryRecords(2, 3),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
assertEquals(0, sharePartition.nextFetchOffset());
- sharePartition.acquire(memberId1, new FetchPartitionData(Errors.NONE, 20, 0, records1,
+ sharePartition.acquire(memberId1, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 20, 0, records1,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
assertEquals(5, sharePartition.nextFetchOffset());
@@ -4799,7 +4961,7 @@ public class SharePartitionTest {
public void testNumberOfWriteCallsOnUpdates() {
SharePartition sharePartition = Mockito.spy(SharePartitionBuilder.builder().build());
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 10, 0, memoryRecords(5, 2),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 10, 0, memoryRecords(5, 2),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4819,11 +4981,11 @@ public class SharePartitionTest {
SharePartition sharePartition = SharePartitionBuilder.builder().build();
MemoryRecords records1 = memoryRecords(5, 5);
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 30, 0, records1,
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 0, records1,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
- sharePartition.acquire(MEMBER_ID, new FetchPartitionData(Errors.NONE, 30, 0, memoryRecords(12, 10),
+ sharePartition.acquire(MEMBER_ID, MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 0, memoryRecords(12, 10),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
@@ -4834,13 +4996,13 @@ public class SharePartitionTest {
new ShareAcknowledgementBatch(17, 20, Collections.singletonList((byte) 2))));
// Reacquire with another member.
- sharePartition.acquire("member-2", new FetchPartitionData(Errors.NONE, 30, 0, records1,
+ sharePartition.acquire("member-2", MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 0, records1,
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
assertEquals(10, sharePartition.nextFetchOffset());
// Reacquire with another member.
- sharePartition.acquire("member-2", new FetchPartitionData(Errors.NONE, 30, 0, memoryRecords(7, 10),
+ sharePartition.acquire("member-2", MAX_FETCH_RECORDS, new FetchPartitionData(Errors.NONE, 30, 0, memoryRecords(7, 10),
Optional.empty(), OptionalLong.empty(), Optional.empty(),
OptionalInt.empty(), false));
assertEquals(17, sharePartition.nextFetchOffset());
@@ -4897,6 +5059,12 @@ public class SharePartitionTest {
assertEquals(734, sharePartition.endOffset());
}
+ private List<AcquiredRecords> fetchAcquiredRecords(ShareAcquiredRecords shareAcquiredRecords, int expectedOffsetCount) {
+ assertNotNull(shareAcquiredRecords);
+ assertEquals(expectedOffsetCount, shareAcquiredRecords.count());
+ return shareAcquiredRecords.acquiredRecords();
+ }
+
private MemoryRecords memoryRecords(int numOfRecords) {
return memoryRecords(numOfRecords, 0);
}
@@ -4906,14 +5074,25 @@ public class SharePartitionTest {
}
private MemoryRecordsBuilder memoryRecordsBuilder(int numOfRecords, long startOffset) {
- MemoryRecordsBuilder builder = MemoryRecords.builder(ByteBuffer.allocate(1024),
- Compression.NONE, TimestampType.CREATE_TIME, startOffset);
+ return memoryRecordsBuilder(ByteBuffer.allocate(1024), numOfRecords, startOffset);
+ }
+
+ private MemoryRecordsBuilder memoryRecordsBuilder(ByteBuffer buffer, int numOfRecords, long startOffset) {
+ MemoryRecordsBuilder builder = MemoryRecords.builder(buffer, Compression.NONE,
+ TimestampType.CREATE_TIME, startOffset, 2);
for (int i = 0; i < numOfRecords; i++) {
builder.appendWithOffset(startOffset + i, 0L, TestUtils.randomString(10).getBytes(), TestUtils.randomString(10).getBytes());
}
return builder;
}
+ private List<AcquiredRecords> expectedAcquiredRecord(long baseOffset, long lastOffset, int deliveryCount) {
+ return Collections.singletonList(new AcquiredRecords()
+ .setFirstOffset(baseOffset)
+ .setLastOffset(lastOffset)
+ .setDeliveryCount((short) deliveryCount));
+ }
+
private List<AcquiredRecords> expectedAcquiredRecords(MemoryRecords memoryRecords, int deliveryCount) {
List<AcquiredRecords> acquiredRecordsList = new ArrayList<>();
memoryRecords.batches().forEach(batch -> acquiredRecordsList.add(new AcquiredRecords()
diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
index ed1a0e6c432..076e2e62472 100755
--- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
@@ -1144,6 +1144,7 @@ class KafkaConfigTest {
case ShareGroupConfig.SHARE_GROUP_MAX_GROUPS_CONFIG => assertPropertyInvalid(baseProperties, name, "not_a_number", 0, -1)
case GroupCoordinatorConfig.SHARE_GROUP_MAX_SIZE_CONFIG => assertPropertyInvalid(baseProperties, name, "not_a_number", 0, -1)
case ShareGroupConfig.SHARE_FETCH_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG => assertPropertyInvalid(baseProperties, name, "not_a_number")
+ case ShareGroupConfig.SHARE_FETCH_MAX_FETCH_RECORDS_CONFIG => assertPropertyInvalid(baseProperties, name, "not_a_number")
case _ => assertPropertyInvalid(baseProperties, name, "not_a_number", "-1")
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroupConfig.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroupConfig.java
index 483dbbb0332..15dbca6b9fa 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroupConfig.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroupConfig.java
@@ -29,6 +29,7 @@ import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
+import static org.apache.kafka.common.config.ConfigDef.Importance.HIGH;
import static org.apache.kafka.common.config.ConfigDef.Importance.MEDIUM;
import static org.apache.kafka.common.config.ConfigDef.Range.between;
import static org.apache.kafka.common.config.ConfigDef.Type.BOOLEAN;
@@ -71,15 +72,21 @@ public class ShareGroupConfig {
public static final int SHARE_FETCH_PURGATORY_PURGE_INTERVAL_REQUESTS_DEFAULT = 1000;
public static final String SHARE_FETCH_PURGATORY_PURGE_INTERVAL_REQUESTS_DOC = "The purge interval (in number of requests) of the share fetch request purgatory";
+ // Broker temporary configuration to limit the number of records fetched by a share fetch request.
+ public static final String SHARE_FETCH_MAX_FETCH_RECORDS_CONFIG = "share.fetch.max.fetch.records";
+ public static final int SHARE_FETCH_MAX_FETCH_RECORDS_DEFAULT = Integer.MAX_VALUE;
+ public static final String SHARE_FETCH_MAX_FETCH_RECORDS_DOC = "The maximum number of records that can be fetched by a share fetch request.";
+
public static final ConfigDef CONFIG_DEF = new ConfigDef()
.defineInternal(SHARE_GROUP_ENABLE_CONFIG, BOOLEAN, SHARE_GROUP_ENABLE_DEFAULT, null, MEDIUM, SHARE_GROUP_ENABLE_DOC)
+ .defineInternal(SHARE_FETCH_MAX_FETCH_RECORDS_CONFIG, INT, SHARE_FETCH_MAX_FETCH_RECORDS_DEFAULT, null, HIGH, SHARE_FETCH_MAX_FETCH_RECORDS_DOC)
.define(SHARE_GROUP_DELIVERY_COUNT_LIMIT_CONFIG, INT, SHARE_GROUP_DELIVERY_COUNT_LIMIT_DEFAULT, between(2, 10), MEDIUM, SHARE_GROUP_DELIVERY_COUNT_LIMIT_DOC)
.define(SHARE_GROUP_RECORD_LOCK_DURATION_MS_CONFIG, INT, SHARE_GROUP_RECORD_LOCK_DURATION_MS_DEFAULT, between(1000, 3600000), MEDIUM, SHARE_GROUP_RECORD_LOCK_DURATION_MS_DOC)
.define(SHARE_GROUP_MIN_RECORD_LOCK_DURATION_MS_CONFIG, INT, SHARE_GROUP_MIN_RECORD_LOCK_DURATION_MS_DEFAULT, between(1000, 30000), MEDIUM, SHARE_GROUP_MIN_RECORD_LOCK_DURATION_MS_DOC)
.define(SHARE_GROUP_MAX_RECORD_LOCK_DURATION_MS_CONFIG, INT, SHARE_GROUP_MAX_RECORD_LOCK_DURATION_MS_DEFAULT, between(30000, 3600000), MEDIUM, SHARE_GROUP_MAX_RECORD_LOCK_DURATION_MS_DOC)
.define(SHARE_GROUP_MAX_GROUPS_CONFIG, SHORT, SHARE_GROUP_MAX_GROUPS_DEFAULT, between(1, 100), MEDIUM, SHARE_GROUP_MAX_GROUPS_DOC)
.define(SHARE_GROUP_PARTITION_MAX_RECORD_LOCKS_CONFIG, INT, SHARE_GROUP_PARTITION_MAX_RECORD_LOCKS_DEFAULT, between(100, 10000), MEDIUM, SHARE_GROUP_PARTITION_MAX_RECORD_LOCKS_DOC)
- .define(SHARE_FETCH_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG, INT, SHARE_FETCH_PURGATORY_PURGE_INTERVAL_REQUESTS_DEFAULT, MEDIUM, SHARE_FETCH_PURGATORY_PURGE_INTERVAL_REQUESTS_DOC);;
+ .define(SHARE_FETCH_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG, INT, SHARE_FETCH_PURGATORY_PURGE_INTERVAL_REQUESTS_DEFAULT, MEDIUM, SHARE_FETCH_PURGATORY_PURGE_INTERVAL_REQUESTS_DOC);
private final boolean isShareGroupEnabled;
private final int shareGroupPartitionMaxRecordLocks;
@@ -89,6 +96,7 @@ public class ShareGroupConfig {
private final int shareGroupMaxRecordLockDurationMs;
private final int shareGroupMinRecordLockDurationMs;
private final int shareFetchPurgatoryPurgeIntervalRequests;
+ private final int shareFetchMaxFetchRecords;
public ShareGroupConfig(AbstractConfig config) {
// Share groups are enabled in two cases: 1) The internal configuration to enable it is
@@ -104,6 +112,7 @@ public class ShareGroupConfig {
shareGroupMaxRecordLockDurationMs = config.getInt(ShareGroupConfig.SHARE_GROUP_MAX_RECORD_LOCK_DURATION_MS_CONFIG);
shareGroupMinRecordLockDurationMs = config.getInt(ShareGroupConfig.SHARE_GROUP_MIN_RECORD_LOCK_DURATION_MS_CONFIG);
shareFetchPurgatoryPurgeIntervalRequests = config.getInt(ShareGroupConfig.SHARE_FETCH_PURGATORY_PURGE_INTERVAL_REQUESTS_CONFIG);
+ shareFetchMaxFetchRecords = config.getInt(ShareGroupConfig.SHARE_FETCH_MAX_FETCH_RECORDS_CONFIG);
validate();
}
@@ -140,6 +149,10 @@ public class ShareGroupConfig {
return shareFetchPurgatoryPurgeIntervalRequests;
}
+ public int shareFetchMaxFetchRecords() {
+ return shareFetchMaxFetchRecords;
+ }
+
private void validate() {
Utils.require(shareGroupRecordLockDurationMs >= shareGroupMinRecordLockDurationMs,
String.format("%s must be greater than or equal to %s",
diff --git a/share/src/main/java/org/apache/kafka/server/share/fetch/ShareAcquiredRecords.java b/share/src/main/java/org/apache/kafka/server/share/fetch/ShareAcquiredRecords.java
new file mode 100644
index 00000000000..3e97e5b11b3
--- /dev/null
+++ b/share/src/main/java/org/apache/kafka/server/share/fetch/ShareAcquiredRecords.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.server.share.fetch;
+
+import org.apache.kafka.common.message.ShareFetchResponseData.AcquiredRecords;
+
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * The ShareAcquiredRecords class is used to send the acquired records and associated metadata.
+ */
+public class ShareAcquiredRecords {
+
+ private static final ShareAcquiredRecords EMPTY_SHARE_ACQUIRED_RECORDS = new ShareAcquiredRecords();
+
+ /**
+ * The list of acquired records.
+ */
+ private final List<AcquiredRecords> acquiredRecords;
+ /**
+ * The number of offsets acquired. The acquired records has a first and last offset, and the count
+ * is the actual number of offsets acquired.
+ */
+ private final int count;
+
+ public ShareAcquiredRecords(
+ List<AcquiredRecords> acquiredRecords,
+ int count
+ ) {
+ this.acquiredRecords = acquiredRecords;
+ this.count = count;
+ }
+
+ private ShareAcquiredRecords() {
+ this.acquiredRecords = Collections.emptyList();
+ this.count = 0;
+ }
+
+ public List<AcquiredRecords> acquiredRecords() {
+ return acquiredRecords;
+ }
+
+ public int count() {
+ return count;
+ }
+
+ public static ShareAcquiredRecords empty() {
+ return EMPTY_SHARE_ACQUIRED_RECORDS;
+ }
+
+ public static ShareAcquiredRecords fromAcquiredRecords(AcquiredRecords acquiredRecords) {
+ return new ShareAcquiredRecords(
+ List.of(acquiredRecords), (int) (acquiredRecords.lastOffset() - acquiredRecords.firstOffset() + 1)
+ );
+ }
+}
diff --git a/share/src/main/java/org/apache/kafka/server/share/fetch/ShareFetchData.java b/share/src/main/java/org/apache/kafka/server/share/fetch/ShareFetchData.java
index 121d46f3200..c32b3280017 100644
--- a/share/src/main/java/org/apache/kafka/server/share/fetch/ShareFetchData.java
+++ b/share/src/main/java/org/apache/kafka/server/share/fetch/ShareFetchData.java
@@ -34,19 +34,22 @@ public class ShareFetchData {
private final String memberId;
private final CompletableFuture<Map<TopicIdPartition, PartitionData>> future;
private final Map<TopicIdPartition, Integer> partitionMaxBytes;
+ private final int maxFetchRecords;
public ShareFetchData(
FetchParams fetchParams,
String groupId,
String memberId,
CompletableFuture<Map<TopicIdPartition, PartitionData>> future,
- Map<TopicIdPartition, Integer> partitionMaxBytes
+ Map<TopicIdPartition, Integer> partitionMaxBytes,
+ int maxFetchRecords
) {
this.fetchParams = fetchParams;
this.groupId = groupId;
this.memberId = memberId;
this.future = future;
this.partitionMaxBytes = partitionMaxBytes;
+ this.maxFetchRecords = maxFetchRecords;
}
public String groupId() {
@@ -68,4 +71,8 @@ public class ShareFetchData {
public FetchParams fetchParams() {
return fetchParams;
}
+
+ public int maxFetchRecords() {
+ return maxFetchRecords;
+ }
}