You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ij...@apache.org on 2025/01/17 15:06:05 UTC
(kafka) branch 4.0 updated (3a71d593b1b -> 146eff7af0d)
This is an automated email from the ASF dual-hosted git repository.
ijuma pushed a change to branch 4.0
in repository https://gitbox.apache.org/repos/asf/kafka.git
from 3a71d593b1b KAFKA-17642: PreVote response handling and ProspectiveState (#18240)
new b021b515952 KAFKA-18404: Remove partitionMaxBytes usage from DelayedShareFetch (#17870)
new a461601ee26 MINOR: Fixing unchecked warning in partition strategy test (#18541)
new 146eff7af0d KAFKA-18514: Refactor share module code to server and server-common (#18524)
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
build.gradle | 32 --
checkstyle/import-control-server-common.xml | 9 +
checkstyle/import-control-share.xml | 62 ----
.../java/kafka/server/share/DelayedShareFetch.java | 108 ++++---
.../kafka/server/share/DelayedShareFetchTest.java | 357 ++++++++++++++++++++-
.../server/share/SharePartitionManagerTest.java | 3 +
.../java/kafka/test/api/ShareConsumerTest.java | 2 +-
.../server/ShareFetchAcknowledgeRequestTest.scala | 14 +-
.../kafka/server/share/SharePartitionKey.java | 0
.../share/persister/DefaultStatePersister.java | 0
.../persister/DeleteShareGroupStateParameters.java | 0
.../persister/DeleteShareGroupStateResult.java | 0
.../share/persister/GroupTopicPartitionData.java | 0
.../InitializeShareGroupStateParameters.java | 0
.../persister/InitializeShareGroupStateResult.java | 0
.../share/persister/NoOpShareStatePersister.java | 0
.../server/share/persister/PartitionAllData.java | 0
.../server/share/persister/PartitionData.java | 0
.../server/share/persister/PartitionErrorData.java | 0
.../server/share/persister/PartitionFactory.java | 0
.../server/share/persister/PartitionIdData.java | 0
.../persister/PartitionIdLeaderEpochData.java | 0
.../server/share/persister/PartitionInfoData.java | 0
.../share/persister/PartitionStateBatchData.java | 0
.../server/share/persister/PartitionStateData.java | 0
.../share/persister/PartitionStateErrorData.java | 0
.../kafka/server/share/persister/Persister.java | 0
.../share/persister/PersisterParameters.java | 0
.../server/share/persister/PersisterResult.java | 0
.../share/persister/PersisterStateBatch.java | 0
.../share/persister/PersisterStateManager.java | 0
.../persister/ReadShareGroupStateParameters.java | 0
.../share/persister/ReadShareGroupStateResult.java | 0
.../ReadShareGroupStateSummaryParameters.java | 0
.../ReadShareGroupStateSummaryResult.java | 0
.../ShareCoordinatorMetadataCacheHelper.java | 0
.../kafka/server/share/persister/TopicData.java | 0
.../persister/WriteShareGroupStateParameters.java | 0
.../persister/WriteShareGroupStateResult.java | 0
.../share/persister/DefaultStatePersisterTest.java | 0
.../share/persister/PersisterStateManagerTest.java | 0
.../kafka/server/share/CachedSharePartition.java | 0
.../share/ErroneousAndValidPartitionData.java | 0
.../acknowledge/ShareAcknowledgementBatch.java | 0
.../kafka/server/share/context/FinalContext.java | 0
.../server/share/context/ShareFetchContext.java | 0
.../server/share/context/ShareSessionContext.java | 0
.../share/fetch/DelayedShareFetchGroupKey.java | 0
.../server/share/fetch/DelayedShareFetchKey.java | 0
.../share/fetch/DelayedShareFetchPartitionKey.java | 0
.../share/fetch/PartitionMaxBytesStrategy.java | 77 +++++
.../server/share/fetch/ShareAcquiredRecords.java | 0
.../kafka/server/share/fetch/ShareFetch.java | 0
.../kafka/server/share/session/LastUsedKey.java | 0
.../kafka/server/share/session/ShareSession.java | 0
.../server/share/session/ShareSessionCache.java | 0
.../server/share/session/ShareSessionKey.java | 0
.../server/share/CachedSharePartitionTest.java | 0
.../share/fetch/DelayedShareFetchKeyTest.java | 0
.../share/fetch/PartitionMaxBytesStrategyTest.java | 89 +++++
.../kafka/server/share/fetch/ShareFetchTest.java | 0
.../share/session/ShareSessionCacheTest.java | 0
.../server/share/session/ShareSessionTest.java | 0
63 files changed, 604 insertions(+), 149 deletions(-)
delete mode 100644 checkstyle/import-control-share.xml
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/SharePartitionKey.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/DefaultStatePersister.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/DeleteShareGroupStateParameters.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/DeleteShareGroupStateResult.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/GroupTopicPartitionData.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/InitializeShareGroupStateParameters.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/InitializeShareGroupStateResult.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/NoOpShareStatePersister.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/PartitionAllData.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/PartitionData.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/PartitionErrorData.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/PartitionFactory.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/PartitionIdData.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/PartitionIdLeaderEpochData.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/PartitionInfoData.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/PartitionStateBatchData.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/PartitionStateData.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/PartitionStateErrorData.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/Persister.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/PersisterParameters.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/PersisterResult.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/PersisterStateBatch.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/ReadShareGroupStateParameters.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/ReadShareGroupStateResult.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/ReadShareGroupStateSummaryParameters.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/ReadShareGroupStateSummaryResult.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/ShareCoordinatorMetadataCacheHelper.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/TopicData.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/WriteShareGroupStateParameters.java (100%)
rename {share => server-common}/src/main/java/org/apache/kafka/server/share/persister/WriteShareGroupStateResult.java (100%)
rename {share => server-common}/src/test/java/org/apache/kafka/server/share/persister/DefaultStatePersisterTest.java (100%)
rename {share => server-common}/src/test/java/org/apache/kafka/server/share/persister/PersisterStateManagerTest.java (100%)
rename {share => server}/src/main/java/org/apache/kafka/server/share/CachedSharePartition.java (100%)
rename {share => server}/src/main/java/org/apache/kafka/server/share/ErroneousAndValidPartitionData.java (100%)
rename {share => server}/src/main/java/org/apache/kafka/server/share/acknowledge/ShareAcknowledgementBatch.java (100%)
rename {share => server}/src/main/java/org/apache/kafka/server/share/context/FinalContext.java (100%)
rename {share => server}/src/main/java/org/apache/kafka/server/share/context/ShareFetchContext.java (100%)
rename {share => server}/src/main/java/org/apache/kafka/server/share/context/ShareSessionContext.java (100%)
rename {share => server}/src/main/java/org/apache/kafka/server/share/fetch/DelayedShareFetchGroupKey.java (100%)
rename {share => server}/src/main/java/org/apache/kafka/server/share/fetch/DelayedShareFetchKey.java (100%)
rename {share => server}/src/main/java/org/apache/kafka/server/share/fetch/DelayedShareFetchPartitionKey.java (100%)
create mode 100644 server/src/main/java/org/apache/kafka/server/share/fetch/PartitionMaxBytesStrategy.java
rename {share => server}/src/main/java/org/apache/kafka/server/share/fetch/ShareAcquiredRecords.java (100%)
rename {share => server}/src/main/java/org/apache/kafka/server/share/fetch/ShareFetch.java (100%)
rename {share => server}/src/main/java/org/apache/kafka/server/share/session/LastUsedKey.java (100%)
rename {share => server}/src/main/java/org/apache/kafka/server/share/session/ShareSession.java (100%)
rename {share => server}/src/main/java/org/apache/kafka/server/share/session/ShareSessionCache.java (100%)
rename {share => server}/src/main/java/org/apache/kafka/server/share/session/ShareSessionKey.java (100%)
rename {share => server}/src/test/java/org/apache/kafka/server/share/CachedSharePartitionTest.java (100%)
rename {share => server}/src/test/java/org/apache/kafka/server/share/fetch/DelayedShareFetchKeyTest.java (100%)
create mode 100644 server/src/test/java/org/apache/kafka/server/share/fetch/PartitionMaxBytesStrategyTest.java
rename {share => server}/src/test/java/org/apache/kafka/server/share/fetch/ShareFetchTest.java (100%)
rename {share => server}/src/test/java/org/apache/kafka/server/share/session/ShareSessionCacheTest.java (100%)
rename {share => server}/src/test/java/org/apache/kafka/server/share/session/ShareSessionTest.java (100%)
(kafka) 02/03: MINOR: Fixing unchecked warning in partition strategy test (#18541)
Posted by ij...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ijuma pushed a commit to branch 4.0
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit a461601ee269f51b752e377c2a4f638cac920600
Author: Apoorv Mittal <ap...@gmail.com>
AuthorDate: Wed Jan 15 20:33:57 2025 +0000
MINOR: Fixing unchecked warning in partition strategy test (#18541)
Reviewers: Viktor Somogyi-Vass <vi...@gmail.com>, Chia-Ping Tsai <ch...@gmail.com>
---
.../apache/kafka/server/share/fetch/PartitionMaxBytesStrategyTest.java | 3 +--
1 file changed, 1 insertion(+), 2 deletions(-)
diff --git a/share/src/test/java/org/apache/kafka/server/share/fetch/PartitionMaxBytesStrategyTest.java b/share/src/test/java/org/apache/kafka/server/share/fetch/PartitionMaxBytesStrategyTest.java
index 3c6c7ad2207..07338673832 100644
--- a/share/src/test/java/org/apache/kafka/server/share/fetch/PartitionMaxBytesStrategyTest.java
+++ b/share/src/test/java/org/apache/kafka/server/share/fetch/PartitionMaxBytesStrategyTest.java
@@ -23,7 +23,6 @@ import org.apache.kafka.server.share.fetch.PartitionMaxBytesStrategy.StrategyTyp
import org.junit.jupiter.api.Test;
-import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
@@ -56,7 +55,7 @@ public class PartitionMaxBytesStrategyTest {
100, partitions, 0));
// empty partitions set.
assertThrows(IllegalArgumentException.class, () -> PartitionMaxBytesStrategy.checkValidArguments(
- 100, Collections.EMPTY_SET, 20));
+ 100, Set.of(), 20));
// partitions is null.
assertThrows(IllegalArgumentException.class, () -> PartitionMaxBytesStrategy.checkValidArguments(
100, null, 20));
(kafka) 01/03: KAFKA-18404: Remove partitionMaxBytes usage from DelayedShareFetch (#17870)
Posted by ij...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ijuma pushed a commit to branch 4.0
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit b021b515952153a8876b48dda134ef3d81781c13
Author: Abhinav Dixit <ad...@confluent.io>
AuthorDate: Mon Jan 13 19:26:19 2025 +0530
KAFKA-18404: Remove partitionMaxBytes usage from DelayedShareFetch (#17870)
Reviewers: Apoorv Mittal <ap...@gmail.com>, Andrew Schofield <as...@confluent.io>
---
.../java/kafka/server/share/DelayedShareFetch.java | 108 ++++---
.../kafka/server/share/DelayedShareFetchTest.java | 357 ++++++++++++++++++++-
.../server/share/SharePartitionManagerTest.java | 3 +
.../java/kafka/test/api/ShareConsumerTest.java | 2 +-
.../server/ShareFetchAcknowledgeRequestTest.scala | 14 +-
.../share/fetch/PartitionMaxBytesStrategy.java | 77 +++++
.../share/fetch/PartitionMaxBytesStrategyTest.java | 90 ++++++
7 files changed, 596 insertions(+), 55 deletions(-)
diff --git a/core/src/main/java/kafka/server/share/DelayedShareFetch.java b/core/src/main/java/kafka/server/share/DelayedShareFetch.java
index 9bab9818c07..1422e08524a 100644
--- a/core/src/main/java/kafka/server/share/DelayedShareFetch.java
+++ b/core/src/main/java/kafka/server/share/DelayedShareFetch.java
@@ -27,6 +27,7 @@ import org.apache.kafka.common.requests.FetchRequest;
import org.apache.kafka.server.purgatory.DelayedOperation;
import org.apache.kafka.server.share.SharePartitionKey;
import org.apache.kafka.server.share.fetch.DelayedShareFetchGroupKey;
+import org.apache.kafka.server.share.fetch.PartitionMaxBytesStrategy;
import org.apache.kafka.server.share.fetch.ShareFetch;
import org.apache.kafka.server.storage.log.FetchIsolation;
import org.apache.kafka.server.storage.log.FetchPartitionData;
@@ -60,10 +61,11 @@ public class DelayedShareFetch extends DelayedOperation {
private final ShareFetch shareFetch;
private final ReplicaManager replicaManager;
private final BiConsumer<SharePartitionKey, Throwable> exceptionHandler;
+ private final PartitionMaxBytesStrategy partitionMaxBytesStrategy;
// The topic partitions that need to be completed for the share fetch request are given by sharePartitions.
// sharePartitions is a subset of shareFetchData. The order of insertion/deletion of entries in sharePartitions is important.
private final LinkedHashMap<TopicIdPartition, SharePartition> sharePartitions;
- private LinkedHashMap<TopicIdPartition, FetchRequest.PartitionData> partitionsAcquired;
+ private LinkedHashMap<TopicIdPartition, Long> partitionsAcquired;
private LinkedHashMap<TopicIdPartition, LogReadResult> partitionsAlreadyFetched;
DelayedShareFetch(
@@ -71,6 +73,15 @@ public class DelayedShareFetch extends DelayedOperation {
ReplicaManager replicaManager,
BiConsumer<SharePartitionKey, Throwable> exceptionHandler,
LinkedHashMap<TopicIdPartition, SharePartition> sharePartitions) {
+ this(shareFetch, replicaManager, exceptionHandler, sharePartitions, PartitionMaxBytesStrategy.type(PartitionMaxBytesStrategy.StrategyType.UNIFORM));
+ }
+
+ DelayedShareFetch(
+ ShareFetch shareFetch,
+ ReplicaManager replicaManager,
+ BiConsumer<SharePartitionKey, Throwable> exceptionHandler,
+ LinkedHashMap<TopicIdPartition, SharePartition> sharePartitions,
+ PartitionMaxBytesStrategy partitionMaxBytesStrategy) {
super(shareFetch.fetchParams().maxWaitMs, Optional.empty());
this.shareFetch = shareFetch;
this.replicaManager = replicaManager;
@@ -78,6 +89,7 @@ public class DelayedShareFetch extends DelayedOperation {
this.partitionsAlreadyFetched = new LinkedHashMap<>();
this.exceptionHandler = exceptionHandler;
this.sharePartitions = sharePartitions;
+ this.partitionMaxBytesStrategy = partitionMaxBytesStrategy;
}
@Override
@@ -99,7 +111,7 @@ public class DelayedShareFetch extends DelayedOperation {
partitionsAcquired.keySet());
try {
- LinkedHashMap<TopicIdPartition, FetchRequest.PartitionData> topicPartitionData;
+ LinkedHashMap<TopicIdPartition, Long> topicPartitionData;
// tryComplete did not invoke forceComplete, so we need to check if we have any partitions to fetch.
if (partitionsAcquired.isEmpty())
topicPartitionData = acquirablePartitions();
@@ -121,11 +133,13 @@ public class DelayedShareFetch extends DelayedOperation {
}
}
- private void completeShareFetchRequest(LinkedHashMap<TopicIdPartition, FetchRequest.PartitionData> topicPartitionData) {
+ private void completeShareFetchRequest(LinkedHashMap<TopicIdPartition, Long> topicPartitionData) {
try {
LinkedHashMap<TopicIdPartition, LogReadResult> responseData;
if (partitionsAlreadyFetched.isEmpty())
- responseData = readFromLog(topicPartitionData);
+ responseData = readFromLog(
+ topicPartitionData,
+ partitionMaxBytesStrategy.maxBytes(shareFetch.fetchParams().maxBytes, topicPartitionData.keySet(), topicPartitionData.size()));
else
// There shouldn't be a case when we have a partitionsAlreadyFetched value here and this variable is getting
// updated in a different tryComplete thread.
@@ -158,7 +172,7 @@ public class DelayedShareFetch extends DelayedOperation {
*/
@Override
public boolean tryComplete() {
- LinkedHashMap<TopicIdPartition, FetchRequest.PartitionData> topicPartitionData = acquirablePartitions();
+ LinkedHashMap<TopicIdPartition, Long> topicPartitionData = acquirablePartitions();
try {
if (!topicPartitionData.isEmpty()) {
@@ -167,7 +181,7 @@ public class DelayedShareFetch extends DelayedOperation {
// those topic partitions.
LinkedHashMap<TopicIdPartition, LogReadResult> replicaManagerReadResponse = maybeReadFromLog(topicPartitionData);
maybeUpdateFetchOffsetMetadata(topicPartitionData, replicaManagerReadResponse);
- if (anyPartitionHasLogReadError(replicaManagerReadResponse) || isMinBytesSatisfied(topicPartitionData)) {
+ if (anyPartitionHasLogReadError(replicaManagerReadResponse) || isMinBytesSatisfied(topicPartitionData, partitionMaxBytesStrategy.maxBytes(shareFetch.fetchParams().maxBytes, topicPartitionData.keySet(), topicPartitionData.size()))) {
partitionsAcquired = topicPartitionData;
partitionsAlreadyFetched = replicaManagerReadResponse;
boolean completedByMe = forceComplete();
@@ -202,28 +216,18 @@ public class DelayedShareFetch extends DelayedOperation {
* Prepare fetch request structure for partitions in the share fetch request for which we can acquire records.
*/
// Visible for testing
- LinkedHashMap<TopicIdPartition, FetchRequest.PartitionData> acquirablePartitions() {
+ LinkedHashMap<TopicIdPartition, Long> acquirablePartitions() {
// Initialize the topic partitions for which the fetch should be attempted.
- LinkedHashMap<TopicIdPartition, FetchRequest.PartitionData> topicPartitionData = new LinkedHashMap<>();
+ LinkedHashMap<TopicIdPartition, Long> topicPartitionData = new LinkedHashMap<>();
sharePartitions.forEach((topicIdPartition, sharePartition) -> {
- int partitionMaxBytes = shareFetch.partitionMaxBytes().getOrDefault(topicIdPartition, 0);
// Add the share partition to the list of partitions to be fetched only if we can
// acquire the fetch lock on it.
if (sharePartition.maybeAcquireFetchLock()) {
try {
// If the share partition is already at capacity, we should not attempt to fetch.
if (sharePartition.canAcquireRecords()) {
- topicPartitionData.put(
- topicIdPartition,
- new FetchRequest.PartitionData(
- topicIdPartition.topicId(),
- sharePartition.nextFetchOffset(),
- 0,
- partitionMaxBytes,
- Optional.empty()
- )
- );
+ topicPartitionData.put(topicIdPartition, sharePartition.nextFetchOffset());
} else {
sharePartition.releaseFetchLock();
log.trace("Record lock partition limit exceeded for SharePartition {}, " +
@@ -239,24 +243,28 @@ public class DelayedShareFetch extends DelayedOperation {
return topicPartitionData;
}
- private LinkedHashMap<TopicIdPartition, LogReadResult> maybeReadFromLog(LinkedHashMap<TopicIdPartition, FetchRequest.PartitionData> topicPartitionData) {
- LinkedHashMap<TopicIdPartition, FetchRequest.PartitionData> partitionsNotMatchingFetchOffsetMetadata = new LinkedHashMap<>();
- topicPartitionData.forEach((topicIdPartition, partitionData) -> {
+ private LinkedHashMap<TopicIdPartition, LogReadResult> maybeReadFromLog(LinkedHashMap<TopicIdPartition, Long> topicPartitionData) {
+ LinkedHashMap<TopicIdPartition, Long> partitionsNotMatchingFetchOffsetMetadata = new LinkedHashMap<>();
+ topicPartitionData.forEach((topicIdPartition, fetchOffset) -> {
SharePartition sharePartition = sharePartitions.get(topicIdPartition);
- if (sharePartition.fetchOffsetMetadata(partitionData.fetchOffset).isEmpty()) {
- partitionsNotMatchingFetchOffsetMetadata.put(topicIdPartition, partitionData);
+ if (sharePartition.fetchOffsetMetadata(fetchOffset).isEmpty()) {
+ partitionsNotMatchingFetchOffsetMetadata.put(topicIdPartition, fetchOffset);
}
});
if (partitionsNotMatchingFetchOffsetMetadata.isEmpty()) {
return new LinkedHashMap<>();
}
// We fetch data from replica manager corresponding to the topic partitions that have missing fetch offset metadata.
- return readFromLog(partitionsNotMatchingFetchOffsetMetadata);
+ // Although we are fetching partition max bytes for partitionsNotMatchingFetchOffsetMetadata,
+ // we will take acquired partitions size = topicPartitionData.size() because we do not want to let the
+ // leftover partitions to starve which will be fetched later.
+ return readFromLog(
+ partitionsNotMatchingFetchOffsetMetadata,
+ partitionMaxBytesStrategy.maxBytes(shareFetch.fetchParams().maxBytes, partitionsNotMatchingFetchOffsetMetadata.keySet(), topicPartitionData.size()));
}
- private void maybeUpdateFetchOffsetMetadata(
- LinkedHashMap<TopicIdPartition, FetchRequest.PartitionData> topicPartitionData,
- LinkedHashMap<TopicIdPartition, LogReadResult> replicaManagerReadResponseData) {
+ private void maybeUpdateFetchOffsetMetadata(LinkedHashMap<TopicIdPartition, Long> topicPartitionData,
+ LinkedHashMap<TopicIdPartition, LogReadResult> replicaManagerReadResponseData) {
for (Map.Entry<TopicIdPartition, LogReadResult> entry : replicaManagerReadResponseData.entrySet()) {
TopicIdPartition topicIdPartition = entry.getKey();
SharePartition sharePartition = sharePartitions.get(topicIdPartition);
@@ -267,17 +275,18 @@ public class DelayedShareFetch extends DelayedOperation {
continue;
}
sharePartition.updateFetchOffsetMetadata(
- topicPartitionData.get(topicIdPartition).fetchOffset,
+ topicPartitionData.get(topicIdPartition),
replicaManagerLogReadResult.info().fetchOffsetMetadata);
}
}
// minByes estimation currently assumes the common case where all fetched data is acquirable.
- private boolean isMinBytesSatisfied(LinkedHashMap<TopicIdPartition, FetchRequest.PartitionData> topicPartitionData) {
+ private boolean isMinBytesSatisfied(LinkedHashMap<TopicIdPartition, Long> topicPartitionData,
+ LinkedHashMap<TopicIdPartition, Integer> partitionMaxBytes) {
long accumulatedSize = 0;
- for (Map.Entry<TopicIdPartition, FetchRequest.PartitionData> entry : topicPartitionData.entrySet()) {
+ for (Map.Entry<TopicIdPartition, Long> entry : topicPartitionData.entrySet()) {
TopicIdPartition topicIdPartition = entry.getKey();
- FetchRequest.PartitionData partitionData = entry.getValue();
+ long fetchOffset = entry.getValue();
LogOffsetMetadata endOffsetMetadata;
try {
@@ -294,7 +303,7 @@ public class DelayedShareFetch extends DelayedOperation {
SharePartition sharePartition = sharePartitions.get(topicIdPartition);
- Optional<LogOffsetMetadata> optionalFetchOffsetMetadata = sharePartition.fetchOffsetMetadata(partitionData.fetchOffset);
+ Optional<LogOffsetMetadata> optionalFetchOffsetMetadata = sharePartition.fetchOffsetMetadata(fetchOffset);
if (optionalFetchOffsetMetadata.isEmpty() || optionalFetchOffsetMetadata.get() == LogOffsetMetadata.UNKNOWN_OFFSET_METADATA)
continue;
LogOffsetMetadata fetchOffsetMetadata = optionalFetchOffsetMetadata.get();
@@ -312,7 +321,7 @@ public class DelayedShareFetch extends DelayedOperation {
return true;
} else if (fetchOffsetMetadata.onSameSegment(endOffsetMetadata)) {
// we take the partition fetch size as upper bound when accumulating the bytes.
- long bytesAvailable = Math.min(endOffsetMetadata.positionDiff(fetchOffsetMetadata), partitionData.maxBytes);
+ long bytesAvailable = Math.min(endOffsetMetadata.positionDiff(fetchOffsetMetadata), partitionMaxBytes.get(topicIdPartition));
accumulatedSize += bytesAvailable;
}
}
@@ -335,13 +344,25 @@ public class DelayedShareFetch extends DelayedOperation {
}
- private LinkedHashMap<TopicIdPartition, LogReadResult> readFromLog(LinkedHashMap<TopicIdPartition, FetchRequest.PartitionData> topicPartitionData) {
+ private LinkedHashMap<TopicIdPartition, LogReadResult> readFromLog(LinkedHashMap<TopicIdPartition, Long> topicPartitionFetchOffsets,
+ LinkedHashMap<TopicIdPartition, Integer> partitionMaxBytes) {
// Filter if there already exists any erroneous topic partition.
- Set<TopicIdPartition> partitionsToFetch = shareFetch.filterErroneousTopicPartitions(topicPartitionData.keySet());
+ Set<TopicIdPartition> partitionsToFetch = shareFetch.filterErroneousTopicPartitions(topicPartitionFetchOffsets.keySet());
if (partitionsToFetch.isEmpty()) {
return new LinkedHashMap<>();
}
+ LinkedHashMap<TopicIdPartition, FetchRequest.PartitionData> topicPartitionData = new LinkedHashMap<>();
+
+ topicPartitionFetchOffsets.forEach((topicIdPartition, fetchOffset) -> topicPartitionData.put(topicIdPartition,
+ new FetchRequest.PartitionData(
+ topicIdPartition.topicId(),
+ fetchOffset,
+ 0,
+ partitionMaxBytes.get(topicIdPartition),
+ Optional.empty())
+ ));
+
Seq<Tuple2<TopicIdPartition, LogReadResult>> responseLogResult = replicaManager.readFromLog(
shareFetch.fetchParams(),
CollectionConverters.asScala(
@@ -390,18 +411,21 @@ public class DelayedShareFetch extends DelayedOperation {
}
// Visible for testing.
- LinkedHashMap<TopicIdPartition, LogReadResult> combineLogReadResponse(LinkedHashMap<TopicIdPartition, FetchRequest.PartitionData> topicPartitionData,
- LinkedHashMap<TopicIdPartition, LogReadResult> existingFetchedData) {
- LinkedHashMap<TopicIdPartition, FetchRequest.PartitionData> missingLogReadTopicPartitions = new LinkedHashMap<>();
- topicPartitionData.forEach((topicIdPartition, partitionData) -> {
+ LinkedHashMap<TopicIdPartition, LogReadResult> combineLogReadResponse(LinkedHashMap<TopicIdPartition, Long> topicPartitionData,
+ LinkedHashMap<TopicIdPartition, LogReadResult> existingFetchedData) {
+ LinkedHashMap<TopicIdPartition, Long> missingLogReadTopicPartitions = new LinkedHashMap<>();
+ topicPartitionData.forEach((topicIdPartition, fetchOffset) -> {
if (!existingFetchedData.containsKey(topicIdPartition)) {
- missingLogReadTopicPartitions.put(topicIdPartition, partitionData);
+ missingLogReadTopicPartitions.put(topicIdPartition, fetchOffset);
}
});
if (missingLogReadTopicPartitions.isEmpty()) {
return existingFetchedData;
}
- LinkedHashMap<TopicIdPartition, LogReadResult> missingTopicPartitionsLogReadResponse = readFromLog(missingLogReadTopicPartitions);
+
+ LinkedHashMap<TopicIdPartition, LogReadResult> missingTopicPartitionsLogReadResponse = readFromLog(
+ missingLogReadTopicPartitions,
+ partitionMaxBytesStrategy.maxBytes(shareFetch.fetchParams().maxBytes, missingLogReadTopicPartitions.keySet(), topicPartitionData.size()));
missingTopicPartitionsLogReadResponse.putAll(existingFetchedData);
return missingTopicPartitionsLogReadResponse;
}
diff --git a/core/src/test/java/kafka/server/share/DelayedShareFetchTest.java b/core/src/test/java/kafka/server/share/DelayedShareFetchTest.java
index 11d3e26eaf3..533ffc95693 100644
--- a/core/src/test/java/kafka/server/share/DelayedShareFetchTest.java
+++ b/core/src/test/java/kafka/server/share/DelayedShareFetchTest.java
@@ -18,6 +18,7 @@ package kafka.server.share;
import kafka.cluster.Partition;
import kafka.server.LogReadResult;
+import kafka.server.QuotaFactory;
import kafka.server.ReplicaManager;
import kafka.server.ReplicaQuota;
@@ -26,11 +27,13 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.message.ShareFetchResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.record.Records;
import org.apache.kafka.common.requests.FetchRequest;
import org.apache.kafka.server.purgatory.DelayedOperationKey;
import org.apache.kafka.server.purgatory.DelayedOperationPurgatory;
import org.apache.kafka.server.share.SharePartitionKey;
import org.apache.kafka.server.share.fetch.DelayedShareFetchGroupKey;
+import org.apache.kafka.server.share.fetch.PartitionMaxBytesStrategy;
import org.apache.kafka.server.share.fetch.ShareAcquiredRecords;
import org.apache.kafka.server.share.fetch.ShareFetch;
import org.apache.kafka.server.storage.log.FetchIsolation;
@@ -39,6 +42,7 @@ 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;
+import org.apache.kafka.storage.internals.log.FetchDataInfo;
import org.apache.kafka.storage.internals.log.LogOffsetMetadata;
import org.apache.kafka.storage.internals.log.LogOffsetSnapshot;
@@ -51,11 +55,17 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.function.BiConsumer;
+import java.util.stream.Collectors;
+
+import scala.Tuple2;
+import scala.jdk.javaapi.CollectionConverters;
import static kafka.server.share.SharePartitionManagerTest.DELAYED_SHARE_FETCH_PURGATORY_PURGE_INTERVAL;
import static kafka.server.share.SharePartitionManagerTest.PARTITION_MAX_BYTES;
@@ -182,6 +192,7 @@ public class DelayedShareFetchTest {
.withSharePartitions(sharePartitions)
.withReplicaManager(replicaManager)
.withExceptionHandler(exceptionHandler)
+ .withPartitionMaxBytesStrategy(PartitionMaxBytesStrategy.type(PartitionMaxBytesStrategy.StrategyType.UNIFORM))
.build());
assertFalse(delayedShareFetch.isCompleted());
@@ -286,6 +297,7 @@ public class DelayedShareFetchTest {
.withShareFetchData(shareFetch)
.withSharePartitions(sharePartitions)
.withReplicaManager(replicaManager)
+ .withPartitionMaxBytesStrategy(PartitionMaxBytesStrategy.type(PartitionMaxBytesStrategy.StrategyType.UNIFORM))
.build());
assertFalse(delayedShareFetch.isCompleted());
@@ -328,6 +340,7 @@ public class DelayedShareFetchTest {
.withShareFetchData(shareFetch)
.withReplicaManager(replicaManager)
.withSharePartitions(sharePartitions)
+ .withPartitionMaxBytesStrategy(PartitionMaxBytesStrategy.type(PartitionMaxBytesStrategy.StrategyType.UNIFORM))
.build());
assertFalse(delayedShareFetch.isCompleted());
delayedShareFetch.forceComplete();
@@ -375,6 +388,7 @@ public class DelayedShareFetchTest {
.withShareFetchData(shareFetch)
.withReplicaManager(replicaManager)
.withSharePartitions(sharePartitions)
+ .withPartitionMaxBytesStrategy(PartitionMaxBytesStrategy.type(PartitionMaxBytesStrategy.StrategyType.UNIFORM))
.build());
assertFalse(delayedShareFetch.isCompleted());
delayedShareFetch.forceComplete();
@@ -507,6 +521,7 @@ public class DelayedShareFetchTest {
.withShareFetchData(shareFetch2)
.withReplicaManager(replicaManager)
.withSharePartitions(sharePartitions2)
+ .withPartitionMaxBytesStrategy(PartitionMaxBytesStrategy.type(PartitionMaxBytesStrategy.StrategyType.UNIFORM))
.build());
// sp1 can be acquired now
@@ -557,15 +572,21 @@ public class DelayedShareFetchTest {
.withShareFetchData(shareFetch)
.withReplicaManager(replicaManager)
.withSharePartitions(sharePartitions)
+ .withPartitionMaxBytesStrategy(PartitionMaxBytesStrategy.type(PartitionMaxBytesStrategy.StrategyType.UNIFORM))
.build();
- LinkedHashMap<TopicIdPartition, FetchRequest.PartitionData> topicPartitionData = new LinkedHashMap<>();
- topicPartitionData.put(tp0, mock(FetchRequest.PartitionData.class));
- topicPartitionData.put(tp1, mock(FetchRequest.PartitionData.class));
+ LinkedHashMap<TopicIdPartition, Long> topicPartitionData = new LinkedHashMap<>();
+ topicPartitionData.put(tp0, 0L);
+ topicPartitionData.put(tp1, 0L);
// Case 1 - logReadResponse contains tp0.
LinkedHashMap<TopicIdPartition, LogReadResult> logReadResponse = new LinkedHashMap<>();
- logReadResponse.put(tp0, mock(LogReadResult.class));
+ LogReadResult logReadResult = mock(LogReadResult.class);
+ Records records = mock(Records.class);
+ when(records.sizeInBytes()).thenReturn(2);
+ FetchDataInfo fetchDataInfo = new FetchDataInfo(mock(LogOffsetMetadata.class), records);
+ when(logReadResult.info()).thenReturn(fetchDataInfo);
+ logReadResponse.put(tp0, logReadResult);
doAnswer(invocation -> buildLogReadResult(Collections.singleton(tp1))).when(replicaManager).readFromLog(any(), any(), any(ReplicaQuota.class), anyBoolean());
LinkedHashMap<TopicIdPartition, LogReadResult> combinedLogReadResponse = delayedShareFetch.combineLogReadResponse(topicPartitionData, logReadResponse);
@@ -619,6 +640,7 @@ public class DelayedShareFetchTest {
.withSharePartitions(sharePartitions)
.withReplicaManager(replicaManager)
.withExceptionHandler(exceptionHandler)
+ .withPartitionMaxBytesStrategy(PartitionMaxBytesStrategy.type(PartitionMaxBytesStrategy.StrategyType.UNIFORM))
.build());
// Try complete should return false as the share partition has errored out.
@@ -707,6 +729,324 @@ public class DelayedShareFetchTest {
delayedShareFetch.lock().unlock();
}
+ @Test
+ public void testTryCompleteWhenPartitionMaxBytesStrategyThrowsException() {
+ String groupId = "grp";
+ TopicIdPartition tp0 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 0));
+ SharePartition sp0 = mock(SharePartition.class);
+ Map<TopicIdPartition, Integer> partitionMaxBytes = new HashMap<>();
+ partitionMaxBytes.put(tp0, PARTITION_MAX_BYTES);
+
+ when(sp0.maybeAcquireFetchLock()).thenReturn(true);
+ when(sp0.canAcquireRecords()).thenReturn(true);
+ LinkedHashMap<TopicIdPartition, SharePartition> sharePartitions = new LinkedHashMap<>();
+ sharePartitions.put(tp0, sp0);
+ CompletableFuture<Map<TopicIdPartition, ShareFetchResponseData.PartitionData>> future = new CompletableFuture<>();
+
+ ShareFetch shareFetch = new ShareFetch(
+ new FetchParams(ApiKeys.SHARE_FETCH.latestVersion(), FetchRequest.ORDINARY_CONSUMER_ID, -1, MAX_WAIT_MS,
+ 2, 1024 * 1024, FetchIsolation.HIGH_WATERMARK, Optional.empty()), groupId, Uuid.randomUuid().toString(),
+ future, partitionMaxBytes, MAX_FETCH_RECORDS);
+
+ // partitionMaxBytesStrategy.maxBytes() function throws an exception
+ PartitionMaxBytesStrategy partitionMaxBytesStrategy = mock(PartitionMaxBytesStrategy.class);
+ when(partitionMaxBytesStrategy.maxBytes(anyInt(), any(), anyInt())).thenThrow(new IllegalArgumentException("Exception thrown"));
+
+ DelayedShareFetch delayedShareFetch = spy(DelayedShareFetchBuilder.builder()
+ .withShareFetchData(shareFetch)
+ .withSharePartitions(sharePartitions)
+ .withExceptionHandler(mockExceptionHandler())
+ .withPartitionMaxBytesStrategy(partitionMaxBytesStrategy)
+ .build());
+
+ assertFalse(delayedShareFetch.isCompleted());
+ assertTrue(delayedShareFetch.tryComplete());
+ assertTrue(delayedShareFetch.isCompleted());
+ // releasePartitionLocks is called twice - first time from tryComplete and second time from onComplete
+ Mockito.verify(delayedShareFetch, times(2)).releasePartitionLocks(any());
+ assertTrue(delayedShareFetch.lock().tryLock());
+ delayedShareFetch.lock().unlock();
+
+ assertTrue(future.isDone());
+ assertFalse(future.isCompletedExceptionally());
+ Map<TopicIdPartition, ShareFetchResponseData.PartitionData> partitionDataMap = future.join();
+ assertEquals(1, partitionDataMap.size());
+ assertTrue(partitionDataMap.containsKey(tp0));
+ assertEquals("Exception thrown", partitionDataMap.get(tp0).errorMessage());
+ }
+
+ @Test
+ public void testPartitionMaxBytesFromUniformStrategyWhenAllPartitionsAreAcquirable() {
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+ String groupId = "grp";
+ TopicIdPartition tp0 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 0));
+ TopicIdPartition tp1 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 1));
+ TopicIdPartition tp2 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 2));
+ TopicIdPartition tp3 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 3));
+ TopicIdPartition tp4 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 4));
+
+ SharePartition sp0 = mock(SharePartition.class);
+ SharePartition sp1 = mock(SharePartition.class);
+ SharePartition sp2 = mock(SharePartition.class);
+ SharePartition sp3 = mock(SharePartition.class);
+ SharePartition sp4 = mock(SharePartition.class);
+
+ Map<TopicIdPartition, Integer> partitionMaxBytes = new HashMap<>();
+ partitionMaxBytes.put(tp0, PARTITION_MAX_BYTES);
+ partitionMaxBytes.put(tp1, PARTITION_MAX_BYTES);
+ partitionMaxBytes.put(tp2, PARTITION_MAX_BYTES);
+ partitionMaxBytes.put(tp3, PARTITION_MAX_BYTES);
+ partitionMaxBytes.put(tp4, PARTITION_MAX_BYTES);
+
+ when(sp0.maybeAcquireFetchLock()).thenReturn(true);
+ when(sp1.maybeAcquireFetchLock()).thenReturn(true);
+ when(sp2.maybeAcquireFetchLock()).thenReturn(true);
+ when(sp3.maybeAcquireFetchLock()).thenReturn(true);
+ when(sp4.maybeAcquireFetchLock()).thenReturn(true);
+ when(sp0.canAcquireRecords()).thenReturn(true);
+ when(sp1.canAcquireRecords()).thenReturn(true);
+ when(sp2.canAcquireRecords()).thenReturn(true);
+ when(sp3.canAcquireRecords()).thenReturn(true);
+ when(sp4.canAcquireRecords()).thenReturn(true);
+
+ LinkedHashMap<TopicIdPartition, SharePartition> sharePartitions = new LinkedHashMap<>();
+ sharePartitions.put(tp0, sp0);
+ sharePartitions.put(tp1, sp1);
+ sharePartitions.put(tp2, sp2);
+ sharePartitions.put(tp3, sp3);
+ sharePartitions.put(tp4, sp4);
+
+ ShareFetch shareFetch = new ShareFetch(FETCH_PARAMS, groupId, Uuid.randomUuid().toString(),
+ new CompletableFuture<>(), partitionMaxBytes, MAX_FETCH_RECORDS);
+
+ 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(0).setLastOffset(3).setDeliveryCount((short) 1)));
+ when(sp2.acquire(anyString(), anyInt(), any(FetchPartitionData.class))).thenReturn(
+ ShareAcquiredRecords.fromAcquiredRecords(new ShareFetchResponseData.AcquiredRecords().setFirstOffset(0).setLastOffset(3).setDeliveryCount((short) 1)));
+ when(sp3.acquire(anyString(), anyInt(), any(FetchPartitionData.class))).thenReturn(
+ ShareAcquiredRecords.fromAcquiredRecords(new ShareFetchResponseData.AcquiredRecords().setFirstOffset(0).setLastOffset(3).setDeliveryCount((short) 1)));
+ when(sp4.acquire(anyString(), anyInt(), any(FetchPartitionData.class))).thenReturn(
+ ShareAcquiredRecords.fromAcquiredRecords(new ShareFetchResponseData.AcquiredRecords().setFirstOffset(0).setLastOffset(3).setDeliveryCount((short) 1)));
+
+ // All 5 partitions are acquirable.
+ doAnswer(invocation -> buildLogReadResult(sharePartitions.keySet())).when(replicaManager).readFromLog(any(), any(), any(ReplicaQuota.class), anyBoolean());
+
+ when(sp0.fetchOffsetMetadata(anyLong())).thenReturn(Optional.of(new LogOffsetMetadata(0, 1, 0)));
+ when(sp1.fetchOffsetMetadata(anyLong())).thenReturn(Optional.of(new LogOffsetMetadata(0, 1, 0)));
+ when(sp2.fetchOffsetMetadata(anyLong())).thenReturn(Optional.of(new LogOffsetMetadata(0, 1, 0)));
+ when(sp3.fetchOffsetMetadata(anyLong())).thenReturn(Optional.of(new LogOffsetMetadata(0, 1, 0)));
+ when(sp4.fetchOffsetMetadata(anyLong())).thenReturn(Optional.of(new LogOffsetMetadata(0, 1, 0)));
+
+ mockTopicIdPartitionToReturnDataEqualToMinBytes(replicaManager, tp0, 1);
+ mockTopicIdPartitionToReturnDataEqualToMinBytes(replicaManager, tp1, 1);
+ mockTopicIdPartitionToReturnDataEqualToMinBytes(replicaManager, tp2, 1);
+ mockTopicIdPartitionToReturnDataEqualToMinBytes(replicaManager, tp3, 1);
+ mockTopicIdPartitionToReturnDataEqualToMinBytes(replicaManager, tp4, 1);
+
+ DelayedShareFetch delayedShareFetch = spy(DelayedShareFetchBuilder.builder()
+ .withShareFetchData(shareFetch)
+ .withSharePartitions(sharePartitions)
+ .withReplicaManager(replicaManager)
+ .withPartitionMaxBytesStrategy(PartitionMaxBytesStrategy.type(PartitionMaxBytesStrategy.StrategyType.UNIFORM))
+ .build());
+
+ assertTrue(delayedShareFetch.tryComplete());
+ assertTrue(delayedShareFetch.isCompleted());
+
+ // Since all partitions are acquirable, maxbytes per partition = requestMaxBytes(i.e. 1024*1024) / acquiredTopicPartitions(i.e. 5)
+ int expectedPartitionMaxBytes = 1024 * 1024 / 5;
+ LinkedHashMap<TopicIdPartition, FetchRequest.PartitionData> expectedReadPartitionInfo = new LinkedHashMap<>();
+ sharePartitions.keySet().forEach(topicIdPartition -> expectedReadPartitionInfo.put(topicIdPartition,
+ new FetchRequest.PartitionData(
+ topicIdPartition.topicId(),
+ 0,
+ 0,
+ expectedPartitionMaxBytes,
+ Optional.empty()
+ )));
+
+ Mockito.verify(replicaManager, times(1)).readFromLog(
+ shareFetch.fetchParams(),
+ CollectionConverters.asScala(
+ sharePartitions.keySet().stream().map(topicIdPartition ->
+ new Tuple2<>(topicIdPartition, expectedReadPartitionInfo.get(topicIdPartition))).collect(Collectors.toList())
+ ),
+ QuotaFactory.UNBOUNDED_QUOTA,
+ true);
+ }
+
+ @Test
+ public void testPartitionMaxBytesFromUniformStrategyWhenFewPartitionsAreAcquirable() {
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+ String groupId = "grp";
+ TopicIdPartition tp0 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 0));
+ TopicIdPartition tp1 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 1));
+ TopicIdPartition tp2 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 2));
+ TopicIdPartition tp3 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 3));
+ TopicIdPartition tp4 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 4));
+
+ SharePartition sp0 = mock(SharePartition.class);
+ SharePartition sp1 = mock(SharePartition.class);
+ SharePartition sp2 = mock(SharePartition.class);
+ SharePartition sp3 = mock(SharePartition.class);
+ SharePartition sp4 = mock(SharePartition.class);
+
+ Map<TopicIdPartition, Integer> partitionMaxBytes = new HashMap<>();
+ partitionMaxBytes.put(tp0, PARTITION_MAX_BYTES);
+ partitionMaxBytes.put(tp1, PARTITION_MAX_BYTES);
+ partitionMaxBytes.put(tp2, PARTITION_MAX_BYTES);
+ partitionMaxBytes.put(tp3, PARTITION_MAX_BYTES);
+ partitionMaxBytes.put(tp4, PARTITION_MAX_BYTES);
+
+ when(sp0.maybeAcquireFetchLock()).thenReturn(true);
+ when(sp1.maybeAcquireFetchLock()).thenReturn(true);
+ when(sp2.maybeAcquireFetchLock()).thenReturn(false);
+ when(sp3.maybeAcquireFetchLock()).thenReturn(true);
+ when(sp4.maybeAcquireFetchLock()).thenReturn(false);
+ when(sp0.canAcquireRecords()).thenReturn(true);
+ when(sp1.canAcquireRecords()).thenReturn(true);
+ when(sp2.canAcquireRecords()).thenReturn(false);
+ when(sp3.canAcquireRecords()).thenReturn(false);
+ when(sp4.canAcquireRecords()).thenReturn(false);
+
+ LinkedHashMap<TopicIdPartition, SharePartition> sharePartitions = new LinkedHashMap<>();
+ sharePartitions.put(tp0, sp0);
+ sharePartitions.put(tp1, sp1);
+ sharePartitions.put(tp2, sp2);
+ sharePartitions.put(tp3, sp3);
+ sharePartitions.put(tp4, sp4);
+
+ ShareFetch shareFetch = new ShareFetch(FETCH_PARAMS, groupId, Uuid.randomUuid().toString(),
+ new CompletableFuture<>(), partitionMaxBytes, MAX_FETCH_RECORDS);
+
+ 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(0).setLastOffset(3).setDeliveryCount((short) 1)));
+
+ // Only 2 out of 5 partitions are acquirable.
+ Set<TopicIdPartition> acquirableTopicPartitions = new LinkedHashSet<>();
+ acquirableTopicPartitions.add(tp0);
+ acquirableTopicPartitions.add(tp1);
+ doAnswer(invocation -> buildLogReadResult(acquirableTopicPartitions)).when(replicaManager).readFromLog(any(), any(), any(ReplicaQuota.class), anyBoolean());
+
+ when(sp0.fetchOffsetMetadata(anyLong())).thenReturn(Optional.of(new LogOffsetMetadata(0, 1, 0)));
+ when(sp1.fetchOffsetMetadata(anyLong())).thenReturn(Optional.of(new LogOffsetMetadata(0, 1, 0)));
+
+ mockTopicIdPartitionToReturnDataEqualToMinBytes(replicaManager, tp0, 1);
+ mockTopicIdPartitionToReturnDataEqualToMinBytes(replicaManager, tp1, 1);
+
+ DelayedShareFetch delayedShareFetch = spy(DelayedShareFetchBuilder.builder()
+ .withShareFetchData(shareFetch)
+ .withSharePartitions(sharePartitions)
+ .withReplicaManager(replicaManager)
+ .withPartitionMaxBytesStrategy(PartitionMaxBytesStrategy.type(PartitionMaxBytesStrategy.StrategyType.UNIFORM))
+ .build());
+
+ assertTrue(delayedShareFetch.tryComplete());
+ assertTrue(delayedShareFetch.isCompleted());
+
+ // Since only 2 partitions are acquirable, maxbytes per partition = requestMaxBytes(i.e. 1024*1024) / acquiredTopicPartitions(i.e. 2)
+ int expectedPartitionMaxBytes = 1024 * 1024 / 2;
+ LinkedHashMap<TopicIdPartition, FetchRequest.PartitionData> expectedReadPartitionInfo = new LinkedHashMap<>();
+ acquirableTopicPartitions.forEach(topicIdPartition -> expectedReadPartitionInfo.put(topicIdPartition,
+ new FetchRequest.PartitionData(
+ topicIdPartition.topicId(),
+ 0,
+ 0,
+ expectedPartitionMaxBytes,
+ Optional.empty()
+ )));
+
+ Mockito.verify(replicaManager, times(1)).readFromLog(
+ shareFetch.fetchParams(),
+ CollectionConverters.asScala(
+ acquirableTopicPartitions.stream().map(topicIdPartition ->
+ new Tuple2<>(topicIdPartition, expectedReadPartitionInfo.get(topicIdPartition))).collect(Collectors.toList())
+ ),
+ QuotaFactory.UNBOUNDED_QUOTA,
+ true);
+ }
+
+ @Test
+ public void testPartitionMaxBytesFromUniformStrategyInCombineLogReadResponse() {
+ String groupId = "grp";
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+ TopicIdPartition tp0 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 0));
+ TopicIdPartition tp1 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 1));
+ TopicIdPartition tp2 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 2));
+ Map<TopicIdPartition, Integer> partitionMaxBytes = new HashMap<>();
+ partitionMaxBytes.put(tp0, PARTITION_MAX_BYTES);
+ partitionMaxBytes.put(tp1, PARTITION_MAX_BYTES);
+ partitionMaxBytes.put(tp2, PARTITION_MAX_BYTES);
+
+ SharePartition sp0 = mock(SharePartition.class);
+ SharePartition sp1 = mock(SharePartition.class);
+ SharePartition sp2 = mock(SharePartition.class);
+
+ LinkedHashMap<TopicIdPartition, SharePartition> sharePartitions = new LinkedHashMap<>();
+ sharePartitions.put(tp0, sp0);
+ sharePartitions.put(tp1, sp1);
+ sharePartitions.put(tp2, sp2);
+
+ ShareFetch shareFetch = new ShareFetch(
+ 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, MAX_FETCH_RECORDS);
+
+ DelayedShareFetch delayedShareFetch = DelayedShareFetchBuilder.builder()
+ .withShareFetchData(shareFetch)
+ .withReplicaManager(replicaManager)
+ .withSharePartitions(sharePartitions)
+ .withPartitionMaxBytesStrategy(PartitionMaxBytesStrategy.type(PartitionMaxBytesStrategy.StrategyType.UNIFORM))
+ .build();
+
+ LinkedHashMap<TopicIdPartition, Long> topicPartitionData = new LinkedHashMap<>();
+ topicPartitionData.put(tp0, 0L);
+ topicPartitionData.put(tp1, 0L);
+ topicPartitionData.put(tp2, 0L);
+
+ // Existing fetched data already contains tp0.
+ LinkedHashMap<TopicIdPartition, LogReadResult> logReadResponse = new LinkedHashMap<>();
+ LogReadResult logReadResult = mock(LogReadResult.class);
+ Records records = mock(Records.class);
+ when(records.sizeInBytes()).thenReturn(2);
+ FetchDataInfo fetchDataInfo = new FetchDataInfo(mock(LogOffsetMetadata.class), records);
+ when(logReadResult.info()).thenReturn(fetchDataInfo);
+ logReadResponse.put(tp0, logReadResult);
+
+ Set<TopicIdPartition> fetchableTopicPartitions = new LinkedHashSet<>();
+ fetchableTopicPartitions.add(tp1);
+ fetchableTopicPartitions.add(tp2);
+ // We will be doing replica manager fetch only for tp1 and tp2.
+ doAnswer(invocation -> buildLogReadResult(fetchableTopicPartitions)).when(replicaManager).readFromLog(any(), any(), any(ReplicaQuota.class), anyBoolean());
+ LinkedHashMap<TopicIdPartition, LogReadResult> combinedLogReadResponse = delayedShareFetch.combineLogReadResponse(topicPartitionData, logReadResponse);
+
+ assertEquals(topicPartitionData.keySet(), combinedLogReadResponse.keySet());
+ // Since only 2 partitions are fetchable but the third one has already been fetched, maxbytes per partition = requestMaxBytes(i.e. 1024*1024) / acquiredTopicPartitions(i.e. 3)
+ int expectedPartitionMaxBytes = 1024 * 1024 / 3;
+ LinkedHashMap<TopicIdPartition, FetchRequest.PartitionData> expectedReadPartitionInfo = new LinkedHashMap<>();
+ fetchableTopicPartitions.forEach(topicIdPartition -> expectedReadPartitionInfo.put(topicIdPartition,
+ new FetchRequest.PartitionData(
+ topicIdPartition.topicId(),
+ 0,
+ 0,
+ expectedPartitionMaxBytes,
+ Optional.empty()
+ )));
+
+ Mockito.verify(replicaManager, times(1)).readFromLog(
+ shareFetch.fetchParams(),
+ CollectionConverters.asScala(
+ fetchableTopicPartitions.stream().map(topicIdPartition ->
+ new Tuple2<>(topicIdPartition, expectedReadPartitionInfo.get(topicIdPartition))).collect(Collectors.toList())
+ ),
+ QuotaFactory.UNBOUNDED_QUOTA,
+ true);
+ }
+
static void mockTopicIdPartitionToReturnDataEqualToMinBytes(ReplicaManager replicaManager, TopicIdPartition topicIdPartition, int minBytes) {
LogOffsetMetadata hwmOffsetMetadata = new LogOffsetMetadata(1, 1, minBytes);
LogOffsetSnapshot endOffsetSnapshot = new LogOffsetSnapshot(1, mock(LogOffsetMetadata.class),
@@ -736,6 +1076,7 @@ public class DelayedShareFetchTest {
private ReplicaManager replicaManager = mock(ReplicaManager.class);
private BiConsumer<SharePartitionKey, Throwable> exceptionHandler = mockExceptionHandler();
private LinkedHashMap<TopicIdPartition, SharePartition> sharePartitions = mock(LinkedHashMap.class);
+ private PartitionMaxBytesStrategy partitionMaxBytesStrategy = mock(PartitionMaxBytesStrategy.class);
DelayedShareFetchBuilder withShareFetchData(ShareFetch shareFetch) {
this.shareFetch = shareFetch;
@@ -757,6 +1098,11 @@ public class DelayedShareFetchTest {
return this;
}
+ DelayedShareFetchBuilder withPartitionMaxBytesStrategy(PartitionMaxBytesStrategy partitionMaxBytesStrategy) {
+ this.partitionMaxBytesStrategy = partitionMaxBytesStrategy;
+ return this;
+ }
+
public static DelayedShareFetchBuilder builder() {
return new DelayedShareFetchBuilder();
}
@@ -766,7 +1112,8 @@ public class DelayedShareFetchTest {
shareFetch,
replicaManager,
exceptionHandler,
- sharePartitions);
+ sharePartitions,
+ partitionMaxBytesStrategy);
}
}
}
diff --git a/core/src/test/java/kafka/server/share/SharePartitionManagerTest.java b/core/src/test/java/kafka/server/share/SharePartitionManagerTest.java
index a8b5941c16e..872862e94a4 100644
--- a/core/src/test/java/kafka/server/share/SharePartitionManagerTest.java
+++ b/core/src/test/java/kafka/server/share/SharePartitionManagerTest.java
@@ -64,6 +64,7 @@ import org.apache.kafka.server.share.context.ShareFetchContext;
import org.apache.kafka.server.share.context.ShareSessionContext;
import org.apache.kafka.server.share.fetch.DelayedShareFetchGroupKey;
import org.apache.kafka.server.share.fetch.DelayedShareFetchKey;
+import org.apache.kafka.server.share.fetch.PartitionMaxBytesStrategy;
import org.apache.kafka.server.share.fetch.ShareAcquiredRecords;
import org.apache.kafka.server.share.fetch.ShareFetch;
import org.apache.kafka.server.share.persister.NoOpShareStatePersister;
@@ -1711,6 +1712,7 @@ public class SharePartitionManagerTest {
.withShareFetchData(shareFetch)
.withReplicaManager(mockReplicaManager)
.withSharePartitions(sharePartitions)
+ .withPartitionMaxBytesStrategy(PartitionMaxBytesStrategy.type(PartitionMaxBytesStrategy.StrategyType.UNIFORM))
.build();
delayedShareFetchPurgatory.tryCompleteElseWatch(delayedShareFetch, delayedShareFetchWatchKeys);
@@ -1912,6 +1914,7 @@ public class SharePartitionManagerTest {
.withShareFetchData(shareFetch)
.withReplicaManager(mockReplicaManager)
.withSharePartitions(sharePartitions)
+ .withPartitionMaxBytesStrategy(PartitionMaxBytesStrategy.type(PartitionMaxBytesStrategy.StrategyType.UNIFORM))
.build();
delayedShareFetchPurgatory.tryCompleteElseWatch(delayedShareFetch, delayedShareFetchWatchKeys);
diff --git a/core/src/test/java/kafka/test/api/ShareConsumerTest.java b/core/src/test/java/kafka/test/api/ShareConsumerTest.java
index 89b357cf4c3..3da92f499a9 100644
--- a/core/src/test/java/kafka/test/api/ShareConsumerTest.java
+++ b/core/src/test/java/kafka/test/api/ShareConsumerTest.java
@@ -905,7 +905,7 @@ public class ShareConsumerTest {
shareConsumer.subscribe(Collections.singleton(tp.topic()));
ConsumerRecords<byte[], byte[]> records = shareConsumer.poll(Duration.ofMillis(5000));
- assertEquals(1, records.count());
+ assertEquals(2, records.count());
}
}
diff --git a/core/src/test/scala/unit/kafka/server/ShareFetchAcknowledgeRequestTest.scala b/core/src/test/scala/unit/kafka/server/ShareFetchAcknowledgeRequestTest.scala
index 73f9fce42e6..b38ec285c40 100644
--- a/core/src/test/scala/unit/kafka/server/ShareFetchAcknowledgeRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ShareFetchAcknowledgeRequestTest.scala
@@ -1310,7 +1310,7 @@ class ShareFetchAcknowledgeRequestTest(cluster: ClusterInstance) extends GroupCo
),
)
)
- def testShareFetchBrokerRespectsPartitionsSizeLimit(): Unit = {
+ def testShareFetchBrokerDoesNotRespectPartitionsSizeLimit(): Unit = {
val groupId: String = "group"
val memberId = Uuid.randomUuid()
@@ -1350,10 +1350,10 @@ class ShareFetchAcknowledgeRequestTest(cluster: ClusterInstance) extends GroupCo
.setPartitionIndex(partition)
.setErrorCode(Errors.NONE.code())
.setAcknowledgeErrorCode(Errors.NONE.code())
- .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(11), Collections.singletonList(1)))
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(12), Collections.singletonList(1)))
// The first 10 records will be consumed as it is. For the last 3 records, each of size MAX_PARTITION_BYTES/3,
- // only 2 of then will be consumed (offsets 10 and 11) because the inclusion of the third last record will exceed
- // the max partition bytes limit
+ // all 3 of then will be consumed (offsets 10, 11 and 12) because even though the inclusion of the third last record will exceed
+ // the max partition bytes limit. We should only consider the request level maxBytes as the hard limit.
val partitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
compareFetchResponsePartitions(expectedPartitionData, partitionData)
@@ -1412,15 +1412,15 @@ class ShareFetchAcknowledgeRequestTest(cluster: ClusterInstance) extends GroupCo
// mocking the behaviour of multiple share consumers from the same share group
val metadata1: ShareRequestMetadata = new ShareRequestMetadata(memberId1, ShareRequestMetadata.INITIAL_EPOCH)
val acknowledgementsMap1: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
- val shareFetchRequest1 = createShareFetchRequest(groupId, metadata1, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMap1)
+ val shareFetchRequest1 = createShareFetchRequest(groupId, metadata1, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMap1, minBytes = 100, maxBytes = 1500)
val metadata2: ShareRequestMetadata = new ShareRequestMetadata(memberId2, ShareRequestMetadata.INITIAL_EPOCH)
val acknowledgementsMap2: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
- val shareFetchRequest2 = createShareFetchRequest(groupId, metadata2, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMap2)
+ val shareFetchRequest2 = createShareFetchRequest(groupId, metadata2, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMap2, minBytes = 100, maxBytes = 1500)
val metadata3: ShareRequestMetadata = new ShareRequestMetadata(memberId3, ShareRequestMetadata.INITIAL_EPOCH)
val acknowledgementsMap3: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
- val shareFetchRequest3 = createShareFetchRequest(groupId, metadata3, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMap3)
+ val shareFetchRequest3 = createShareFetchRequest(groupId, metadata3, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMap3, minBytes = 100, maxBytes = 1500)
val shareFetchResponse1 = connectAndReceive[ShareFetchResponse](shareFetchRequest1)
val shareFetchResponse2 = connectAndReceive[ShareFetchResponse](shareFetchRequest2)
diff --git a/share/src/main/java/org/apache/kafka/server/share/fetch/PartitionMaxBytesStrategy.java b/share/src/main/java/org/apache/kafka/server/share/fetch/PartitionMaxBytesStrategy.java
new file mode 100644
index 00000000000..e0600e02842
--- /dev/null
+++ b/share/src/main/java/org/apache/kafka/server/share/fetch/PartitionMaxBytesStrategy.java
@@ -0,0 +1,77 @@
+/*
+ * 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.TopicIdPartition;
+
+import java.util.LinkedHashMap;
+import java.util.Locale;
+import java.util.Set;
+
+/**
+ * This interface helps identify the max bytes for topic partitions in a share fetch request based on different strategy types.
+ */
+public interface PartitionMaxBytesStrategy {
+
+ enum StrategyType {
+ UNIFORM;
+
+ @Override
+ public String toString() {
+ return super.toString().toLowerCase(Locale.ROOT);
+ }
+ }
+
+ /**
+ * Returns the partition max bytes for a given partition based on the strategy type.
+ *
+ * @param requestMaxBytes - The total max bytes available for the share fetch request
+ * @param partitions - The topic partitions in the order for which we compute the partition max bytes.
+ * @param acquiredPartitionsSize - The total partitions that have been acquired.
+ * @return the partition max bytes for the topic partitions
+ */
+ LinkedHashMap<TopicIdPartition, Integer> maxBytes(int requestMaxBytes, Set<TopicIdPartition> partitions, int acquiredPartitionsSize);
+
+ static PartitionMaxBytesStrategy type(StrategyType type) {
+ if (type == null)
+ throw new IllegalArgumentException("Strategy type cannot be null");
+ return switch (type) {
+ case UNIFORM -> PartitionMaxBytesStrategy::uniformPartitionMaxBytes;
+ };
+ }
+
+
+ private static LinkedHashMap<TopicIdPartition, Integer> uniformPartitionMaxBytes(int requestMaxBytes, Set<TopicIdPartition> partitions, int acquiredPartitionsSize) {
+ checkValidArguments(requestMaxBytes, partitions, acquiredPartitionsSize);
+ LinkedHashMap<TopicIdPartition, Integer> partitionMaxBytes = new LinkedHashMap<>();
+ partitions.forEach(partition -> partitionMaxBytes.put(partition, requestMaxBytes / acquiredPartitionsSize));
+ return partitionMaxBytes;
+ }
+
+ // Visible for testing.
+ static void checkValidArguments(int requestMaxBytes, Set<TopicIdPartition> partitions, int acquiredPartitionsSize) {
+ if (partitions == null || partitions.isEmpty()) {
+ throw new IllegalArgumentException("Partitions to generate max bytes is null or empty");
+ }
+ if (requestMaxBytes <= 0) {
+ throw new IllegalArgumentException("Request max bytes must be greater than 0");
+ }
+ if (acquiredPartitionsSize <= 0) {
+ throw new IllegalArgumentException("Acquired partitions size must be greater than 0");
+ }
+ }
+}
diff --git a/share/src/test/java/org/apache/kafka/server/share/fetch/PartitionMaxBytesStrategyTest.java b/share/src/test/java/org/apache/kafka/server/share/fetch/PartitionMaxBytesStrategyTest.java
new file mode 100644
index 00000000000..3c6c7ad2207
--- /dev/null
+++ b/share/src/test/java/org/apache/kafka/server/share/fetch/PartitionMaxBytesStrategyTest.java
@@ -0,0 +1,90 @@
+/*
+ * 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.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.server.share.fetch.PartitionMaxBytesStrategy.StrategyType;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class PartitionMaxBytesStrategyTest {
+
+ @Test
+ public void testConstructor() {
+ assertThrows(IllegalArgumentException.class, () -> PartitionMaxBytesStrategy.type(null));
+ assertDoesNotThrow(() -> PartitionMaxBytesStrategy.type(StrategyType.UNIFORM));
+ }
+
+ @Test
+ public void testCheckValidArguments() {
+ TopicIdPartition topicIdPartition1 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("topic1", 0));
+ TopicIdPartition topicIdPartition2 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("topic1", 1));
+ TopicIdPartition topicIdPartition3 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("topic2", 0));
+ Set<TopicIdPartition> partitions = new LinkedHashSet<>();
+ partitions.add(topicIdPartition1);
+ partitions.add(topicIdPartition2);
+ partitions.add(topicIdPartition3);
+
+ // acquired partitions size is 0.
+ assertThrows(IllegalArgumentException.class, () -> PartitionMaxBytesStrategy.checkValidArguments(
+ 100, partitions, 0));
+ // empty partitions set.
+ assertThrows(IllegalArgumentException.class, () -> PartitionMaxBytesStrategy.checkValidArguments(
+ 100, Collections.EMPTY_SET, 20));
+ // partitions is null.
+ assertThrows(IllegalArgumentException.class, () -> PartitionMaxBytesStrategy.checkValidArguments(
+ 100, null, 20));
+ // request max bytes is 0.
+ assertThrows(IllegalArgumentException.class, () -> PartitionMaxBytesStrategy.checkValidArguments(
+ 0, partitions, 20));
+
+ // Valid arguments.
+ assertDoesNotThrow(() -> PartitionMaxBytesStrategy.checkValidArguments(100, partitions, 20));
+ }
+
+ @Test
+ public void testUniformStrategy() {
+ PartitionMaxBytesStrategy partitionMaxBytesStrategy = PartitionMaxBytesStrategy.type(StrategyType.UNIFORM);
+ TopicIdPartition topicIdPartition1 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("topic1", 0));
+ TopicIdPartition topicIdPartition2 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("topic1", 1));
+ TopicIdPartition topicIdPartition3 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("topic2", 0));
+ Set<TopicIdPartition> partitions = new LinkedHashSet<>();
+ partitions.add(topicIdPartition1);
+ partitions.add(topicIdPartition2);
+ partitions.add(topicIdPartition3);
+
+ LinkedHashMap<TopicIdPartition, Integer> result = partitionMaxBytesStrategy.maxBytes(
+ 100, partitions, 3);
+ assertEquals(result.values().stream().toList(), List.of(33, 33, 33));
+
+ result = partitionMaxBytesStrategy.maxBytes(
+ 100, partitions, 5);
+ assertEquals(result.values().stream().toList(), List.of(20, 20, 20));
+ }
+}
(kafka) 03/03: KAFKA-18514: Refactor share module code to server and server-common (#18524)
Posted by ij...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ijuma pushed a commit to branch 4.0
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit 146eff7af0d0750d55ee750655d72636b21891b3
Author: Apoorv Mittal <ap...@gmail.com>
AuthorDate: Thu Jan 16 08:14:25 2025 +0000
KAFKA-18514: Refactor share module code to server and server-common (#18524)
As per the discussion with @ijuma and @mumrah, the `share` module seems not required and it's advised to user `server` and `server-common` instead. The PR moves the classes from `share` module to respective server related modules.
Following has been refactored in the PR:
- Moved Share Fetch, Acknowledge, Session, Context and Cache related classes to `server` module as the classes are used by `core` and `tools` modules.
- Moved `Persister` releated classes from `share` to `server-common` as the Persister classes though currently just being used by `core` module but in [near future](https://github.com/apache/kafka/pull/17775) will also be used by `group-coordinator`. Hence the Persister classes shouldn't go in `server`. The debate is mostly between `coordinator-common` vs `server-common`. We have kept the Persister in `server-common` for now, the classes are more related to the server than the coordin [...]
- Updated build.gradle to remove `share` module.
- Removed `import-control-share.xml`
Reviewers: Ismael Juma <is...@juma.me.uk>
---
build.gradle | 32 -----------
checkstyle/import-control-server-common.xml | 9 ++++
checkstyle/import-control-share.xml | 62 ----------------------
.../kafka/server/share/SharePartitionKey.java | 0
.../share/persister/DefaultStatePersister.java | 0
.../persister/DeleteShareGroupStateParameters.java | 0
.../persister/DeleteShareGroupStateResult.java | 0
.../share/persister/GroupTopicPartitionData.java | 0
.../InitializeShareGroupStateParameters.java | 0
.../persister/InitializeShareGroupStateResult.java | 0
.../share/persister/NoOpShareStatePersister.java | 0
.../server/share/persister/PartitionAllData.java | 0
.../server/share/persister/PartitionData.java | 0
.../server/share/persister/PartitionErrorData.java | 0
.../server/share/persister/PartitionFactory.java | 0
.../server/share/persister/PartitionIdData.java | 0
.../persister/PartitionIdLeaderEpochData.java | 0
.../server/share/persister/PartitionInfoData.java | 0
.../share/persister/PartitionStateBatchData.java | 0
.../server/share/persister/PartitionStateData.java | 0
.../share/persister/PartitionStateErrorData.java | 0
.../kafka/server/share/persister/Persister.java | 0
.../share/persister/PersisterParameters.java | 0
.../server/share/persister/PersisterResult.java | 0
.../share/persister/PersisterStateBatch.java | 0
.../share/persister/PersisterStateManager.java | 0
.../persister/ReadShareGroupStateParameters.java | 0
.../share/persister/ReadShareGroupStateResult.java | 0
.../ReadShareGroupStateSummaryParameters.java | 0
.../ReadShareGroupStateSummaryResult.java | 0
.../ShareCoordinatorMetadataCacheHelper.java | 0
.../kafka/server/share/persister/TopicData.java | 0
.../persister/WriteShareGroupStateParameters.java | 0
.../persister/WriteShareGroupStateResult.java | 0
.../share/persister/DefaultStatePersisterTest.java | 0
.../share/persister/PersisterStateManagerTest.java | 0
.../kafka/server/share/CachedSharePartition.java | 0
.../share/ErroneousAndValidPartitionData.java | 0
.../acknowledge/ShareAcknowledgementBatch.java | 0
.../kafka/server/share/context/FinalContext.java | 0
.../server/share/context/ShareFetchContext.java | 0
.../server/share/context/ShareSessionContext.java | 0
.../share/fetch/DelayedShareFetchGroupKey.java | 0
.../server/share/fetch/DelayedShareFetchKey.java | 0
.../share/fetch/DelayedShareFetchPartitionKey.java | 0
.../share/fetch/PartitionMaxBytesStrategy.java | 0
.../server/share/fetch/ShareAcquiredRecords.java | 0
.../kafka/server/share/fetch/ShareFetch.java | 0
.../kafka/server/share/session/LastUsedKey.java | 0
.../kafka/server/share/session/ShareSession.java | 0
.../server/share/session/ShareSessionCache.java | 0
.../server/share/session/ShareSessionKey.java | 0
.../server/share/CachedSharePartitionTest.java | 0
.../share/fetch/DelayedShareFetchKeyTest.java | 0
.../share/fetch/PartitionMaxBytesStrategyTest.java | 0
.../kafka/server/share/fetch/ShareFetchTest.java | 0
.../share/session/ShareSessionCacheTest.java | 0
.../server/share/session/ShareSessionTest.java | 0
58 files changed, 9 insertions(+), 94 deletions(-)
diff --git a/build.gradle b/build.gradle
index 54dca862a9f..ecc786b2d5f 100644
--- a/build.gradle
+++ b/build.gradle
@@ -1027,34 +1027,6 @@ project(':server') {
}
}
-project(':share') {
- base {
- archivesName = "kafka-share"
- }
-
- dependencies {
- implementation project(':server-common')
- implementation libs.slf4jApi
-
- testImplementation project(':clients').sourceSets.test.output
- testImplementation project(':server-common').sourceSets.test.output
-
- testImplementation libs.junitJupiter
- testImplementation libs.mockitoCore
- testImplementation testLog4j2Libs
-
- testRuntimeOnly runtimeTestLibs
- }
-
- checkstyle {
- configProperties = checkstyleConfigProperties("import-control-share.xml")
- }
-
- javadoc {
- enabled = false
- }
-}
-
project(':core') {
apply plugin: 'scala'
@@ -1106,7 +1078,6 @@ project(':core') {
implementation project(':storage')
implementation project(':server')
implementation project(':coordinator-common')
- implementation project(':share')
implementation project(':share-coordinator')
implementation libs.argparse4j
@@ -1131,7 +1102,6 @@ project(':core') {
testImplementation project(':server-common').sourceSets.test.output
testImplementation project(':storage:storage-api').sourceSets.test.output
testImplementation project(':server').sourceSets.test.output
- testImplementation project(':share').sourceSets.test.output
testImplementation project(':test-common')
testImplementation project(':test-common:test-common-api')
testImplementation libs.bcpkix
@@ -1767,7 +1737,6 @@ project(':share-coordinator') {
implementation project(':coordinator-common')
implementation project(':metadata')
implementation project(':server-common')
- implementation project(':share')
implementation libs.metrics
implementation libs.slf4jApi
@@ -2465,7 +2434,6 @@ project(':tools') {
implementation project(':group-coordinator')
implementation project(':coordinator-common')
implementation project(':share-coordinator')
- implementation project(':share')
implementation libs.argparse4j
implementation libs.jacksonDatabind
implementation libs.jacksonDataformatCsv
diff --git a/checkstyle/import-control-server-common.xml b/checkstyle/import-control-server-common.xml
index c576ea76f9a..ac870668661 100644
--- a/checkstyle/import-control-server-common.xml
+++ b/checkstyle/import-control-server-common.xml
@@ -98,6 +98,15 @@
<allow pkg="org.apache.kafka.server.util" />
</subpackage>
+ <subpackage name="share">
+ <allow pkg="org.apache.kafka.server.share" />
+ <subpackage name="persister">
+ <allow pkg="org.apache.kafka.clients" />
+ <allow pkg="org.apache.kafka.server.util" />
+ <allow pkg="org.apache.kafka.test" />
+ </subpackage>
+ </subpackage>
+
<subpackage name="util">
<!-- InterBrokerSendThread uses some clients classes that are not part of the public -->
<!-- API but are still relatively common -->
diff --git a/checkstyle/import-control-share.xml b/checkstyle/import-control-share.xml
deleted file mode 100644
index 8abc1b9f7b7..00000000000
--- a/checkstyle/import-control-share.xml
+++ /dev/null
@@ -1,62 +0,0 @@
-<!DOCTYPE import-control PUBLIC
- "-//Puppy Crawl//DTD Import Control 1.1//EN"
- "http://www.puppycrawl.com/dtds/import_control_1_1.dtd">
-<!--
- 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.
--->
-
-<import-control pkg="org.apache.kafka">
-
- <!-- THINK HARD ABOUT THE LAYERING OF THE PROJECT BEFORE CHANGING THIS FILE -->
-
- <!-- common library dependencies -->
- <allow pkg="java" />
- <allow pkg="org.slf4j" />
- <allow pkg="org.junit" />
- <allow pkg="org.mockito" />
-
- <!-- no one depends on the server -->
- <disallow pkg="kafka" />
-
- <!-- anyone can use public classes -->
- <allow pkg="org.apache.kafka.common" exact-match="true" />
- <allow pkg="org.apache.kafka.common.utils" />
- <allow pkg="org.apache.kafka.common.annotation" />
-
- <!-- protocol, records and request/response utilities -->
- <allow pkg="org.apache.kafka.common.message" />
- <allow pkg="org.apache.kafka.common.protocol" />
- <allow pkg="org.apache.kafka.common.requests" />
-
- <subpackage name="server">
- <subpackage name="share">
- <allow pkg="org.apache.kafka.server.share" />
-
- <subpackage name="fetch">
- <allow pkg="org.apache.kafka.server.purgatory" />
- <allow class="org.apache.kafka.server.storage.log.FetchParams"/>
- </subpackage>
-
- <subpackage name="persister">
- <allow pkg="org.apache.kafka.clients" />
- <allow pkg="org.apache.kafka.common.internals" />
- <allow pkg="org.apache.kafka.server.util" />
- <allow pkg="org.apache.kafka.test" />
- </subpackage>
- </subpackage>
- </subpackage>
-
-</import-control>
diff --git a/share/src/main/java/org/apache/kafka/server/share/SharePartitionKey.java b/server-common/src/main/java/org/apache/kafka/server/share/SharePartitionKey.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/SharePartitionKey.java
rename to server-common/src/main/java/org/apache/kafka/server/share/SharePartitionKey.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/DefaultStatePersister.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/DefaultStatePersister.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/DefaultStatePersister.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/DefaultStatePersister.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/DeleteShareGroupStateParameters.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/DeleteShareGroupStateParameters.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/DeleteShareGroupStateParameters.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/DeleteShareGroupStateParameters.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/DeleteShareGroupStateResult.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/DeleteShareGroupStateResult.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/DeleteShareGroupStateResult.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/DeleteShareGroupStateResult.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/GroupTopicPartitionData.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/GroupTopicPartitionData.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/GroupTopicPartitionData.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/GroupTopicPartitionData.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/InitializeShareGroupStateParameters.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/InitializeShareGroupStateParameters.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/InitializeShareGroupStateParameters.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/InitializeShareGroupStateParameters.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/InitializeShareGroupStateResult.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/InitializeShareGroupStateResult.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/InitializeShareGroupStateResult.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/InitializeShareGroupStateResult.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/NoOpShareStatePersister.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/NoOpShareStatePersister.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/NoOpShareStatePersister.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/NoOpShareStatePersister.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/PartitionAllData.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/PartitionAllData.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/PartitionAllData.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/PartitionAllData.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/PartitionData.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/PartitionData.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/PartitionData.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/PartitionData.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/PartitionErrorData.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/PartitionErrorData.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/PartitionErrorData.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/PartitionErrorData.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/PartitionFactory.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/PartitionFactory.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/PartitionFactory.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/PartitionFactory.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/PartitionIdData.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/PartitionIdData.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/PartitionIdData.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/PartitionIdData.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/PartitionIdLeaderEpochData.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/PartitionIdLeaderEpochData.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/PartitionIdLeaderEpochData.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/PartitionIdLeaderEpochData.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/PartitionInfoData.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/PartitionInfoData.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/PartitionInfoData.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/PartitionInfoData.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/PartitionStateBatchData.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/PartitionStateBatchData.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/PartitionStateBatchData.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/PartitionStateBatchData.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/PartitionStateData.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/PartitionStateData.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/PartitionStateData.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/PartitionStateData.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/PartitionStateErrorData.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/PartitionStateErrorData.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/PartitionStateErrorData.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/PartitionStateErrorData.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/Persister.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/Persister.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/Persister.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/Persister.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/PersisterParameters.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/PersisterParameters.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/PersisterParameters.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/PersisterParameters.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/PersisterResult.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/PersisterResult.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/PersisterResult.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/PersisterResult.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/PersisterStateBatch.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/PersisterStateBatch.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/PersisterStateBatch.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/PersisterStateBatch.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/ReadShareGroupStateParameters.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/ReadShareGroupStateParameters.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/ReadShareGroupStateParameters.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/ReadShareGroupStateParameters.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/ReadShareGroupStateResult.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/ReadShareGroupStateResult.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/ReadShareGroupStateResult.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/ReadShareGroupStateResult.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/ReadShareGroupStateSummaryParameters.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/ReadShareGroupStateSummaryParameters.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/ReadShareGroupStateSummaryParameters.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/ReadShareGroupStateSummaryParameters.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/ReadShareGroupStateSummaryResult.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/ReadShareGroupStateSummaryResult.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/ReadShareGroupStateSummaryResult.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/ReadShareGroupStateSummaryResult.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/ShareCoordinatorMetadataCacheHelper.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/ShareCoordinatorMetadataCacheHelper.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/ShareCoordinatorMetadataCacheHelper.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/ShareCoordinatorMetadataCacheHelper.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/TopicData.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/TopicData.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/TopicData.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/TopicData.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/WriteShareGroupStateParameters.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/WriteShareGroupStateParameters.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/WriteShareGroupStateParameters.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/WriteShareGroupStateParameters.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/persister/WriteShareGroupStateResult.java b/server-common/src/main/java/org/apache/kafka/server/share/persister/WriteShareGroupStateResult.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/persister/WriteShareGroupStateResult.java
rename to server-common/src/main/java/org/apache/kafka/server/share/persister/WriteShareGroupStateResult.java
diff --git a/share/src/test/java/org/apache/kafka/server/share/persister/DefaultStatePersisterTest.java b/server-common/src/test/java/org/apache/kafka/server/share/persister/DefaultStatePersisterTest.java
similarity index 100%
rename from share/src/test/java/org/apache/kafka/server/share/persister/DefaultStatePersisterTest.java
rename to server-common/src/test/java/org/apache/kafka/server/share/persister/DefaultStatePersisterTest.java
diff --git a/share/src/test/java/org/apache/kafka/server/share/persister/PersisterStateManagerTest.java b/server-common/src/test/java/org/apache/kafka/server/share/persister/PersisterStateManagerTest.java
similarity index 100%
rename from share/src/test/java/org/apache/kafka/server/share/persister/PersisterStateManagerTest.java
rename to server-common/src/test/java/org/apache/kafka/server/share/persister/PersisterStateManagerTest.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/CachedSharePartition.java b/server/src/main/java/org/apache/kafka/server/share/CachedSharePartition.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/CachedSharePartition.java
rename to server/src/main/java/org/apache/kafka/server/share/CachedSharePartition.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/ErroneousAndValidPartitionData.java b/server/src/main/java/org/apache/kafka/server/share/ErroneousAndValidPartitionData.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/ErroneousAndValidPartitionData.java
rename to server/src/main/java/org/apache/kafka/server/share/ErroneousAndValidPartitionData.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/acknowledge/ShareAcknowledgementBatch.java b/server/src/main/java/org/apache/kafka/server/share/acknowledge/ShareAcknowledgementBatch.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/acknowledge/ShareAcknowledgementBatch.java
rename to server/src/main/java/org/apache/kafka/server/share/acknowledge/ShareAcknowledgementBatch.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/context/FinalContext.java b/server/src/main/java/org/apache/kafka/server/share/context/FinalContext.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/context/FinalContext.java
rename to server/src/main/java/org/apache/kafka/server/share/context/FinalContext.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/context/ShareFetchContext.java b/server/src/main/java/org/apache/kafka/server/share/context/ShareFetchContext.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/context/ShareFetchContext.java
rename to server/src/main/java/org/apache/kafka/server/share/context/ShareFetchContext.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/context/ShareSessionContext.java b/server/src/main/java/org/apache/kafka/server/share/context/ShareSessionContext.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/context/ShareSessionContext.java
rename to server/src/main/java/org/apache/kafka/server/share/context/ShareSessionContext.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/fetch/DelayedShareFetchGroupKey.java b/server/src/main/java/org/apache/kafka/server/share/fetch/DelayedShareFetchGroupKey.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/fetch/DelayedShareFetchGroupKey.java
rename to server/src/main/java/org/apache/kafka/server/share/fetch/DelayedShareFetchGroupKey.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/fetch/DelayedShareFetchKey.java b/server/src/main/java/org/apache/kafka/server/share/fetch/DelayedShareFetchKey.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/fetch/DelayedShareFetchKey.java
rename to server/src/main/java/org/apache/kafka/server/share/fetch/DelayedShareFetchKey.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/fetch/DelayedShareFetchPartitionKey.java b/server/src/main/java/org/apache/kafka/server/share/fetch/DelayedShareFetchPartitionKey.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/fetch/DelayedShareFetchPartitionKey.java
rename to server/src/main/java/org/apache/kafka/server/share/fetch/DelayedShareFetchPartitionKey.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/fetch/PartitionMaxBytesStrategy.java b/server/src/main/java/org/apache/kafka/server/share/fetch/PartitionMaxBytesStrategy.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/fetch/PartitionMaxBytesStrategy.java
rename to server/src/main/java/org/apache/kafka/server/share/fetch/PartitionMaxBytesStrategy.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/fetch/ShareAcquiredRecords.java b/server/src/main/java/org/apache/kafka/server/share/fetch/ShareAcquiredRecords.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/fetch/ShareAcquiredRecords.java
rename to server/src/main/java/org/apache/kafka/server/share/fetch/ShareAcquiredRecords.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/fetch/ShareFetch.java b/server/src/main/java/org/apache/kafka/server/share/fetch/ShareFetch.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/fetch/ShareFetch.java
rename to server/src/main/java/org/apache/kafka/server/share/fetch/ShareFetch.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/session/LastUsedKey.java b/server/src/main/java/org/apache/kafka/server/share/session/LastUsedKey.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/session/LastUsedKey.java
rename to server/src/main/java/org/apache/kafka/server/share/session/LastUsedKey.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/session/ShareSession.java b/server/src/main/java/org/apache/kafka/server/share/session/ShareSession.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/session/ShareSession.java
rename to server/src/main/java/org/apache/kafka/server/share/session/ShareSession.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/session/ShareSessionCache.java b/server/src/main/java/org/apache/kafka/server/share/session/ShareSessionCache.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/session/ShareSessionCache.java
rename to server/src/main/java/org/apache/kafka/server/share/session/ShareSessionCache.java
diff --git a/share/src/main/java/org/apache/kafka/server/share/session/ShareSessionKey.java b/server/src/main/java/org/apache/kafka/server/share/session/ShareSessionKey.java
similarity index 100%
rename from share/src/main/java/org/apache/kafka/server/share/session/ShareSessionKey.java
rename to server/src/main/java/org/apache/kafka/server/share/session/ShareSessionKey.java
diff --git a/share/src/test/java/org/apache/kafka/server/share/CachedSharePartitionTest.java b/server/src/test/java/org/apache/kafka/server/share/CachedSharePartitionTest.java
similarity index 100%
rename from share/src/test/java/org/apache/kafka/server/share/CachedSharePartitionTest.java
rename to server/src/test/java/org/apache/kafka/server/share/CachedSharePartitionTest.java
diff --git a/share/src/test/java/org/apache/kafka/server/share/fetch/DelayedShareFetchKeyTest.java b/server/src/test/java/org/apache/kafka/server/share/fetch/DelayedShareFetchKeyTest.java
similarity index 100%
rename from share/src/test/java/org/apache/kafka/server/share/fetch/DelayedShareFetchKeyTest.java
rename to server/src/test/java/org/apache/kafka/server/share/fetch/DelayedShareFetchKeyTest.java
diff --git a/share/src/test/java/org/apache/kafka/server/share/fetch/PartitionMaxBytesStrategyTest.java b/server/src/test/java/org/apache/kafka/server/share/fetch/PartitionMaxBytesStrategyTest.java
similarity index 100%
rename from share/src/test/java/org/apache/kafka/server/share/fetch/PartitionMaxBytesStrategyTest.java
rename to server/src/test/java/org/apache/kafka/server/share/fetch/PartitionMaxBytesStrategyTest.java
diff --git a/share/src/test/java/org/apache/kafka/server/share/fetch/ShareFetchTest.java b/server/src/test/java/org/apache/kafka/server/share/fetch/ShareFetchTest.java
similarity index 100%
rename from share/src/test/java/org/apache/kafka/server/share/fetch/ShareFetchTest.java
rename to server/src/test/java/org/apache/kafka/server/share/fetch/ShareFetchTest.java
diff --git a/share/src/test/java/org/apache/kafka/server/share/session/ShareSessionCacheTest.java b/server/src/test/java/org/apache/kafka/server/share/session/ShareSessionCacheTest.java
similarity index 100%
rename from share/src/test/java/org/apache/kafka/server/share/session/ShareSessionCacheTest.java
rename to server/src/test/java/org/apache/kafka/server/share/session/ShareSessionCacheTest.java
diff --git a/share/src/test/java/org/apache/kafka/server/share/session/ShareSessionTest.java b/server/src/test/java/org/apache/kafka/server/share/session/ShareSessionTest.java
similarity index 100%
rename from share/src/test/java/org/apache/kafka/server/share/session/ShareSessionTest.java
rename to server/src/test/java/org/apache/kafka/server/share/session/ShareSessionTest.java