clolov commented on code in PR #13999:
URL: https://github.com/apache/kafka/pull/13999#discussion_r1266720289
##########
core/src/main/java/kafka/log/remote/RemoteLogReader.java:
##########
@@ -31,14 +32,18 @@ public class RemoteLogReader implements Callable<Void> {
private final Logger logger;
private final RemoteStorageFetchInfo fetchInfo;
private final RemoteLogManager rlm;
+ private final BrokerTopicStats brokerTopicStats;
private final Consumer<RemoteLogReadResult> callback;
public RemoteLogReader(RemoteStorageFetchInfo fetchInfo,
RemoteLogManager rlm,
- Consumer<RemoteLogReadResult> callback) {
+ Consumer<RemoteLogReadResult> callback,
+ BrokerTopicStats brokerTopicStats) {
this.fetchInfo = fetchInfo;
this.rlm = rlm;
+ this.brokerTopicStats = brokerTopicStats;
this.callback = callback;
+
this.brokerTopicStats.topicStats(fetchInfo.topicPartition.topic()).remoteReadRequestRate().mark();
Review Comment:
What is the purpose of this being marked here?
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -594,13 +618,15 @@ private void copyLogSegment(UnifiedLog log, LogSegment
segment, long nextSegment
LogSegmentData segmentData = new LogSegmentData(logFile.toPath(),
toPathIfExists(segment.lazyOffsetIndex().get().file()),
toPathIfExists(segment.lazyTimeIndex().get().file()),
Optional.ofNullable(toPathIfExists(segment.txnIndex().file())),
producerStateSnapshotFile.toPath(), leaderEpochsIndex);
+
brokerTopicStats.topicStats(log.topicPartition().topic()).remoteWriteRequestRate().mark();
Review Comment:
Can you add this to the description of the pull request as well? I suspect
this is a previously missed metric, but I wasn't expecting it when I was
reviewing the pull request.
##########
core/src/main/java/kafka/log/remote/RemoteLogReader.java:
##########
@@ -31,14 +32,18 @@ public class RemoteLogReader implements Callable<Void> {
private final Logger logger;
private final RemoteStorageFetchInfo fetchInfo;
private final RemoteLogManager rlm;
+ private final BrokerTopicStats brokerTopicStats;
private final Consumer<RemoteLogReadResult> callback;
public RemoteLogReader(RemoteStorageFetchInfo fetchInfo,
RemoteLogManager rlm,
- Consumer<RemoteLogReadResult> callback) {
+ Consumer<RemoteLogReadResult> callback,
+ BrokerTopicStats brokerTopicStats) {
this.fetchInfo = fetchInfo;
this.rlm = rlm;
+ this.brokerTopicStats = brokerTopicStats;
this.callback = callback;
+
this.brokerTopicStats.topicStats(fetchInfo.topicPartition.topic()).remoteReadRequestRate().mark();
Review Comment:
Oh, I see, every time an async read is done we create a new RemoteLogReader.
Still, is it possible to do this increment as part of the call method?
--
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
