You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mi...@apache.org on 2026/07/15 12:21:13 UTC
(kafka) branch trunk updated: KAFKA-20783: Release locks on exception for local reads within remote calls (5/N) (#22830)
This is an automated email from the ASF dual-hosted git repository.
apoorvmittal10 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 2cf3c3ad9b4 KAFKA-20783: Release locks on exception for local reads within remote calls (5/N) (#22830)
2cf3c3ad9b4 is described below
commit 2cf3c3ad9b47df75724b0cd6f17029bb702691ef
Author: Apoorv Mittal <ap...@gmail.com>
AuthorDate: Wed Jul 15 13:21:06 2026 +0100
KAFKA-20783: Release locks on exception for local reads within remote calls (5/N) (#22830)
This PR adds exception handling when the future for local reads raise
exception from calls within remote calls processing i.e. min bytes is
not satisfied and additional data needs to be fetched from local log.
PR also makes some tests to utilize common code.
Reviewers: Andrew Schofield <as...@confluent.io>, Abhinav Dixit
<ad...@confluent.io>
---
checkstyle/suppressions.xml | 2 +-
.../java/kafka/server/share/DelayedShareFetch.java | 8 +
.../kafka/server/share/DelayedShareFetchTest.java | 172 ++++++++++++++-------
3 files changed, 127 insertions(+), 55 deletions(-)
diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml
index ecf36ae6484..b1c71df9861 100644
--- a/checkstyle/suppressions.xml
+++ b/checkstyle/suppressions.xml
@@ -39,7 +39,7 @@
<suppress checks="NPathComplexity" files="(ClusterTestExtensions|KafkaApisBuilder|SharePartition|SharePartitionManager).java"/>
<suppress checks="NPathComplexity" files="TestKitNodes.java"/>
<suppress checks="JavaNCSS"
- files="(SharePartitionManagerTest|SharePartitionTest|ShareConsumerTest).java"/>
+ files="(SharePartitionManagerTest|SharePartitionTest|ShareConsumerTest|DelayedShareFetchTest).java"/>
<suppress checks="ClassDataAbstractionCoupling|ClassFanOutComplexity" files="SharePartitionManagerTest"/>
<suppress checks="CyclomaticComplexity" files="SharePartition.java"/>
diff --git a/core/src/main/java/kafka/server/share/DelayedShareFetch.java b/core/src/main/java/kafka/server/share/DelayedShareFetch.java
index a554f92f2f9..9c410261a29 100644
--- a/core/src/main/java/kafka/server/share/DelayedShareFetch.java
+++ b/core/src/main/java/kafka/server/share/DelayedShareFetch.java
@@ -1198,6 +1198,10 @@ public class DelayedShareFetch extends DelayedOperation {
// Handle synchronous exception from readFromLog()
log.error("Error initiating async fetch in remote completion for group {}, member {}",
shareFetch.groupId(), shareFetch.memberId(), e);
+ // processFetchResultAndComplete won't be invoked in this error path, so it won't release the
+ // locks for partitionsToFetch. Release them here to avoid leaking the local partition locks
+ // before completing with just the remote data.
+ releasePartitionLocksAndAddToActionQueue(partitionsToFetch.keySet(), Set.of());
// Continue to complete with just remote data.
completionConsumer.accept(List.of());
return;
@@ -1210,6 +1214,10 @@ public class DelayedShareFetch extends DelayedOperation {
if (throwable != null) {
log.error("Async fetch failed in remote completion for group {}, member {}",
shareFetch.groupId(), shareFetch.memberId(), throwable);
+ // processFetchResultAndComplete won't be invoked in this error path, so it won't release
+ // the locks for partitionsToFetch. Release them here to avoid leaking the local partition
+ // locks before completing with just the remote data.
+ releasePartitionLocksAndAddToActionQueue(partitionsToFetch.keySet(), Set.of());
// Continue to complete with just remote data.
completionConsumer.accept(List.of());
return;
diff --git a/core/src/test/java/kafka/server/share/DelayedShareFetchTest.java b/core/src/test/java/kafka/server/share/DelayedShareFetchTest.java
index 55219fa14c3..ff2bb171dae 100644
--- a/core/src/test/java/kafka/server/share/DelayedShareFetchTest.java
+++ b/core/src/test/java/kafka/server/share/DelayedShareFetchTest.java
@@ -2653,6 +2653,20 @@ public class DelayedShareFetchTest {
@SuppressWarnings("unchecked")
@Test
public void testAsyncReadMinBytesNotSatisfiedClearsAcquiredStateForRetry() throws Exception {
+ // On next re-try, tryComplete() must not assume the released locks are still held. Since the acquired
+ // state was cleared, it re-acquires the partitions afresh (a second maybeAcquireFetchLock).
+ verifyAsyncReadMinBytesNotSatisfiedClearsAcquiredState(false);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void testAsyncReadMinBytesNotSatisfiedClearsAcquiredStateForOnComplete() throws Exception {
+ // On timeout, onComplete() must not assume the released locks are still held. Since the acquired
+ // state was cleared, it re-acquires the partitions afresh (a second maybeAcquireFetchLock).
+ verifyAsyncReadMinBytesNotSatisfiedClearsAcquiredState(true);
+ }
+
+ private void verifyAsyncReadMinBytesNotSatisfiedClearsAcquiredState(boolean requestTimedOut) throws Exception {
String groupId = "grp";
Uuid topicId = Uuid.randomUuid();
LogReader logReader = mock(LogReader.class);
@@ -2713,79 +2727,117 @@ public class DelayedShareFetchTest {
assertNull(delayedShareFetch.pendingFetch());
Mockito.verify(sp0, times(1)).releaseFetchLock(fetchId);
- // On next re-try, tryComplete() must not assume the released locks are still held. Since the acquired
- // state was cleared, it re-acquires the partitions afresh (a second maybeAcquireFetchLock).
- delayedShareFetch.tryComplete();
+ // Trigger onComplete() if requestTimedOut else tryComplete(). The re-try must not assume the
+ // released locks are still held. Since the acquired state was cleared, it re-acquires the
+ // partitions afresh (a second maybeAcquireFetchLock).
+ if (requestTimedOut) {
+ delayedShareFetch.forceComplete();
+ } else {
+ delayedShareFetch.tryComplete();
+ }
Mockito.verify(sp0, times(2)).maybeAcquireFetchLock(fetchId);
}
- @SuppressWarnings("unchecked")
@Test
- public void testAsyncReadMinBytesNotSatisfiedClearsAcquiredStateForOnComplete() throws Exception {
- String groupId = "grp";
- Uuid topicId = Uuid.randomUuid();
- LogReader logReader = mock(LogReader.class);
+ public void testRemoteStorageFetchLocalReadFailureReleasesLocks() {
+ verifyRemoteStorageFetchLocalReadFailureReleasesLocks(true);
+ }
+
+ @Test
+ public void testRemoteStorageFetchLocalReadCompletionFailureReleasesLocks() {
+ verifyRemoteStorageFetchLocalReadFailureReleasesLocks(false);
+ }
+
+ private void verifyRemoteStorageFetchLocalReadFailureReleasesLocks(boolean syncFailure) {
ReplicaManager replicaManager = mock(ReplicaManager.class);
+ LogReader logReader = mock(LogReader.class);
PartitionMetadataProvider metadataProvider = mock(PartitionMetadataProvider.class);
- TopicIdPartition tp0 = new TopicIdPartition(topicId, new TopicPartition("foo", 0));
+ 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));
SharePartition sp0 = mock(SharePartition.class);
+ SharePartition sp1 = mock(SharePartition.class);
+ SharePartition sp2 = mock(SharePartition.class);
+ when(sp0.canAcquireRecords()).thenReturn(true);
+ when(sp1.canAcquireRecords()).thenReturn(true);
+ when(sp2.canAcquireRecords()).thenReturn(true);
+
LinkedHashMap<TopicIdPartition, SharePartition> sharePartitions = new LinkedHashMap<>();
sharePartitions.put(tp0, sp0);
+ sharePartitions.put(tp1, sp1);
+ sharePartitions.put(tp2, sp2);
- // Set a high minBytes so that the async read result does not satisfy minBytes and the request
- // is returned to the purgatory.
- FetchParams fetchParams = new FetchParams(FetchRequest.ORDINARY_CONSUMER_ID, -1, MAX_WAIT_MS, 5000, 10000,
- FetchIsolation.HIGH_WATERMARK, Optional.empty());
CompletableFuture<Map<TopicIdPartition, ShareFetchResponseData.PartitionData>> future = new CompletableFuture<>();
- ShareFetch shareFetch = new ShareFetch(fetchParams, groupId, Uuid.randomUuid().toString(),
- future, List.of(tp0), BATCH_OPTIMIZED, BATCH_SIZE, MAX_FETCH_RECORDS, BROKER_TOPIC_STATS);
+ ShareFetch shareFetch = new ShareFetch(FETCH_PARAMS, "grp", Uuid.randomUuid().toString(),
+ future, List.of(tp0, tp1, tp2), BATCH_OPTIMIZED, BATCH_SIZE, MAX_FETCH_RECORDS, BROKER_TOPIC_STATS);
- CompletableFuture<LinkedHashMap<TopicIdPartition, LogReadResult>> asyncReadFuture = mock(CompletableFuture.class);
- when(logReader.readAsync(any(), any(), any(), any(), anyBoolean())).thenReturn(asyncReadFuture);
- // isDone returns false for the first tryComplete (store pending fetch) and true for the second
- // tryComplete (process the completed result).
- when(asyncReadFuture.isDone()).thenReturn(false).thenReturn(true);
- when(asyncReadFuture.get()).thenReturn(buildLogReadResultMap(List.of(tp0)));
+ when(sp0.nextFetchOffset()).thenReturn(10L);
+ when(sp1.nextFetchOffset()).thenReturn(20L);
+ when(sp2.nextFetchOffset()).thenReturn(30L);
+ when(sp0.fetchOffsetMetadata(anyLong())).thenReturn(Optional.empty());
+ when(sp1.fetchOffsetMetadata(anyLong())).thenReturn(Optional.empty());
+ when(sp2.fetchOffsetMetadata(anyLong())).thenReturn(Optional.empty());
- when(sp0.maybeAcquireFetchLock(any())).thenReturn(true);
- when(sp0.canAcquireRecords()).thenReturn(true);
- when(sp0.nextFetchOffset()).thenReturn(0L);
- when(sp0.fetchOffsetMetadata(anyLong()))
- .thenReturn(Optional.empty())
- .thenReturn(Optional.of(new LogOffsetMetadata(0, 1, 0)));
- when(metadataProvider.endOffsetMetadata(any(), any())).thenReturn(new LogOffsetMetadata(1, 1, 1));
+ // First read (tryComplete): tp0, tp1 local and tp2 remote fetch. The second read (additional local
+ // read in onComplete) fails, exercising an error path in readAndProcessFetchResultCompletion:
+ // either synchronously (readAsync throws) or asynchronously (an exceptionally-completed future).
+ LinkedHashMap<TopicIdPartition, LogReadResult> firstReadResult = buildLocalAndRemoteFetchResultMap(Set.of(tp0, tp1), Set.of(tp2));
+ if (syncFailure) {
+ when(logReader.readAsync(any(), any(), any(), any(), anyBoolean()))
+ .thenReturn(CompletableFuture.completedFuture(firstReadResult))
+ .thenThrow(new RuntimeException("sync local read failure"));
+ } else {
+ when(logReader.readAsync(any(), any(), any(), any(), anyBoolean()))
+ .thenReturn(CompletableFuture.completedFuture(firstReadResult))
+ .thenReturn(CompletableFuture.failedFuture(new RuntimeException("async local read failure")));
+ }
- PartitionMaxBytesStrategy partitionMaxBytesStrategy = mockPartitionMaxBytes(Set.of(tp0));
+ try (MockedStatic<ShareFetchUtils> mockedShareFetchUtils = Mockito.mockStatic(ShareFetchUtils.class)) {
+ mockedShareFetchUtils.when(() -> ShareFetchUtils.processFetchResponse(any(), any(), any(), any(), any()))
+ .thenReturn(Map.of(tp2, mock(ShareFetchResponseData.PartitionData.class)));
- Uuid fetchId = Uuid.randomUuid();
- DelayedShareFetch delayedShareFetch = DelayedShareFetchBuilder.builder()
- .withShareFetchData(shareFetch)
- .withLogReader(logReader)
- .withMetadataProvider(metadataProvider)
- .withPartitionMaxBytesStrategy(partitionMaxBytesStrategy)
- .withSharePartitions(sharePartitions)
- .withReplicaManager(replicaManager)
- .withFetchId(fetchId)
- .build();
+ // Remote fetch for tp2 completes within tryComplete so the request moves on to
+ // onComplete where the additional local read is attempted.
+ RemoteLogReadResult remoteFetchResult = new RemoteLogReadResult(Optional.of(REMOTE_FETCH_INFO), Optional.empty());
+ RemoteLogManager remoteLogManager = mock(RemoteLogManager.class);
+ doAnswer(invocationOnMock -> {
+ Consumer<RemoteLogReadResult> callback = invocationOnMock.getArgument(1);
+ callback.accept(remoteFetchResult);
+ return CompletableFuture.completedFuture(remoteFetchResult);
+ }).when(remoteLogManager).asyncRead(any(), any());
+ when(replicaManager.remoteLogManager()).thenReturn(Option.apply(remoteLogManager));
- // First tryComplete: async read not done, so a pending fetch is created and tp0 lock is held
- // (partitionsAcquired is populated internally).
- assertNull(delayedShareFetch.pendingFetch());
- assertFalse(delayedShareFetch.tryComplete());
- assertNotNull(delayedShareFetch.pendingFetch());
- Mockito.verify(sp0, times(1)).maybeAcquireFetchLock(fetchId);
+ // maybeCompletePendingRemoteFetch checks partition leadership for the remote fetch partition.
+ Partition p2 = mock(Partition.class);
+ when(p2.isLeader()).thenReturn(true);
+ when(replicaManager.getPartitionOrException(tp2.topicPartition())).thenReturn(p2);
- // Second tryComplete (purgatory re-check): async read is done but minBytes is not satisfied, so
- // the lock is released, the pending fetch is cleared and the request returns to the purgatory.
- assertFalse(delayedShareFetch.tryComplete());
- assertNull(delayedShareFetch.pendingFetch());
- Mockito.verify(sp0, times(1)).releaseFetchLock(fetchId);
+ Uuid fetchId = Uuid.randomUuid();
+ DelayedShareFetch delayedShareFetch = spy(DelayedShareFetchBuilder.builder()
+ .withShareFetchData(shareFetch)
+ .withReplicaManager(replicaManager)
+ .withLogReader(logReader)
+ .withMetadataProvider(metadataProvider)
+ .withSharePartitions(sharePartitions)
+ .withPartitionMaxBytesStrategy(mockPartitionMaxBytes(Set.of(tp0, tp1, tp2)))
+ .withFetchId(fetchId)
+ .build());
- // On timeout, onComplete() must not assume the released locks are still held. Since the acquired
- // state was cleared, it re-acquires the partitions afresh (a second maybeAcquireFetchLock).
- delayedShareFetch.forceComplete();
- Mockito.verify(sp0, times(2)).maybeAcquireFetchLock(fetchId);
+ when(sp0.maybeAcquireFetchLock(fetchId)).thenReturn(true);
+ when(sp1.maybeAcquireFetchLock(fetchId)).thenReturn(true);
+ when(sp2.maybeAcquireFetchLock(fetchId)).thenReturn(true);
+
+ assertTrue(delayedShareFetch.tryComplete());
+ assertTrue(delayedShareFetch.isCompleted());
+ assertTrue(shareFetch.isCompleted());
+
+ // tp0 and tp1 (local partitions being fetched) locks must be released even though the
+ // additional local read failed: once when the remote fetch is prepared (as non-remote
+ // partitions) and once in the error path. tp2 (remote fetch) lock is released once.
+ Mockito.verify(delayedShareFetch, times(2)).releasePartitionLocks(Set.of(tp0, tp1));
+ Mockito.verify(delayedShareFetch, times(1)).releasePartitionLocks(Set.of(tp2));
+ }
}
static void mockTopicIdPartitionToReturnDataEqualToMinBytes(ReplicaManager replicaManager, TopicIdPartition topicIdPartition, int minBytes) {
@@ -2844,6 +2896,18 @@ public class DelayedShareFetchTest {
return CollectionConverters.asScala(logReadResults).toSeq();
}
+ private LinkedHashMap<TopicIdPartition, LogReadResult> buildLocalAndRemoteFetchResultMap(
+ Set<TopicIdPartition> localLogReadTopicIdPartitions,
+ Set<TopicIdPartition> remoteReadTopicIdPartitions) {
+ return CollectionConverters.asJava(buildLocalAndRemoteFetchResult(localLogReadTopicIdPartitions, remoteReadTopicIdPartitions)).stream()
+ .collect(Collectors.toMap(
+ Tuple2::_1,
+ Tuple2::_2,
+ (oldValue, newValue) -> oldValue,
+ LinkedHashMap::new
+ ));
+ }
+
@SuppressWarnings("unchecked")
private static BiConsumer<SharePartitionKey, Throwable> mockExceptionHandler() {
return mock(BiConsumer.class);