lucasbru commented on code in PR #15210:
URL: https://github.com/apache/kafka/pull/15210#discussion_r1457244991
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java:
##########
@@ -140,6 +141,7 @@ public class AsyncKafkaConsumerTest {
private final ApplicationEventHandler applicationEventHandler =
mock(ApplicationEventHandler.class);
private final ConsumerMetadata metadata = mock(ConsumerMetadata.class);
private final LinkedBlockingQueue<BackgroundEvent> backgroundEventQueue =
new LinkedBlockingQueue<>();
+ private final Metrics metrics = new Metrics();
Review Comment:
This is only accessed in one place, so not sure why you added it, when you
are accessing the metrics using `consumer.metrics()` inside the test.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java:
##########
@@ -1547,6 +1553,29 @@ public String toString() {
}
}
+ private class ConsumerCoordinatorMetrics {
Review Comment:
Why not? Why did you create separate implementations
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java:
##########
@@ -735,6 +743,23 @@ public void
testOffsetCommitFailsWithStaleEpochAndRetriesWithNewEpoch() {
assertEquals(memberId, reqData.memberId());
}
+ @Test
+ public void testEnsureCommitSensorRecordsMetric() {
+ CommitRequestManager commitRequestManager = create(true, 100);
+
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
+ Map<TopicPartition, OffsetAndMetadata> offsets =
Collections.singletonMap(
+ new TopicPartition("topic", 1),
+ new OffsetAndMetadata(0));
+ commitRequestManager.addOffsetCommitRequest(offsets, Optional.empty(),
true);
+ NetworkClientDelegate.PollResult res =
commitRequestManager.poll(time.milliseconds());
+ assertEquals(1, res.unsentRequests.size());
+
res.unsentRequests.get(0).future().complete(mockOffsetCommitResponse("topic",
1, (short) 1, Errors.NONE));
+ assertNotNull(getMetric("commit-latency-avg"));
Review Comment:
Could we please send two commits, mock the createTime and the receivedTime
and then test the metric for concrete values?
Alternatively, we could test the metric individually, similar to
`ConsumerCoordinatorTest.testMetrics`.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -371,6 +392,24 @@ public NetworkClientDelegate.PollResult
drainPendingOffsetCommitRequests() {
return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, requests);
}
+ private Sensor addCommitSensor(Metrics metrics, String metricGrpPrefix) {
+ System.out.println("hello");
Review Comment:
good day sir
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
