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 <[email protected]>
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 <[email protected]>, Abhinav Dixit
<[email protected]>
---
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);
