brandboat commented on code in PR #20765:
URL: https://github.com/apache/kafka/pull/20765#discussion_r2461159383
##########
core/src/test/java/kafka/server/share/DelayedShareFetchTest.java:
##########
@@ -256,21 +252,26 @@ public void
testTryCompleteWhenMinBytesNotSatisfiedOnFirstFetch() {
when(sp0.maybeAcquireFetchLock(fetchId)).thenReturn(true);
when(sp1.maybeAcquireFetchLock(fetchId)).thenReturn(true);
- assertFalse(delayedShareFetch.isCompleted());
+ 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)));
- // Since 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));
+ assertFalse(delayedShareFetch.isCompleted());
- delayedShareFetch.lock().unlock();
- Mockito.verify(exceptionHandler, times(1)).accept(any(), any());
+ // 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());
Review Comment:
`exceptionHandler` shouldn't be trigger 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
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
