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 1 commit
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
14 changes: 12 additions & 2 deletions 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,13 @@ public boolean tryComplete() {
"topic partitions {}", shareFetch.groupId(), shareFetch.memberId(),
sharePartitions.keySet());
}
for (TopicIdPartition topicIdPartition : sharePartitions.keySet()) {
Partition partition = replicaManager.getPartitionOrException(topicIdPartition.topicPartition());
if (!partition.isLeader()) {
log.error("Broker is no longer the leader of topicIdPartition {}", topicIdPartition);
throw new NotLeaderOrFollowerException("Broker is no longer the leader of topicIdPartition: " + topicIdPartition);

@AndrewJSchofield AndrewJSchofield Jul 31, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

"topicPartition" :)

}
}
return false;
} catch (Exception e) {
log.error("Error processing delayed share fetch request", e);
Expand Down Expand Up @@ -765,15 +772,18 @@ 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 topicIdPartition: " + topicIdPartition);

@AndrewJSchofield AndrewJSchofield Jul 31, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Maybe "topicPartition" is more aligned with the rest of this code.

}
} catch (KafkaStorageException e) { // Case a
log.debug("TopicPartition {} is in an offline log directory, satisfy {} immediately", topicIdPartition, shareFetch.fetchParams());
canComplete = true;
} catch (UnknownTopicOrPartitionException e) { // Case b
log.debug("Broker no longer knows of topicPartition {}, satisfy {} immediately", topicIdPartition, shareFetch.fetchParams());
canComplete = true;
} catch (NotLeaderOrFollowerException e) { // Case c
log.debug("Broker is no longer the leader or follower of topicPartition {}, satisfy {} immediately", topicIdPartition, shareFetch.fetchParams());
log.debug("Broker is no longer the leader or follower of topicIdPartition {}, satisfy {} immediately", topicIdPartition, shareFetch.fetchParams());

@AndrewJSchofield AndrewJSchofield Jul 31, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

I'd revert this line because all of the other log lines just call it a topicPartition, in spite of technically having the topic ID. That's really fine because it is just an identifier for a topic partition.

canComplete = true;
}
if (canComplete)
Expand Down
162 changes: 162 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 All @@ -178,6 +189,60 @@ public void testDelayedShareFetchTryCompleteReturnsFalseDueToNonAcquirablePartit
delayedShareFetch.lock().unlock();
}

@Test
public void testDelayedShareFetchTryCompleteReturnsTrueDueToPartitionLeadershipChange() {
String groupId = "grp";
Uuid topicId = Uuid.randomUuid();
TopicIdPartition tp0 = new TopicIdPartition(topicId, new TopicPartition("foo", 0));
TopicIdPartition tp1 = new TopicIdPartition(topicId, new TopicPartition("foo", 1));

SharePartition sp0 = mock(SharePartition.class);
SharePartition sp1 = mock(SharePartition.class);

LinkedHashMap<TopicIdPartition, SharePartition> sharePartitions = new LinkedHashMap<>();
sharePartitions.put(tp0, sp0);
sharePartitions.put(tp1, sp1);

ShareFetch shareFetch = new ShareFetch(FETCH_PARAMS, groupId, Uuid.randomUuid().toString(),
new CompletableFuture<>(), List.of(tp0, tp1), BATCH_SIZE, MAX_FETCH_RECORDS,
BROKER_TOPIC_STATS);

when(sp0.canAcquireRecords()).thenReturn(false);
when(sp1.canAcquireRecords()).thenReturn(false);

// Since partitions are not leaders, the delayed share fetch should not be completed instantaneously.
Partition p0 = mock(Partition.class);
when(p0.isLeader()).thenReturn(false);

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

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);
when(sp1.maybeAcquireFetchLock(fetchId)).thenReturn(true);

assertTrue(delayedShareFetch.tryComplete());
assertTrue(delayedShareFetch.isCompleted());
Mockito.verify(delayedShareFetch, times(1)).releasePartitionLocks(any());
assertTrue(delayedShareFetch.lock().tryLock());
assertEquals(0, delayedShareFetch.expiredRequestMeter().count());

delayedShareFetch.lock().unlock();
}

@Test
public void testTryCompleteWhenMinBytesNotSatisfiedOnFirstFetch() {
String groupId = "grp";
Expand Down Expand Up @@ -218,6 +283,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 +361,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 +663,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 +833,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 +983,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 +1373,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 +1639,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 +1719,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 +1818,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 +1913,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
Loading