kamalcph commented on code in PR #15133:
URL: https://github.com/apache/kafka/pull/15133#discussion_r1462927795
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -1650,15 +1650,27 @@ private static void
shutdownAndAwaitTermination(ExecutorService pool, String poo
}
private void removeRemoteTopicPartitionMetrics(TopicIdPartition
topicIdPartition) {
- BrokerTopicMetrics topicMetrics =
brokerTopicStats.topicStats(topicIdPartition.topic());
- int partition = topicIdPartition.partition();
- topicMetrics.removeRemoteCopyLagBytes(partition);
- topicMetrics.removeRemoteCopyLagSegments(partition);
- topicMetrics.removeRemoteDeleteLagBytes(partition);
- topicMetrics.removeRemoteDeleteLagSegments(partition);
- topicMetrics.removeRemoteLogMetadataCount(partition);
- topicMetrics.removeRemoteLogSizeComputationTime(partition);
- topicMetrics.removeRemoteLogSizeBytes(partition);
+ String topic = topicIdPartition.topic();
+ if (!brokerTopicStats.isTopicStatsExisted(topicIdPartition.topic())) {
+ // The topic metrics are already removed, removing this topic key
from broker-level metrics
+ brokerTopicStats.removeBrokerLevelRemoteCopyLagBytes(topic);
Review Comment:
I'm not able to understand this block. Here, we are checking
`brokerTopicStats.isTopicStatsExisted(topicIdPartition.topic())` which will
always return true instead of `allTopicStats`
When the block in L1670-1677 will be executed?
--
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
