Repository: apache/kafka
File: core/src/test/java/kafka/server/share/DelayedShareFetchTest.java (Java)

Code before:
            .withTime(time)
            .withFetchId(fetchId)
            .build());

        when(sp0.maybeAcquireFetchLock(fetchId)).thenReturn(true);
        when(sp1.maybeAcquireFetchLock(fetchId)).thenReturn(true);

        try (MockedStatic<ShareFetchUtils> mockedShareFetchUtils = Mockito.mockStatic(ShareFetchUtils.class, Mockito.CALLS_REAL_METHODS)) {
            mockedShareFetchUtils.when(() -> ShareFetchUtils.processFetchResponse(any(), any(), any(), any(), any()))
                    .thenReturn(Map.of(tp0, mock(ShareFetchResponseData.PartitionData.class)));

            assertFalse(delayedShareFetch.isCompleted());

            // Since minBytes(2) is not satisfied (only 1 byte available from sp0), and sp1 cannot be acquired, tryComplete should return false.
            assertFalse(delayedShareFetch.tryComplete());
            assertFalse(delayedShareFetch.isCompleted());
            Mockito.verify(delayedShareFetch, times(1)).releasePartitionLocks(any());
            assertTrue(delayedShareFetch.lock().tryLock());
            // Though the request is not completed but sp0 was acquired and hence the metric should be recorded.
            assertEquals(1, shareGroupMetrics.topicPartitionsAcquireTimeMs(groupId).count());
            assertEquals(10, shareGroupMetrics.topicPartitionsAcquireTimeMs(groupId).sum());
            // Since the request is not completed, the fetch ratio should be null.
            assertNull(shareGroupMetrics.topicPartitionsFetchRatio(groupId));

            delayedShareFetch.lock().unlock();
            Mockito.verify(exceptionHandler, never()).accept(any(), any());
        }
    }

    @Test
    public void testTryCompleteWhenMinBytesNotSatisfiedOnSubsequentFetch() {
        String groupId = "grp";
        Uuid topicId = Uuid.randomUuid();
        ReplicaManager replicaManager = mock(ReplicaManager.class);
        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(
            new FetchParams(FetchRequest.ORDINARY_CONSUMER_ID, -1, MAX_WAIT_MS,
                2, 1024 * 1024, FetchIsolation.HIGH_WATERMARK, Optional.empty()), groupId, Uuid.randomUuid().toString(),
            new CompletableFuture<>(), List.of(tp0, tp1), BATCH_SIZE, MAX_FETCH_RECORDS,
            BROKER_TOPIC_STATS);

        when(sp0.canAcquireRecords()).thenReturn(true);

Review comment (bug):
`exceptionHandler` shouldn't be triggered in this test case. 
The reason it throw error simply due to incorrect test mock up, https://github.com/apache/kafka/blob/342a8e677340d19ad7c3f21b5664a6d629bdd9d3/core/src/test/java/kafka/server/share/DelayedShareFetchTest.java#L238 the above code snippet override the return value of `replicaManager.getPartitionOrException(tp0.topicPartition())` defined in https://github.com/apache/kafka/blob/342a8e677340d19ad7c3f21b5664a6d629bdd9d3/core/src/test/java/kafka/server/share/DelayedShareFetchTest.java#L225, which make isMinBytesSatisfied raise NPE https://github.com/apache/kafka/blob/43a25043dd06458754e4eee767f93cfd406936ef/core/src/main/java/kafka/server/share/DelayedShareFetch.java#L481, and this is unexpected.

Code after:
            .withTime(time)
            .withFetchId(fetchId)
            .build());

        when(sp0.maybeAcquireFetchLock(fetchId)).thenReturn(true);
        when(sp1.maybeAcquireFetchLock(fetchId)).thenReturn(true);

        assertFalse(delayedShareFetch.isCompleted());

        // Since minBytes(2) is not satisfied (only 1 byte available from sp0), and sp1 cannot be acquired, tryComplete should return false.
        assertFalse(delayedShareFetch.tryComplete());
        assertFalse(delayedShareFetch.isCompleted());
        Mockito.verify(delayedShareFetch, times(1)).releasePartitionLocks(any());
        assertTrue(delayedShareFetch.lock().tryLock());
        // Though the request is not completed but sp0 was acquired and hence the metric should be recorded.
        assertEquals(1, shareGroupMetrics.topicPartitionsAcquireTimeMs(groupId).count());
        assertEquals(10, shareGroupMetrics.topicPartitionsAcquireTimeMs(groupId).sum());
        // Since the request is not completed, the fetch ratio should be null.
        assertNull(shareGroupMetrics.topicPartitionsFetchRatio(groupId));

        delayedShareFetch.lock().unlock();
        Mockito.verify(exceptionHandler, never()).accept(any(), any());
    }

    @Test
    public void testTryCompleteWhenMinBytesNotSatisfiedOnSubsequentFetch() {
        String groupId = "grp";
        Uuid topicId = Uuid.randomUuid();
        ReplicaManager replicaManager = mock(ReplicaManager.class);
        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(
            new FetchParams(FetchRequest.ORDINARY_CONSUMER_ID, -1, MAX_WAIT_MS,
                2, 1024 * 1024, FetchIsolation.HIGH_WATERMARK, Optional.empty()), groupId, Uuid.randomUuid().toString(),
            new CompletableFuture<>(), List.of(tp0, tp1), BATCH_SIZE, MAX_FETCH_RECORDS,
            BROKER_TOPIC_STATS);

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

        // We are testing the case when the share partition has been fetched before, hence we are mocking positionDiff
        // functionality to give the file position difference as 1 byte, so it doesn't satisfy the minBytes(2).
        LogOffsetMetadata hwmOffsetMetadata = mock(LogOffsetMetadata.class);