apoorvmittal10 commented on code in PR #18826:
URL: https://github.com/apache/kafka/pull/18826#discussion_r1954422684
##########
core/src/test/java/kafka/server/share/SharePartitionManagerTest.java:
##########
@@ -2172,14 +2124,88 @@ public void
testPendingInitializationShouldCompleteFetchRequest() throws Excepti
Mockito.verify(mockReplicaManager, times(0)).readFromLog(
any(), any(), any(ReplicaQuota.class), anyBoolean());
assertFalse(pendingInitializationFuture.isDone());
+ assertEquals(0, shareGroupMetrics.partitionLoadTimeMs().count());
// Complete the pending initialization future.
pendingInitializationFuture.complete(null);
+ // Verify the partition load time metrics.
+ assertEquals(1, shareGroupMetrics.partitionLoadTimeMs().count());
+ assertEquals(90.0, shareGroupMetrics.partitionLoadTimeMs().min());
+ assertEquals(90.0, shareGroupMetrics.partitionLoadTimeMs().max());
+ assertEquals(90.0, shareGroupMetrics.partitionLoadTimeMs().sum());
// Should have 1 fetch recorded.
validateBrokerTopicStatsMetrics(
brokerTopicStats,
new TopicMetrics(1, 0, 0, 0),
Map.of(tp0.topic(), new TopicMetrics(1, 0, 0, 0))
);
+ shareGroupMetrics.close();
+ }
+
+ @Test
+ public void testPartitionLoadTimeMetricWithMultiplePartitions() throws
Exception {
+ String groupId = "grp";
+ TopicIdPartition tp0 = new TopicIdPartition(Uuid.randomUuid(), new
TopicPartition("foo", 0));
+ TopicIdPartition tp1 = new TopicIdPartition(Uuid.randomUuid(), new
TopicPartition("foo", 1));
+ LinkedHashMap<TopicIdPartition, Integer> partitionMaxBytes =
orderedMap(PARTITION_MAX_BYTES, tp0, tp1);
+
+ SharePartition sp0 = mock(SharePartition.class);
+ SharePartition sp1 = mock(SharePartition.class);
+ Map<SharePartitionKey, SharePartition> partitionCacheMap = new
HashMap<>();
+ partitionCacheMap.put(new SharePartitionKey(groupId, tp0), sp0);
+ partitionCacheMap.put(new SharePartitionKey(groupId, tp1), sp1);
+
+ // Keep the initialization future pending, so fetch request is stuck.
+ CompletableFuture<Void> pendingInitializationFuture1 = new
CompletableFuture<>();
+ when(sp0.maybeInitialize()).thenReturn(pendingInitializationFuture1);
+ when(sp0.loadStartTimeMs()).thenReturn(10L);
+
+ CompletableFuture<Void> pendingInitializationFuture2 = new
CompletableFuture<>();
+ when(sp1.maybeInitialize()).thenReturn(pendingInitializationFuture2);
+ when(sp1.loadStartTimeMs()).thenReturn(40L);
+
+ DelayedOperationPurgatory<DelayedShareFetch>
delayedShareFetchPurgatory = new DelayedOperationPurgatory<>(
Review Comment:
I have added thread related fixes here in this PR and additional checks:
https://github.com/apache/kafka/pull/18862. The shutdown of
`delayedShareFetchPurgatory` will close the reaper thread which is also closed
by SharePartitionManager.
Invoking close again on timer, sometimes gives exception as `add` while
close fails. So I have tied things with SharePartitionManager close and added
check for no pending threads, so in future if some test case fails to close
things then we can catch the error early.
--
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]
