Skip to content

KAFKA-19567: Added the check for underlying partition being the leader in delayedShareFetch tryComplete method #20280

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
apoorvmittal10 merged 7 commits into from
Aug 10, 2025
Merged

KAFKA-19567: Added the check for underlying partition being the leader in delayedShareFetch tryComplete method #20280

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
a3a2f79
Added the check for underlying partition being the leader in delayedS…
chirag-wadhwa5 Jul 31, 2025
594d79a
KAFKA-19567: Removed the change for checking leadership change and ad…
chirag-wadhwa5 Aug 4, 2025
fcd4a79
changed a few comment lines and resolved a test failure
chirag-wadhwa5 Aug 4, 2025
4d6c123
Minor changes and added a test case
chirag-wadhwa5 Aug 6, 2025
3b0faae
KAFKA-19567: Minor changes and improvements
chirag-wadhwa5 Aug 8, 2025
b6716e8
KAFKA-19567: Changed one small log line
chirag-wadhwa5 Aug 8, 2025
de22d8f
Merge remote-tracking branch 'upstream/trunk' into KAFKA-19567
chirag-wadhwa5 Aug 8, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion core/src/main/java/kafka/server/share/DelayedShareFetch.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -368,6 +368,12 @@ public boolean tryComplete() {
"topic partitions {}", shareFetch.groupId(), shareFetch.memberId(),
sharePartitions.keySet());
}
// At this point, there could be delayed requests sitting in the purgatory which are waiting on
// DelayedShareFetchPartitionKeys corresponding to partitions, whose leader has been changed to a different broker.
// In that case, such partitions would not be able to get acquired, and the tryComplete will keep on returning false.
// Eventually the operation will get timed out and completed, but it might not get removed from the purgatory.
// This has been eventually left it like this because the purge interval will make sure that the remaining operations
// in the purgatory do not grow indefinitely and are purged time to time.

@apoorvmittal10 apoorvmittal10 Aug 6, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is it time to time, or when keys or operations in purgatory are exceeded against a config. Can you please be explicit in the comment.

return false;
} catch (Exception e) {
log.error("Error processing delayed share fetch request", e);
Expand Down Expand Up @@ -765,7 +771,10 @@ private boolean maybeCompletePendingRemoteFetch() {

for (TopicIdPartition topicIdPartition : pendingRemoteFetchesOpt.get().fetchOffsetMetadataMap().keySet()) {
try {
replicaManager.getPartitionOrException(topicIdPartition.topicPartition());
Partition partition = replicaManager.getPartitionOrException(topicIdPartition.topicPartition());
if (!partition.isLeader()) {
throw new NotLeaderOrFollowerException("Broker is no longer the leader of topicPartition: " + topicIdPartition);

@apoorvmittal10 apoorvmittal10 Aug 6, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Are you certain the partition is also not follower? If not then shall we use NotLeaderException?

}
} catch (KafkaStorageException e) { // Case a
log.debug("TopicPartition {} is in an offline log directory, satisfy {} immediately", topicIdPartition, shareFetch.fetchParams());
canComplete = true;
Expand Down
1 change: 1 addition & 0 deletions core/src/main/java/kafka/server/share/SharePartitionManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -774,6 +774,7 @@ private static void removeSharePartitionFromCache(
if (sharePartition != null) {
sharePartition.markFenced();
replicaManager.removeListener(sharePartitionKey.topicIdPartition().topicPartition(), sharePartition.listener());
replicaManager.completeDelayedShareFetchRequest(new DelayedShareFetchGroupKey(sharePartitionKey.groupId(), sharePartitionKey.topicIdPartition()));
}
}

Expand Down
108 changes: 108 additions & 0 deletions core/src/test/java/kafka/server/share/DelayedShareFetchTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -153,13 +153,24 @@ public void testDelayedShareFetchTryCompleteReturnsFalseDueToNonAcquirablePartit
when(sp0.canAcquireRecords()).thenReturn(false);
when(sp1.canAcquireRecords()).thenReturn(false);

Partition p0 = mock(Partition.class);
when(p0.isLeader()).thenReturn(true);

Partition p1 = mock(Partition.class);
when(p1.isLeader()).thenReturn(true);

ReplicaManager replicaManager = mock(ReplicaManager.class);
when(replicaManager.getPartitionOrException(tp0.topicPartition())).thenReturn(p0);
when(replicaManager.getPartitionOrException(tp1.topicPartition())).thenReturn(p1);

ShareGroupMetrics shareGroupMetrics = new ShareGroupMetrics(new MockTime());
Uuid fetchId = Uuid.randomUuid();
DelayedShareFetch delayedShareFetch = spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
.withSharePartitions(sharePartitions)
.withShareGroupMetrics(shareGroupMetrics)
.withFetchId(fetchId)
.withReplicaManager(replicaManager)
.build());

when(sp0.maybeAcquireFetchLock(fetchId)).thenReturn(true);
Expand Down Expand Up @@ -218,6 +229,15 @@ public void testTryCompleteWhenMinBytesNotSatisfiedOnFirstFetch() {

PartitionMaxBytesStrategy partitionMaxBytesStrategy = mockPartitionMaxBytes(Set.of(tp0));

Partition p0 = mock(Partition.class);
when(p0.isLeader()).thenReturn(true);

Partition p1 = mock(Partition.class);
when(p1.isLeader()).thenReturn(true);

when(replicaManager.getPartitionOrException(tp0.topicPartition())).thenReturn(p0);
when(replicaManager.getPartitionOrException(tp1.topicPartition())).thenReturn(p1);

Time time = mock(Time.class);
when(time.hiResClockMs()).thenReturn(100L).thenReturn(110L);
ShareGroupMetrics shareGroupMetrics = new ShareGroupMetrics(time);
Expand Down Expand Up @@ -287,6 +307,15 @@ public void testTryCompleteWhenMinBytesNotSatisfiedOnSubsequentFetch() {
mockTopicIdPartitionFetchBytes(replicaManager, tp0, hwmOffsetMetadata);
BiConsumer<SharePartitionKey, Throwable> exceptionHandler = mockExceptionHandler();

Partition p0 = mock(Partition.class);
when(p0.isLeader()).thenReturn(true);

Partition p1 = mock(Partition.class);
when(p1.isLeader()).thenReturn(true);

when(replicaManager.getPartitionOrException(tp0.topicPartition())).thenReturn(p0);
when(replicaManager.getPartitionOrException(tp1.topicPartition())).thenReturn(p1);

Uuid fetchId = Uuid.randomUuid();
DelayedShareFetch delayedShareFetch = spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
Expand Down Expand Up @@ -580,6 +609,19 @@ public void testForceCompleteTriggersDelayedActionsQueue() {
List<DelayedOperationKey> delayedShareFetchWatchKeys = new ArrayList<>();
topicIdPartitions1.forEach(topicIdPartition -> delayedShareFetchWatchKeys.add(new DelayedShareFetchGroupKey(groupId, topicIdPartition.topicId(), topicIdPartition.partition())));

Partition p0 = mock(Partition.class);
when(p0.isLeader()).thenReturn(true);

Partition p1 = mock(Partition.class);
when(p1.isLeader()).thenReturn(true);

Partition p2 = mock(Partition.class);
when(p2.isLeader()).thenReturn(true);

when(replicaManager.getPartitionOrException(tp0.topicPartition())).thenReturn(p0);
when(replicaManager.getPartitionOrException(tp1.topicPartition())).thenReturn(p1);
when(replicaManager.getPartitionOrException(tp2.topicPartition())).thenReturn(p2);

Uuid fetchId1 = Uuid.randomUuid();
DelayedShareFetch delayedShareFetch1 = DelayedShareFetchTest.DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch1)
Expand Down Expand Up @@ -737,6 +779,12 @@ public void testExceptionInMinBytesCalculation() {
when(time.hiResClockMs()).thenReturn(100L).thenReturn(110L).thenReturn(170L);
ShareGroupMetrics shareGroupMetrics = new ShareGroupMetrics(time);
Uuid fetchId = Uuid.randomUuid();

Partition p0 = mock(Partition.class);
when(p0.isLeader()).thenReturn(true);

when(replicaManager.getPartitionOrException(tp0.topicPartition())).thenReturn(p0);

DelayedShareFetch delayedShareFetch = spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
.withSharePartitions(sharePartitions)
Expand Down Expand Up @@ -881,10 +929,18 @@ public void testLocksReleasedAcquireException() {
BROKER_TOPIC_STATS);

Uuid fetchId = Uuid.randomUuid();

Partition p0 = mock(Partition.class);
when(p0.isLeader()).thenReturn(true);

ReplicaManager replicaManager = mock(ReplicaManager.class);
when(replicaManager.getPartitionOrException(tp0.topicPartition())).thenReturn(p0);

DelayedShareFetch delayedShareFetch = DelayedShareFetchTest.DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
.withSharePartitions(sharePartitions)
.withFetchId(fetchId)
.withReplicaManager(replicaManager)
.build();

when(sp0.maybeAcquireFetchLock(fetchId)).thenReturn(true);
Expand Down Expand Up @@ -1263,6 +1319,19 @@ public void testRemoteStorageFetchTryCompleteReturnsFalse() {
when(remoteLogManager.asyncRead(any(), any())).thenReturn(mock(Future.class));
when(replicaManager.remoteLogManager()).thenReturn(Option.apply(remoteLogManager));

Partition p0 = mock(Partition.class);
when(p0.isLeader()).thenReturn(true);

Partition p1 = mock(Partition.class);
when(p1.isLeader()).thenReturn(true);

Partition p2 = mock(Partition.class);
when(p2.isLeader()).thenReturn(true);

when(replicaManager.getPartitionOrException(tp0.topicPartition())).thenReturn(p0);
when(replicaManager.getPartitionOrException(tp1.topicPartition())).thenReturn(p1);
when(replicaManager.getPartitionOrException(tp2.topicPartition())).thenReturn(p2);

Uuid fetchId = Uuid.randomUuid();
DelayedShareFetch delayedShareFetch = spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
Expand Down Expand Up @@ -1516,6 +1585,16 @@ public void testRemoteStorageFetchRequestCompletionOnFutureCompletionFailure() {
when(replicaManager.remoteLogManager()).thenReturn(Option.apply(remoteLogManager));

Uuid fetchId = Uuid.randomUuid();

Partition p0 = mock(Partition.class);
when(p0.isLeader()).thenReturn(true);

Partition p1 = mock(Partition.class);
when(p1.isLeader()).thenReturn(true);

when(replicaManager.getPartitionOrException(tp0.topicPartition())).thenReturn(p0);
when(replicaManager.getPartitionOrException(tp1.topicPartition())).thenReturn(p1);

DelayedShareFetch delayedShareFetch = spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
.withSharePartitions(sharePartitions)
Expand Down Expand Up @@ -1586,6 +1665,12 @@ public void testRemoteStorageFetchRequestCompletionOnFutureCompletionSuccessfull
when(replicaManager.remoteLogManager()).thenReturn(Option.apply(remoteLogManager));

Uuid fetchId = Uuid.randomUuid();

Partition p0 = mock(Partition.class);
when(p0.isLeader()).thenReturn(true);

when(replicaManager.getPartitionOrException(tp0.topicPartition())).thenReturn(p0);

DelayedShareFetch delayedShareFetch = spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
.withSharePartitions(sharePartitions)
Expand Down Expand Up @@ -1679,6 +1764,19 @@ public void testRemoteStorageFetchRequestCompletionAlongWithLocalLogRead() {
}).when(remoteLogManager).asyncRead(any(), any());
when(replicaManager.remoteLogManager()).thenReturn(Option.apply(remoteLogManager));

Partition p0 = mock(Partition.class);
when(p0.isLeader()).thenReturn(true);

Partition p1 = mock(Partition.class);
when(p1.isLeader()).thenReturn(true);

Partition p2 = mock(Partition.class);
when(p2.isLeader()).thenReturn(true);

when(replicaManager.getPartitionOrException(tp0.topicPartition())).thenReturn(p0);
when(replicaManager.getPartitionOrException(tp1.topicPartition())).thenReturn(p1);
when(replicaManager.getPartitionOrException(tp2.topicPartition())).thenReturn(p2);

Uuid fetchId = Uuid.randomUuid();
DelayedShareFetch delayedShareFetch = spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
Expand Down Expand Up @@ -1761,6 +1859,16 @@ public void testRemoteStorageFetchHappensForAllTopicPartitions() {
when(replicaManager.remoteLogManager()).thenReturn(Option.apply(remoteLogManager));

Uuid fetchId = Uuid.randomUuid();

Partition p0 = mock(Partition.class);
when(p0.isLeader()).thenReturn(true);

Partition p1 = mock(Partition.class);
when(p1.isLeader()).thenReturn(true);

when(replicaManager.getPartitionOrException(tp0.topicPartition())).thenReturn(p0);
when(replicaManager.getPartitionOrException(tp1.topicPartition())).thenReturn(p1);

@apoorvmittal10 apoorvmittal10 Aug 6, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Did we add a new test case that validates the added functionality?

@chirag-wadhwa5 chirag-wadhwa5 Aug 6, 2025

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the review. I have made the required updates and added a test case for the remote fetch throwing a NotLeaderException as well


DelayedShareFetch delayedShareFetch = spy(DelayedShareFetchBuilder.builder()
.withShareFetchData(shareFetch)
.withSharePartitions(sharePartitions)
Expand Down
3 changes: 2 additions & 1 deletion core/src/test/java/kafka/server/share/SharePartitionManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2622,7 +2622,8 @@ public void testSharePartitionPartialInitializationFailure() throws Exception {
assertEquals(Errors.FENCED_STATE_EPOCH.code(), partitionDataMap.get(tp2).errorCode());
assertEquals("Fenced state epoch", partitionDataMap.get(tp2).errorMessage());

Mockito.verify(replicaManager, times(0)).completeDelayedShareFetchRequest(any());
Mockito.verify(replicaManager, times(1)).completeDelayedShareFetchRequest(
new DelayedShareFetchGroupKey(groupId, tp2));
Mockito.verify(replicaManager, times(1)).readFromLog(
any(), any(), any(ReplicaQuota.class), anyBoolean());
// Should have 1 fetch recorded and 1 failure as single topic has multiple partition fetch
Expand Down