You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "showuon (via GitHub)" <gi...@apache.org> on 2024/01/05 08:29:59 UTC
[PR] KAFKA-16085: add all topic metrics [kafka]
showuon opened a new pull request, #15133:
URL: https://github.com/apache/kafka/pull/15133
*More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.*
*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon merged PR #15133:
URL: https://github.com/apache/kafka/pull/15133
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #15133:
URL: https://github.com/apache/kafka/pull/15133#discussion_r1463172726
##########
core/src/test/scala/unit/kafka/utils/TestUtils.scala:
##########
@@ -1396,11 +1396,13 @@ object TestUtils extends Logging {
// Note: Call this method in the test itself, rather than the @AfterEach method.
// Because of the assert, if assertNoNonDaemonThreads fails, nothing after would be executed.
def assertNoNonDaemonThreads(threadNamePrefix: String): Unit = {
- val nonDaemonThreads = Thread.getAllStackTraces.keySet.asScala.filter { t =>
- !t.isDaemon && t.isAlive && t.getName.startsWith(threadNamePrefix)
- }
- val threadCount = nonDaemonThreads.size
- assertEquals(0, threadCount, s"Found unexpected $threadCount NonDaemon threads=${nonDaemonThreads.map(t => t.getName).mkString(", ")}")
+ var nonDemonThreads: mutable.Set[Thread] = mutable.Set.empty[Thread]
+ waitUntilTrue(() => {
+ nonDemonThreads = Thread.getAllStackTraces.keySet.asScala.filter { t =>
+ !t.isDaemon && t.isAlive && t.getName.startsWith(threadNamePrefix)
+ }
+ 0 == nonDemonThreads.size
+ }, s"Found unexpected ${nonDemonThreads.size} NonDaemon threads=${nonDemonThreads.map(t => t.getName).mkString(", ")}", 1000)
Review Comment:
cc @divijvaidya , I found sometimes the [CI](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-15133/9/testReport/junit/kafka.server/ReplicaManagerTest/Build___JDK_11_and_Scala_2_13___testSuccessfulBuildRemoteLogAuxStateMetrics__/) is too sensitive to the non demean threads check. There are some shutdown are in async way. So you can check the failed results [here](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-15133/9/): Basically, if there are some resource not closed, all the following tests should also fail (I verified in my local env). But in the CI results, it only fail 2 of replicaManagertest, and only in jdk11. So I'm going to verify it using `waitUntilTrue` to give it some chance to wait for the threads shutdown.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #15133:
URL: https://github.com/apache/kafka/pull/15133#discussion_r1461839764
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -659,6 +618,132 @@ class BrokerTopicStats(configOpt: java.util.Optional[KafkaConfig] = java.util.Op
}
}
+ // Update the broker-level all topic metric values so that we have a sample right for all topics metric after update of partition.
+ def recordRemoteCopyLagBytes(topic: String, partition: Int, value: Long): Unit = {
+ val topicMetric = topicStats(topic)
+ topicMetric.remoteCopyLagBytesAggrMetric().setValue(String.valueOf(partition), value)
+ allTopicsStats.remoteCopyLagBytesAggrMetric().setValue(topic, topicMetric.remoteCopyLagBytes)
+ }
+
+ // Update the broker-level all topic metric values so that we have a sample right for all topics metric after removal of partition.
+ def removeRemoteCopyLagBytes(topic: String, partition: Int): Unit = {
+ val topicMetric = topicStats(topic)
+ topicMetric.remoteCopyLagBytesAggrMetric().removeKey(String.valueOf(partition))
+ allTopicsStats.remoteCopyLagBytesAggrMetric().setValue(topic, topicMetric.remoteCopyLagBytes)
+ }
Review Comment:
After refactor, the per-topic and per-broker metric will be updated within one method to avoid forgetting about it.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #15133:
URL: https://github.com/apache/kafka/pull/15133#discussion_r1462932786
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -406,93 +408,51 @@ class BrokerTopicMetrics(name: Option[String], configOpt: java.util.Optional[Kaf
def invalidOffsetOrSequenceRecordsPerSec: Meter = metricTypeMap.get(BrokerTopicStats.InvalidOffsetOrSequenceRecordsPerSec).meter()
- def recordRemoteCopyLagBytes(partition: Int, bytesLag: Long): Unit = {
- val brokerTopicAggregatedMetric = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_BYTES_METRIC.getName).brokerTopicAggregatedMetric
- brokerTopicAggregatedMetric.setPartitionMetricValue(partition, bytesLag)
- }
-
- def removeRemoteCopyLagBytes(partition: Int): Unit = {
- val brokerTopicAggregatedMetric = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_BYTES_METRIC.getName).brokerTopicAggregatedMetric
- brokerTopicAggregatedMetric.removePartition(partition)
+ def remoteCopyLagBytesAggrMetric(): AggregatedMetric = {
+ metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_BYTES_METRIC.getName).aggregatedMetric
}
// Visible for testing
- def remoteCopyLagBytes: Long = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_BYTES_METRIC.getName).brokerTopicAggregatedMetric.value()
-
- def recordRemoteCopyLagSegments(partition: Int, segmentsLag: Long): Unit = {
- val brokerTopicAggregatedMetric = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_SEGMENTS_METRIC.getName).brokerTopicAggregatedMetric
- brokerTopicAggregatedMetric.setPartitionMetricValue(partition, segmentsLag)
- }
+ def remoteCopyLagBytes: Long = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_BYTES_METRIC.getName).aggregatedMetric.value()
- def removeRemoteCopyLagSegments(partition: Int): Unit = {
- val brokerTopicAggregatedMetric = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_SEGMENTS_METRIC.getName).brokerTopicAggregatedMetric
- brokerTopicAggregatedMetric.removePartition(partition)
+ def remoteCopyLagSegmentsAggrMetric(): AggregatedMetric = {
+ metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_SEGMENTS_METRIC.getName).aggregatedMetric
}
// Visible for testing
- def remoteCopyLagSegments: Long = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_SEGMENTS_METRIC.getName).brokerTopicAggregatedMetric.value()
-
- def recordRemoteLogMetadataCount(partition: Int, count: Long): Unit = {
- val brokerTopicAggregatedMetric = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_METADATA_COUNT_METRIC.getName).brokerTopicAggregatedMetric
- brokerTopicAggregatedMetric.setPartitionMetricValue(partition, count)
- }
+ def remoteCopyLagSegments: Long = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_SEGMENTS_METRIC.getName).aggregatedMetric.value()
- def removeRemoteLogMetadataCount(partition: Int): Unit = {
- val brokerTopicAggregatedMetric = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_METADATA_COUNT_METRIC.getName).brokerTopicAggregatedMetric
- brokerTopicAggregatedMetric.removePartition(partition)
+ def remoteLogMetadataCountAggrMetric(): AggregatedMetric = {
+ metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_METADATA_COUNT_METRIC.getName).aggregatedMetric
}
- def remoteLogMetadataCount: Long = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_METADATA_COUNT_METRIC.getName).brokerTopicAggregatedMetric.value()
-
- def recordRemoteLogSizeBytes(partition: Int, size: Long): Unit = {
- val brokerTopicAggregatedMetric = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_SIZE_BYTES_METRIC.getName).brokerTopicAggregatedMetric
- brokerTopicAggregatedMetric.setPartitionMetricValue(partition, size)
- }
+ def remoteLogMetadataCount: Long = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_METADATA_COUNT_METRIC.getName).aggregatedMetric.value()
- def removeRemoteLogSizeBytes(partition: Int): Unit = {
- val brokerTopicAggregatedMetric = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_SIZE_BYTES_METRIC.getName).brokerTopicAggregatedMetric
- brokerTopicAggregatedMetric.removePartition(partition)
+ def remoteLogSizeBytesAggrMetric(): AggregatedMetric = {
+ metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_SIZE_BYTES_METRIC.getName).aggregatedMetric
}
- def remoteLogSizeBytes: Long = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_SIZE_BYTES_METRIC.getName).brokerTopicAggregatedMetric.value()
-
- def recordRemoteLogSizeComputationTime(partition: Int, timeSpent: Long): Unit = {
- val brokerTopicAggregatedMetric = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_SIZE_COMPUTATION_TIME_METRIC.getName).brokerTopicAggregatedMetric
- brokerTopicAggregatedMetric.setPartitionMetricValue(partition, timeSpent)
- }
+ def remoteLogSizeBytes: Long = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_SIZE_BYTES_METRIC.getName).aggregatedMetric.value()
- def removeRemoteLogSizeComputationTime(partition: Int): Unit = {
- val brokerTopicAggregatedMetric = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_SIZE_COMPUTATION_TIME_METRIC.getName).brokerTopicAggregatedMetric
- brokerTopicAggregatedMetric.removePartition(partition)
+ def remoteLogSizeComputationTimeAggrMetric(): AggregatedMetric = {
+ metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_SIZE_COMPUTATION_TIME_METRIC.getName).aggregatedMetric
}
- def remoteLogSizeComputationTime: Long = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_SIZE_COMPUTATION_TIME_METRIC.getName).brokerTopicAggregatedMetric.value()
-
- def recordRemoteDeleteLagBytes(partition: Int, segmentsLag: Long): Unit = {
- val brokerTopicAggregatedMetric = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_DELETE_LAG_BYTES_METRIC.getName).brokerTopicAggregatedMetric
- brokerTopicAggregatedMetric.setPartitionMetricValue(partition, segmentsLag)
- }
+ def remoteLogSizeComputationTime: Long = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_SIZE_COMPUTATION_TIME_METRIC.getName).aggregatedMetric.value()
Review Comment:
```suggestion
def remoteLogSizeComputationTime: Long = remoteLogSizeComputationTimeAggrMetric().value()
```
similar in other places
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on PR #15133:
URL: https://github.com/apache/kafka/pull/15133#issuecomment-1880409444
The proposed patch LGTM. Here, the "all-topic" metric value refers to per-broker level aggregate. Is this correct? Can we have a Jconsole screenshot of the metrics? I'll also build one locally and verify it:
> The metrics added in KIP-963 is BrokerTopicMetrics, which means it should provide per-topic metric value and all topics metric value. But current implementation doesn't update all topic metric value.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on PR #15133:
URL: https://github.com/apache/kafka/pull/15133#issuecomment-1899108113
> @divijvaidya , since Satish is busy, could you help review this PR? We'd like to get it into v3.7.0 for the completion of KIP-963. Thanks.
Sorry I have been busy with work lately. Will look at this first thing tomorrow.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #15133:
URL: https://github.com/apache/kafka/pull/15133#issuecomment-1886791219
@divijvaidya @kamalcph @clolov , I finally got some time to complete the PR. Please take a look when available. Thanks.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #15133:
URL: https://github.com/apache/kafka/pull/15133#issuecomment-1897627545
@divijvaidya , since Satish is busy, could you help review this PR? We'd like to get it into v3.7.0 for the completion of KIP-963. Thanks.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #15133:
URL: https://github.com/apache/kafka/pull/15133#discussion_r1461839024
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -529,20 +489,16 @@ class BrokerTopicMetrics(name: Option[String], configOpt: java.util.Optional[Kaf
}
}
-class BrokerTopicAggregatedMetric() {
- private val partitionMetricValues = new ConcurrentHashMap[Int, Long]()
-
- def setPartitionMetricValue(partition: Int, partitionValue: Long): Unit = {
- partitionMetricValues.put(partition, partitionValue)
- }
-
- def removePartition(partition: Int): Option[Long] = {
- Option.apply(partitionMetricValues.remove(partition))
- }
-
- def value(): Long = partitionMetricValues.values().stream().mapToLong(v => v).sum()
-
- def close(): Unit = partitionMetricValues.clear()
+class AggregatedMetric {
+ // The map to store:
+ // - per-partition value for topic-level metrics. The key will be the partition number
+ // - per-topic value for broker-level metrics. The key will be the topic name
+ private val metricValues = new ConcurrentHashMap[String, Long]()
+ def setValue(key: String, value: Long): Unit = metricValues.put(key, value)
+ def removeKey(key: String): Option[Long] = Option.apply(metricValues.remove(key))
+ // Sum all values in the metricValues map
+ def value(): Long = metricValues.values().stream().mapToLong(v => v).sum()
+ def close(): Unit = metricValues.clear()
}
Review Comment:
After the latest refactor, we will use the `AggregatedMetric` for per-topic and per-broker metrics, which makes it simpler.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on code in PR #15133:
URL: https://github.com/apache/kafka/pull/15133#discussion_r1458819718
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -529,20 +564,33 @@ class BrokerTopicMetrics(name: Option[String], configOpt: java.util.Optional[Kaf
}
}
-class BrokerTopicAggregatedMetric() {
+class BrokerTopicAggregatedMetric(allTopics: Boolean = false) {
Review Comment:
This class is specific to aggregated metrics on a topic. We are hacking it to provide broker level metrics. Maybe a BrokerAggregatedMetric is required?
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -787,9 +787,14 @@ private void copyLogSegment(UnifiedLog log, LogSegment segment, long nextSegment
long bytesLag = log.onlyLocalLogSegmentsSize() - log.activeSegment().size();
String topic = topicIdPartition.topic();
int partition = topicIdPartition.partition();
- brokerTopicStats.topicStats(topic).recordRemoteCopyLagBytes(partition, bytesLag);
+
+ BrokerTopicMetrics brokerTopicMetrics = brokerTopicStats.topicStats(topic);
+ brokerTopicMetrics.recordRemoteCopyLagBytes(partition, bytesLag);
+ brokerTopicStats.allTopicsStats().recordRemoteCopyLagBytes(topic, brokerTopicMetrics.remoteCopyLagBytes());
Review Comment:
nit
perhaps worth adding a comment on why we are using `brokerTopicMetrics.remoteCopyLagBytes()` instead simply updating with `bytesLag` here. I think the answer is that we want to update the "topic" level metric with an aggregate value across partitions and `bytesLag` just represents this partition.
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -1651,14 +1665,25 @@ private static void shutdownAndAwaitTermination(ExecutorService pool, String poo
private void removeRemoteTopicPartitionMetrics(TopicIdPartition topicIdPartition) {
BrokerTopicMetrics topicMetrics = brokerTopicStats.topicStats(topicIdPartition.topic());
+ String topic = topicIdPartition.topic();
int partition = topicIdPartition.partition();
+ // remove the partition metric values
topicMetrics.removeRemoteCopyLagBytes(partition);
topicMetrics.removeRemoteCopyLagSegments(partition);
topicMetrics.removeRemoteDeleteLagBytes(partition);
topicMetrics.removeRemoteDeleteLagSegments(partition);
topicMetrics.removeRemoteLogMetadataCount(partition);
topicMetrics.removeRemoteLogSizeComputationTime(partition);
topicMetrics.removeRemoteLogSizeBytes(partition);
+
+ // update the all topic metric values
Review Comment:
suggestion
change comment to:
Update the all topic metric values so that we have a sample right for all topics metric after removal of partition. In absence of this update, the all topic metric will not update until the next sample point is generated for it.
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -1175,7 +1186,10 @@ private Optional<RetentionSizeData> buildRetentionSizeData(long retentionSize,
}
}
}
- brokerTopicStats.topicStats(topicIdPartition.topic()).recordRemoteLogSizeComputationTime(topicIdPartition.partition(), time.milliseconds() - startTimeMs);
+
+ BrokerTopicMetrics brokerTopicMetrics = brokerTopicStats.topicStats(topicIdPartition.topic());
+ brokerTopicMetrics.recordRemoteLogSizeComputationTime(topicIdPartition.partition(), time.milliseconds() - startTimeMs);
+ brokerTopicStats.allTopicsStats().recordRemoteLogSizeComputationTime(topicIdPartition.topic(), brokerTopicMetrics.remoteLogSizeComputationTime());
Review Comment:
could go in a single function RequestHandler with name updateRemoteLogSizeComputationTime(), similar to existing updateBytesOut(). This will ensure that in future folks don't forget to update allTopics if they update the broekrTopic metric.
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -529,20 +564,33 @@ class BrokerTopicMetrics(name: Option[String], configOpt: java.util.Optional[Kaf
}
}
-class BrokerTopicAggregatedMetric() {
+class BrokerTopicAggregatedMetric(allTopics: Boolean = false) {
private val partitionMetricValues = new ConcurrentHashMap[Int, Long]()
+ private val topicMetricValues = new ConcurrentHashMap[String, Long]()
def setPartitionMetricValue(partition: Int, partitionValue: Long): Unit = {
partitionMetricValues.put(partition, partitionValue)
}
+ def setTopicMetricValue(topic: String, topicValue: Long): Unit = {
Review Comment:
do we want to remove the topic metric after the topic is deleted?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #15133:
URL: https://github.com/apache/kafka/pull/15133#discussion_r1463136420
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -406,93 +408,51 @@ class BrokerTopicMetrics(name: Option[String], configOpt: java.util.Optional[Kaf
def invalidOffsetOrSequenceRecordsPerSec: Meter = metricTypeMap.get(BrokerTopicStats.InvalidOffsetOrSequenceRecordsPerSec).meter()
- def recordRemoteCopyLagBytes(partition: Int, bytesLag: Long): Unit = {
- val brokerTopicAggregatedMetric = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_BYTES_METRIC.getName).brokerTopicAggregatedMetric
- brokerTopicAggregatedMetric.setPartitionMetricValue(partition, bytesLag)
- }
-
- def removeRemoteCopyLagBytes(partition: Int): Unit = {
- val brokerTopicAggregatedMetric = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_BYTES_METRIC.getName).brokerTopicAggregatedMetric
- brokerTopicAggregatedMetric.removePartition(partition)
+ def remoteCopyLagBytesAggrMetric(): AggregatedMetric = {
+ metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_BYTES_METRIC.getName).aggregatedMetric
}
// Visible for testing
- def remoteCopyLagBytes: Long = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_BYTES_METRIC.getName).brokerTopicAggregatedMetric.value()
-
- def recordRemoteCopyLagSegments(partition: Int, segmentsLag: Long): Unit = {
- val brokerTopicAggregatedMetric = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_SEGMENTS_METRIC.getName).brokerTopicAggregatedMetric
- brokerTopicAggregatedMetric.setPartitionMetricValue(partition, segmentsLag)
- }
+ def remoteCopyLagBytes: Long = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_BYTES_METRIC.getName).aggregatedMetric.value()
- def removeRemoteCopyLagSegments(partition: Int): Unit = {
- val brokerTopicAggregatedMetric = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_SEGMENTS_METRIC.getName).brokerTopicAggregatedMetric
- brokerTopicAggregatedMetric.removePartition(partition)
+ def remoteCopyLagSegmentsAggrMetric(): AggregatedMetric = {
+ metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_SEGMENTS_METRIC.getName).aggregatedMetric
}
// Visible for testing
- def remoteCopyLagSegments: Long = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_SEGMENTS_METRIC.getName).brokerTopicAggregatedMetric.value()
-
- def recordRemoteLogMetadataCount(partition: Int, count: Long): Unit = {
- val brokerTopicAggregatedMetric = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_METADATA_COUNT_METRIC.getName).brokerTopicAggregatedMetric
- brokerTopicAggregatedMetric.setPartitionMetricValue(partition, count)
- }
+ def remoteCopyLagSegments: Long = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_SEGMENTS_METRIC.getName).aggregatedMetric.value()
- def removeRemoteLogMetadataCount(partition: Int): Unit = {
- val brokerTopicAggregatedMetric = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_METADATA_COUNT_METRIC.getName).brokerTopicAggregatedMetric
- brokerTopicAggregatedMetric.removePartition(partition)
+ def remoteLogMetadataCountAggrMetric(): AggregatedMetric = {
+ metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_METADATA_COUNT_METRIC.getName).aggregatedMetric
}
- def remoteLogMetadataCount: Long = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_METADATA_COUNT_METRIC.getName).brokerTopicAggregatedMetric.value()
-
- def recordRemoteLogSizeBytes(partition: Int, size: Long): Unit = {
- val brokerTopicAggregatedMetric = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_SIZE_BYTES_METRIC.getName).brokerTopicAggregatedMetric
- brokerTopicAggregatedMetric.setPartitionMetricValue(partition, size)
- }
+ def remoteLogMetadataCount: Long = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_METADATA_COUNT_METRIC.getName).aggregatedMetric.value()
- def removeRemoteLogSizeBytes(partition: Int): Unit = {
- val brokerTopicAggregatedMetric = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_SIZE_BYTES_METRIC.getName).brokerTopicAggregatedMetric
- brokerTopicAggregatedMetric.removePartition(partition)
+ def remoteLogSizeBytesAggrMetric(): AggregatedMetric = {
+ metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_SIZE_BYTES_METRIC.getName).aggregatedMetric
}
- def remoteLogSizeBytes: Long = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_SIZE_BYTES_METRIC.getName).brokerTopicAggregatedMetric.value()
-
- def recordRemoteLogSizeComputationTime(partition: Int, timeSpent: Long): Unit = {
- val brokerTopicAggregatedMetric = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_SIZE_COMPUTATION_TIME_METRIC.getName).brokerTopicAggregatedMetric
- brokerTopicAggregatedMetric.setPartitionMetricValue(partition, timeSpent)
- }
+ def remoteLogSizeBytes: Long = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_SIZE_BYTES_METRIC.getName).aggregatedMetric.value()
- def removeRemoteLogSizeComputationTime(partition: Int): Unit = {
- val brokerTopicAggregatedMetric = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_SIZE_COMPUTATION_TIME_METRIC.getName).brokerTopicAggregatedMetric
- brokerTopicAggregatedMetric.removePartition(partition)
+ def remoteLogSizeComputationTimeAggrMetric(): AggregatedMetric = {
+ metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_SIZE_COMPUTATION_TIME_METRIC.getName).aggregatedMetric
}
- def remoteLogSizeComputationTime: Long = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_SIZE_COMPUTATION_TIME_METRIC.getName).brokerTopicAggregatedMetric.value()
-
- def recordRemoteDeleteLagBytes(partition: Int, segmentsLag: Long): Unit = {
- val brokerTopicAggregatedMetric = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_DELETE_LAG_BYTES_METRIC.getName).brokerTopicAggregatedMetric
- brokerTopicAggregatedMetric.setPartitionMetricValue(partition, segmentsLag)
- }
+ def remoteLogSizeComputationTime: Long = metricGaugeTypeMap.get(RemoteStorageMetrics.REMOTE_LOG_SIZE_COMPUTATION_TIME_METRIC.getName).aggregatedMetric.value()
Review Comment:
Ah, nice! Let me update it soon.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #15133:
URL: https://github.com/apache/kafka/pull/15133#discussion_r1461840656
##########
core/src/test/scala/kafka/server/KafkaRequestHandlerTest.scala:
##########
@@ -245,403 +251,438 @@ class KafkaRequestHandlerTest {
@ParameterizedTest
@ValueSource(booleans = Array(true, false))
def testSingularCopyLagBytesMetric(systemRemoteStorageEnabled: Boolean): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics(systemRemoteStorageEnabled)
+ val props = kafka.utils.TestUtils.createDummyBrokerConfig()
+ props.setProperty(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, systemRemoteStorageEnabled.toString)
+ val brokerTopicStats = new BrokerTopicStats(java.util.Optional.of(KafkaConfig.fromProps(props)))
+ val brokerTopicMetrics = brokerTopicStats.topicStats(topic)
if (systemRemoteStorageEnabled) {
- brokerTopicMetrics.recordRemoteCopyLagBytes(0, 100);
- brokerTopicMetrics.recordRemoteCopyLagBytes(1, 150);
- brokerTopicMetrics.recordRemoteCopyLagBytes(2, 250);
+ brokerTopicStats.recordRemoteCopyLagBytes(topic, 0, 100)
+ brokerTopicStats.recordRemoteCopyLagBytes(topic, 1, 150)
+ brokerTopicStats.recordRemoteCopyLagBytes(topic, 2, 250)
assertEquals(500, brokerTopicMetrics.remoteCopyLagBytes)
+ assertEquals(500, brokerTopicStats.allTopicsStats.remoteCopyLagBytes)
+ brokerTopicStats.recordRemoteCopyLagBytes(topic2, 0, 100)
+ assertEquals(600, brokerTopicStats.allTopicsStats.remoteCopyLagBytes)
} else {
assertEquals(None, brokerTopicMetrics.metricGaugeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_BYTES_METRIC.getName))
+ assertEquals(None, brokerTopicStats.allTopicsStats.metricGaugeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_BYTES_METRIC.getName))
}
Review Comment:
Update the `KafkaRequestHandlerTest` to add allTopic metrics verification.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on code in PR #15133:
URL: https://github.com/apache/kafka/pull/15133#discussion_r1461999960
##########
core/src/test/scala/kafka/server/KafkaRequestHandlerTest.scala:
##########
@@ -245,403 +251,438 @@ class KafkaRequestHandlerTest {
@ParameterizedTest
@ValueSource(booleans = Array(true, false))
def testSingularCopyLagBytesMetric(systemRemoteStorageEnabled: Boolean): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics(systemRemoteStorageEnabled)
+ val props = kafka.utils.TestUtils.createDummyBrokerConfig()
+ props.setProperty(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, systemRemoteStorageEnabled.toString)
+ val brokerTopicStats = new BrokerTopicStats(java.util.Optional.of(KafkaConfig.fromProps(props)))
+ val brokerTopicMetrics = brokerTopicStats.topicStats(topic)
if (systemRemoteStorageEnabled) {
- brokerTopicMetrics.recordRemoteCopyLagBytes(0, 100);
- brokerTopicMetrics.recordRemoteCopyLagBytes(1, 150);
- brokerTopicMetrics.recordRemoteCopyLagBytes(2, 250);
+ brokerTopicStats.recordRemoteCopyLagBytes(topic, 0, 100)
+ brokerTopicStats.recordRemoteCopyLagBytes(topic, 1, 150)
+ brokerTopicStats.recordRemoteCopyLagBytes(topic, 2, 250)
assertEquals(500, brokerTopicMetrics.remoteCopyLagBytes)
+ assertEquals(500, brokerTopicStats.allTopicsStats.remoteCopyLagBytes)
+ brokerTopicStats.recordRemoteCopyLagBytes(topic2, 0, 100)
+ assertEquals(600, brokerTopicStats.allTopicsStats.remoteCopyLagBytes)
} else {
assertEquals(None, brokerTopicMetrics.metricGaugeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_BYTES_METRIC.getName))
+ assertEquals(None, brokerTopicStats.allTopicsStats.metricGaugeMap.get(RemoteStorageMetrics.REMOTE_COPY_LAG_BYTES_METRIC.getName))
}
}
@Test
def testMultipleCopyLagBytesMetrics(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
-
- brokerTopicMetrics.recordRemoteCopyLagBytes(0, 1);
- brokerTopicMetrics.recordRemoteCopyLagBytes(1, 2);
- brokerTopicMetrics.recordRemoteCopyLagBytes(2, 3);
+ brokerTopicStats.recordRemoteCopyLagBytes(topic, 0, 1)
+ brokerTopicStats.recordRemoteCopyLagBytes(topic, 1, 2)
+ brokerTopicStats.recordRemoteCopyLagBytes(topic, 2, 3)
- brokerTopicMetrics.recordRemoteCopyLagBytes(0, 4);
- brokerTopicMetrics.recordRemoteCopyLagBytes(1, 5);
- brokerTopicMetrics.recordRemoteCopyLagBytes(2, 6);
+ brokerTopicStats.recordRemoteCopyLagBytes(topic, 0, 4)
+ brokerTopicStats.recordRemoteCopyLagBytes(topic, 1, 5)
+ brokerTopicStats.recordRemoteCopyLagBytes(topic, 2, 6)
assertEquals(15, brokerTopicMetrics.remoteCopyLagBytes)
+ assertEquals(15, allTopicMetrics.remoteCopyLagBytes)
+ brokerTopicStats.recordRemoteCopyLagBytes(topic2, 2, 5)
+ assertEquals(20, allTopicMetrics.remoteCopyLagBytes)
}
@Test
def testCopyLagBytesMetricWithPartitionExpansion(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
-
- brokerTopicMetrics.recordRemoteCopyLagBytes(0, 1);
- brokerTopicMetrics.recordRemoteCopyLagBytes(1, 2);
+ brokerTopicStats.recordRemoteCopyLagBytes(topic, 0, 1)
+ brokerTopicStats.recordRemoteCopyLagBytes(topic, 1, 2)
assertEquals(3, brokerTopicMetrics.remoteCopyLagBytes)
+ assertEquals(3, allTopicMetrics.remoteCopyLagBytes)
- brokerTopicMetrics.recordRemoteCopyLagBytes(2, 3);
+ brokerTopicStats.recordRemoteCopyLagBytes(topic, 2, 3)
assertEquals(6, brokerTopicMetrics.remoteCopyLagBytes)
+ assertEquals(6, allTopicMetrics.remoteCopyLagBytes)
+ brokerTopicStats.recordRemoteCopyLagBytes(topic2, 0, 1)
+ assertEquals(7, allTopicMetrics.remoteCopyLagBytes)
}
@Test
def testCopyLagBytesMetricWithPartitionShrinking(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
-
- brokerTopicMetrics.recordRemoteCopyLagBytes(0, 1);
- brokerTopicMetrics.recordRemoteCopyLagBytes(1, 2);
+ brokerTopicStats.recordRemoteCopyLagBytes(topic, 0, 1)
+ brokerTopicStats.recordRemoteCopyLagBytes(topic, 1, 2)
assertEquals(3, brokerTopicMetrics.remoteCopyLagBytes)
+ assertEquals(3, allTopicMetrics.remoteCopyLagBytes)
- brokerTopicMetrics.removeRemoteCopyLagBytes(1);
+ brokerTopicStats.removeRemoteCopyLagBytes(topic, 1)
assertEquals(1, brokerTopicMetrics.remoteCopyLagBytes)
+ assertEquals(1, allTopicMetrics.remoteCopyLagBytes)
+
+ brokerTopicStats.recordRemoteCopyLagBytes(topic2, 0, 1)
+ assertEquals(2, allTopicMetrics.remoteCopyLagBytes)
}
@Test
def testCopyLagBytesMetricWithRemovingNonexistentPartitions(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
-
- brokerTopicMetrics.recordRemoteCopyLagBytes(0, 1);
- brokerTopicMetrics.recordRemoteCopyLagBytes(1, 2);
+ brokerTopicStats.recordRemoteCopyLagBytes(topic, 0, 1)
+ brokerTopicStats.recordRemoteCopyLagBytes(topic, 1, 2)
assertEquals(3, brokerTopicMetrics.remoteCopyLagBytes)
+ assertEquals(3, allTopicMetrics.remoteCopyLagBytes)
- brokerTopicMetrics.removeRemoteCopyLagBytes(3);
+
+ brokerTopicStats.removeRemoteCopyLagBytes(topic, 3)
assertEquals(3, brokerTopicMetrics.remoteCopyLagBytes)
+ assertEquals(3, allTopicMetrics.remoteCopyLagBytes)
}
@Test
def testCopyLagBytesMetricClear(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
-
- brokerTopicMetrics.recordRemoteCopyLagBytes(0, 1);
- brokerTopicMetrics.recordRemoteCopyLagBytes(1, 2);
+ brokerTopicStats.recordRemoteCopyLagBytes(topic, 0, 1)
+ brokerTopicStats.recordRemoteCopyLagBytes(topic, 1, 2)
assertEquals(3, brokerTopicMetrics.remoteCopyLagBytes)
+ assertEquals(3, allTopicMetrics.remoteCopyLagBytes)
- brokerTopicMetrics.close()
+ brokerTopicStats.close()
assertEquals(0, brokerTopicMetrics.remoteCopyLagBytes)
+ assertEquals(0, allTopicMetrics.remoteCopyLagBytes)
+
+ brokerTopicStats.recordRemoteCopyLagBytes(topic2, 0, 1)
+ assertEquals(1, allTopicMetrics.remoteCopyLagBytes)
}
@Test
def testMultipleCopyLagSegmentsMetrics(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
-
- brokerTopicMetrics.recordRemoteCopyLagSegments(0, 1);
- brokerTopicMetrics.recordRemoteCopyLagSegments(1, 2);
- brokerTopicMetrics.recordRemoteCopyLagSegments(2, 3);
+ brokerTopicStats.recordRemoteCopyLagSegments(topic, 0, 1)
+ brokerTopicStats.recordRemoteCopyLagSegments(topic, 1, 2)
+ brokerTopicStats.recordRemoteCopyLagSegments(topic, 2, 3)
- brokerTopicMetrics.recordRemoteCopyLagSegments(0, 4);
- brokerTopicMetrics.recordRemoteCopyLagSegments(1, 5);
- brokerTopicMetrics.recordRemoteCopyLagSegments(2, 6);
+ brokerTopicStats.recordRemoteCopyLagSegments(topic, 0, 4)
+ brokerTopicStats.recordRemoteCopyLagSegments(topic, 1, 5)
+ brokerTopicStats.recordRemoteCopyLagSegments(topic, 2, 6)
assertEquals(15, brokerTopicMetrics.remoteCopyLagSegments)
+ assertEquals(15, allTopicMetrics.remoteCopyLagSegments)
+
+ brokerTopicStats.recordRemoteCopyLagSegments(topic2, 0, 1)
+ assertEquals(16, allTopicMetrics.remoteCopyLagSegments)
}
@Test
def testCopyLagSegmentsMetricWithPartitionExpansion(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
-
- brokerTopicMetrics.recordRemoteCopyLagSegments(0, 1);
- brokerTopicMetrics.recordRemoteCopyLagSegments(1, 2);
+ brokerTopicStats.recordRemoteCopyLagSegments(topic, 0, 1)
+ brokerTopicStats.recordRemoteCopyLagSegments(topic, 1, 2)
assertEquals(3, brokerTopicMetrics.remoteCopyLagSegments)
+ assertEquals(3, allTopicMetrics.remoteCopyLagSegments)
- brokerTopicMetrics.recordRemoteCopyLagSegments(2, 3);
+ brokerTopicStats.recordRemoteCopyLagSegments(topic, 2, 3)
assertEquals(6, brokerTopicMetrics.remoteCopyLagSegments)
+ assertEquals(6, allTopicMetrics.remoteCopyLagSegments)
}
@Test
def testCopyLagSegmentsMetricWithPartitionShrinking(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
-
- brokerTopicMetrics.recordRemoteCopyLagSegments(0, 1);
- brokerTopicMetrics.recordRemoteCopyLagSegments(1, 2);
+ brokerTopicStats.recordRemoteCopyLagSegments(topic, 0, 1)
+ brokerTopicStats.recordRemoteCopyLagSegments(topic, 1, 2)
assertEquals(3, brokerTopicMetrics.remoteCopyLagSegments)
+ assertEquals(3, allTopicMetrics.remoteCopyLagSegments)
- brokerTopicMetrics.removeRemoteCopyLagSegments(1);
+ brokerTopicStats.removeRemoteCopyLagSegments(topic, 1)
assertEquals(1, brokerTopicMetrics.remoteCopyLagSegments)
+ assertEquals(1, allTopicMetrics.remoteCopyLagSegments)
}
@Test
def testCopyLagSegmentsMetricWithRemovingNonexistentPartitions(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
-
- brokerTopicMetrics.recordRemoteCopyLagSegments(0, 1);
- brokerTopicMetrics.recordRemoteCopyLagSegments(1, 2);
+ brokerTopicStats.recordRemoteCopyLagSegments(topic, 0, 1)
+ brokerTopicStats.recordRemoteCopyLagSegments(topic, 1, 2)
assertEquals(3, brokerTopicMetrics.remoteCopyLagSegments)
+ assertEquals(3, allTopicMetrics.remoteCopyLagSegments)
- brokerTopicMetrics.removeRemoteCopyLagSegments(3);
+ brokerTopicStats.removeRemoteCopyLagSegments(topic, 3)
assertEquals(3, brokerTopicMetrics.remoteCopyLagSegments)
+ assertEquals(3, allTopicMetrics.remoteCopyLagSegments)
}
@Test
def testCopyLagSegmentsMetricClear(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
-
- brokerTopicMetrics.recordRemoteCopyLagSegments(0, 1);
- brokerTopicMetrics.recordRemoteCopyLagSegments(1, 2);
+ brokerTopicStats.recordRemoteCopyLagSegments(topic, 0, 1)
+ brokerTopicStats.recordRemoteCopyLagSegments(topic, 1, 2)
assertEquals(3, brokerTopicMetrics.remoteCopyLagSegments)
+ assertEquals(3, allTopicMetrics.remoteCopyLagSegments)
- brokerTopicMetrics.close()
+ brokerTopicStats.close()
assertEquals(0, brokerTopicMetrics.remoteCopyLagSegments)
+ assertEquals(0, allTopicMetrics.remoteCopyLagSegments)
+
}
@Test
def testMultipleDeleteLagBytesMetrics(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
+ brokerTopicStats.recordRemoteDeleteLagBytes(topic, 0, 1)
+ brokerTopicStats.recordRemoteDeleteLagBytes(topic, 1, 2)
+ brokerTopicStats.recordRemoteDeleteLagBytes(topic, 2, 3)
- brokerTopicMetrics.recordRemoteDeleteLagBytes(0, 1);
- brokerTopicMetrics.recordRemoteDeleteLagBytes(1, 2);
- brokerTopicMetrics.recordRemoteDeleteLagBytes(2, 3);
-
- brokerTopicMetrics.recordRemoteDeleteLagBytes(0, 4);
- brokerTopicMetrics.recordRemoteDeleteLagBytes(1, 5);
- brokerTopicMetrics.recordRemoteDeleteLagBytes(2, 6);
+ brokerTopicStats.recordRemoteDeleteLagBytes(topic, 0, 4)
+ brokerTopicStats.recordRemoteDeleteLagBytes(topic, 1, 5)
+ brokerTopicStats.recordRemoteDeleteLagBytes(topic, 2, 6)
assertEquals(15, brokerTopicMetrics.remoteDeleteLagBytes)
+ assertEquals(15, allTopicMetrics.remoteDeleteLagBytes)
+
+ brokerTopicStats.recordRemoteDeleteLagBytes(topic2, 0, 1)
+ assertEquals(16, allTopicMetrics.remoteDeleteLagBytes)
}
@Test
def testDeleteLagBytesMetricWithPartitionExpansion(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
-
- brokerTopicMetrics.recordRemoteDeleteLagBytes(0, 1);
- brokerTopicMetrics.recordRemoteDeleteLagBytes(1, 2);
+ brokerTopicStats.recordRemoteDeleteLagBytes(topic, 0, 1)
+ brokerTopicStats.recordRemoteDeleteLagBytes(topic, 1, 2)
assertEquals(3, brokerTopicMetrics.remoteDeleteLagBytes)
+ assertEquals(3, allTopicMetrics.remoteDeleteLagBytes)
- brokerTopicMetrics.recordRemoteDeleteLagBytes(2, 3);
+ brokerTopicStats.recordRemoteDeleteLagBytes(topic, 2, 3)
assertEquals(6, brokerTopicMetrics.remoteDeleteLagBytes)
+ assertEquals(6, allTopicMetrics.remoteDeleteLagBytes)
}
@Test
def testDeleteLagBytesMetricWithPartitionShrinking(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
-
- brokerTopicMetrics.recordRemoteDeleteLagBytes(0, 1);
- brokerTopicMetrics.recordRemoteDeleteLagBytes(1, 2);
+ brokerTopicStats.recordRemoteDeleteLagBytes(topic, 0, 1)
+ brokerTopicStats.recordRemoteDeleteLagBytes(topic, 1, 2)
assertEquals(3, brokerTopicMetrics.remoteDeleteLagBytes)
+ assertEquals(3, allTopicMetrics.remoteDeleteLagBytes)
- brokerTopicMetrics.removeRemoteDeleteLagBytes(1);
+ brokerTopicStats.removeRemoteDeleteLagBytes(topic, 1)
assertEquals(1, brokerTopicMetrics.remoteDeleteLagBytes)
+ assertEquals(1, allTopicMetrics.remoteDeleteLagBytes)
}
@Test
def testDeleteLagBytesMetricWithRemovingNonexistentPartitions(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
-
- brokerTopicMetrics.recordRemoteDeleteLagBytes(0, 1);
- brokerTopicMetrics.recordRemoteDeleteLagBytes(1, 2);
+ brokerTopicStats.recordRemoteDeleteLagBytes(topic, 0, 1)
+ brokerTopicStats.recordRemoteDeleteLagBytes(topic, 1, 2)
assertEquals(3, brokerTopicMetrics.remoteDeleteLagBytes)
+ assertEquals(3, allTopicMetrics.remoteDeleteLagBytes)
- brokerTopicMetrics.removeRemoteDeleteLagBytes(3);
+ brokerTopicStats.removeRemoteDeleteLagBytes(topic, 3)
assertEquals(3, brokerTopicMetrics.remoteDeleteLagBytes)
+ assertEquals(3, allTopicMetrics.remoteDeleteLagBytes)
}
@Test
def testDeleteLagBytesMetricClear(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
-
- brokerTopicMetrics.recordRemoteDeleteLagBytes(0, 1);
- brokerTopicMetrics.recordRemoteDeleteLagBytes(1, 2);
+ brokerTopicStats.recordRemoteDeleteLagBytes(topic, 0, 1)
+ brokerTopicStats.recordRemoteDeleteLagBytes(topic, 1, 2)
assertEquals(3, brokerTopicMetrics.remoteDeleteLagBytes)
+ assertEquals(3, allTopicMetrics.remoteDeleteLagBytes)
- brokerTopicMetrics.close()
+ brokerTopicStats.close()
assertEquals(0, brokerTopicMetrics.remoteDeleteLagBytes)
+ assertEquals(0, allTopicMetrics.remoteDeleteLagBytes)
}
@Test
def testMultipleDeleteLagSegmentsMetrics(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
-
- brokerTopicMetrics.recordRemoteDeleteLagSegments(0, 1);
- brokerTopicMetrics.recordRemoteDeleteLagSegments(1, 2);
- brokerTopicMetrics.recordRemoteDeleteLagSegments(2, 3);
+ brokerTopicStats.recordRemoteDeleteLagSegments(topic, 0, 1)
+ brokerTopicStats.recordRemoteDeleteLagSegments(topic, 1, 2)
+ brokerTopicStats.recordRemoteDeleteLagSegments(topic, 2, 3)
- brokerTopicMetrics.recordRemoteDeleteLagSegments(0, 4);
- brokerTopicMetrics.recordRemoteDeleteLagSegments(1, 5);
- brokerTopicMetrics.recordRemoteDeleteLagSegments(2, 6);
+ brokerTopicStats.recordRemoteDeleteLagSegments(topic, 0, 4)
+ brokerTopicStats.recordRemoteDeleteLagSegments(topic, 1, 5)
+ brokerTopicStats.recordRemoteDeleteLagSegments(topic, 2, 6)
assertEquals(15, brokerTopicMetrics.remoteDeleteLagSegments)
+ assertEquals(15, allTopicMetrics.remoteDeleteLagSegments)
+
+ brokerTopicStats.recordRemoteDeleteLagSegments(topic2, 1, 5)
+ assertEquals(20, allTopicMetrics.remoteDeleteLagSegments)
}
@Test
def testDeleteLagSegmentsMetricWithPartitionExpansion(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
-
- brokerTopicMetrics.recordRemoteDeleteLagSegments(0, 1);
- brokerTopicMetrics.recordRemoteDeleteLagSegments(1, 2);
+ brokerTopicStats.recordRemoteDeleteLagSegments(topic, 0, 1)
+ brokerTopicStats.recordRemoteDeleteLagSegments(topic, 1, 2)
assertEquals(3, brokerTopicMetrics.remoteDeleteLagSegments)
+ assertEquals(3, allTopicMetrics.remoteDeleteLagSegments)
- brokerTopicMetrics.recordRemoteDeleteLagSegments(2, 3);
+ brokerTopicStats.recordRemoteDeleteLagSegments(topic, 2, 3)
assertEquals(6, brokerTopicMetrics.remoteDeleteLagSegments)
+ assertEquals(6, allTopicMetrics.remoteDeleteLagSegments)
}
@Test
def testDeleteLagSegmentsMetricWithPartitionShrinking(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
-
- brokerTopicMetrics.recordRemoteDeleteLagSegments(0, 1);
- brokerTopicMetrics.recordRemoteDeleteLagSegments(1, 2);
+ brokerTopicStats.recordRemoteDeleteLagSegments(topic, 0, 1)
+ brokerTopicStats.recordRemoteDeleteLagSegments(topic, 1, 2)
assertEquals(3, brokerTopicMetrics.remoteDeleteLagSegments)
+ assertEquals(3, allTopicMetrics.remoteDeleteLagSegments)
- brokerTopicMetrics.removeRemoteDeleteLagSegments(1);
+ brokerTopicStats.removeRemoteDeleteLagSegments(topic, 1)
assertEquals(1, brokerTopicMetrics.remoteDeleteLagSegments)
+ assertEquals(1, allTopicMetrics.remoteDeleteLagSegments)
}
@Test
def testDeleteLagSegmentsMetricWithRemovingNonexistentPartitions(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
-
- brokerTopicMetrics.recordRemoteDeleteLagSegments(0, 1);
- brokerTopicMetrics.recordRemoteDeleteLagSegments(1, 2);
+ brokerTopicStats.recordRemoteDeleteLagSegments(topic, 0, 1)
+ brokerTopicStats.recordRemoteDeleteLagSegments(topic, 1, 2)
assertEquals(3, brokerTopicMetrics.remoteDeleteLagSegments)
+ assertEquals(3, allTopicMetrics.remoteDeleteLagSegments)
- brokerTopicMetrics.removeRemoteDeleteLagSegments(3);
+ brokerTopicStats.removeRemoteDeleteLagSegments(topic, 3)
assertEquals(3, brokerTopicMetrics.remoteDeleteLagSegments)
+ assertEquals(3, allTopicMetrics.remoteDeleteLagSegments)
}
@Test
def testDeleteLagSegmentsMetricClear(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
-
- brokerTopicMetrics.recordRemoteDeleteLagSegments(0, 1);
- brokerTopicMetrics.recordRemoteDeleteLagSegments(1, 2);
+ brokerTopicStats.recordRemoteDeleteLagSegments(topic, 0, 1)
+ brokerTopicStats.recordRemoteDeleteLagSegments(topic, 1, 2)
assertEquals(3, brokerTopicMetrics.remoteDeleteLagSegments)
+ assertEquals(3, allTopicMetrics.remoteDeleteLagSegments)
- brokerTopicMetrics.close()
+ brokerTopicStats.close()
assertEquals(0, brokerTopicMetrics.remoteDeleteLagSegments)
+ assertEquals(0, allTopicMetrics.remoteDeleteLagSegments)
}
@Test
def testRemoteLogMetadataCount(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
-
assertEquals(0, brokerTopicMetrics.remoteLogMetadataCount)
- brokerTopicMetrics.recordRemoteLogMetadataCount(0, 1)
+ assertEquals(0, allTopicMetrics.remoteLogMetadataCount)
+ brokerTopicStats.recordRemoteLogMetadataCount(topic, 0, 1)
assertEquals(1, brokerTopicMetrics.remoteLogMetadataCount)
+ assertEquals(1, allTopicMetrics.remoteLogMetadataCount)
- brokerTopicMetrics.recordRemoteLogMetadataCount(1, 2)
- brokerTopicMetrics.recordRemoteLogMetadataCount(2, 3)
+ brokerTopicStats.recordRemoteLogMetadataCount(topic, 1, 2)
+ brokerTopicStats.recordRemoteLogMetadataCount(topic, 2, 3)
assertEquals(6, brokerTopicMetrics.remoteLogMetadataCount)
+ assertEquals(6, allTopicMetrics.remoteLogMetadataCount)
- brokerTopicMetrics.close()
+ brokerTopicStats.close()
assertEquals(0, brokerTopicMetrics.remoteLogMetadataCount)
+ assertEquals(0, allTopicMetrics.remoteLogMetadataCount)
}
@ParameterizedTest
@ValueSource(booleans = Array(true, false))
def testSingularLogSizeBytesMetric(systemRemoteStorageEnabled: Boolean): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics(systemRemoteStorageEnabled)
-
+ val props = kafka.utils.TestUtils.createDummyBrokerConfig()
+ props.setProperty(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, systemRemoteStorageEnabled.toString)
+ val brokerTopicStats = new BrokerTopicStats(java.util.Optional.of(KafkaConfig.fromProps(props)))
+ val brokerTopicMetrics = brokerTopicStats.topicStats(topic)
if (systemRemoteStorageEnabled) {
- brokerTopicMetrics.recordRemoteLogSizeBytes(0, 100);
- brokerTopicMetrics.recordRemoteLogSizeBytes(1, 150);
- brokerTopicMetrics.recordRemoteLogSizeBytes(2, 250);
+ brokerTopicStats.recordRemoteLogSizeBytes(topic, 0, 100)
+ brokerTopicStats.recordRemoteLogSizeBytes(topic, 1, 150)
+ brokerTopicStats.recordRemoteLogSizeBytes(topic, 2, 250)
assertEquals(500, brokerTopicMetrics.remoteLogSizeBytes)
+ assertEquals(500, brokerTopicStats.allTopicsStats.remoteLogSizeBytes)
+
+ brokerTopicStats.recordRemoteLogSizeBytes(topic2, 0, 100)
+ assertEquals(600, brokerTopicStats.allTopicsStats.remoteLogSizeBytes)
} else {
assertEquals(None, brokerTopicMetrics.metricGaugeMap.get(RemoteStorageMetrics.REMOTE_LOG_SIZE_BYTES_METRIC.getName))
}
}
@Test
def testMultipleLogSizeBytesMetrics(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
+ brokerTopicStats.recordRemoteLogSizeBytes(topic, 0, 1)
+ brokerTopicStats.recordRemoteLogSizeBytes(topic, 1, 2)
+ brokerTopicStats.recordRemoteLogSizeBytes(topic, 2, 3)
- brokerTopicMetrics.recordRemoteLogSizeBytes(0, 1);
- brokerTopicMetrics.recordRemoteLogSizeBytes(1, 2);
- brokerTopicMetrics.recordRemoteLogSizeBytes(2, 3);
-
- brokerTopicMetrics.recordRemoteLogSizeBytes(0, 4);
- brokerTopicMetrics.recordRemoteLogSizeBytes(1, 5);
- brokerTopicMetrics.recordRemoteLogSizeBytes(2, 6);
+ brokerTopicStats.recordRemoteLogSizeBytes(topic, 0, 4)
+ brokerTopicStats.recordRemoteLogSizeBytes(topic, 1, 5)
+ brokerTopicStats.recordRemoteLogSizeBytes(topic, 2, 6)
assertEquals(15, brokerTopicMetrics.remoteLogSizeBytes)
+ assertEquals(15, allTopicMetrics.remoteLogSizeBytes)
}
@Test
def testLogSizeBytesMetricWithPartitionExpansion(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
-
- brokerTopicMetrics.recordRemoteLogSizeBytes(0, 1);
- brokerTopicMetrics.recordRemoteLogSizeBytes(1, 2);
+ brokerTopicStats.recordRemoteLogSizeBytes(topic, 0, 1)
+ brokerTopicStats.recordRemoteLogSizeBytes(topic, 1, 2)
assertEquals(3, brokerTopicMetrics.remoteLogSizeBytes)
+ assertEquals(3, allTopicMetrics.remoteLogSizeBytes)
- brokerTopicMetrics.recordRemoteLogSizeBytes(2, 3);
+ brokerTopicStats.recordRemoteLogSizeBytes(topic, 2, 3)
assertEquals(6, brokerTopicMetrics.remoteLogSizeBytes)
+ assertEquals(6, allTopicMetrics.remoteLogSizeBytes)
}
@Test
def testLogSizeBytesMetricWithPartitionShrinking(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
-
- brokerTopicMetrics.recordRemoteLogSizeBytes(0, 1);
- brokerTopicMetrics.recordRemoteLogSizeBytes(1, 2);
+ brokerTopicStats.recordRemoteLogSizeBytes(topic, 0, 1)
+ brokerTopicStats.recordRemoteLogSizeBytes(topic, 1, 2)
assertEquals(3, brokerTopicMetrics.remoteLogSizeBytes)
+ assertEquals(3, allTopicMetrics.remoteLogSizeBytes)
- brokerTopicMetrics.removeRemoteLogSizeBytes(1);
+ brokerTopicStats.removeRemoteLogSizeBytes(topic, 1)
assertEquals(1, brokerTopicMetrics.remoteLogSizeBytes)
+ assertEquals(1, allTopicMetrics.remoteLogSizeBytes)
}
@Test
def testLogSizeBytesMetricWithRemovingNonexistentPartitions(): Unit = {
- val brokerTopicMetrics = setupBrokerTopicMetrics()
-
- brokerTopicMetrics.recordRemoteLogSizeBytes(0, 1);
- brokerTopicMetrics.recordRemoteLogSizeBytes(1, 2);
+// val props = kafka.utils.TestUtils.createDummyBrokerConfig()
Review Comment:
nit
delete this?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #15133:
URL: https://github.com/apache/kafka/pull/15133#discussion_r1461439023
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -529,20 +564,33 @@ class BrokerTopicMetrics(name: Option[String], configOpt: java.util.Optional[Kaf
}
}
-class BrokerTopicAggregatedMetric() {
+class BrokerTopicAggregatedMetric(allTopics: Boolean = false) {
private val partitionMetricValues = new ConcurrentHashMap[Int, Long]()
+ private val topicMetricValues = new ConcurrentHashMap[String, Long]()
def setPartitionMetricValue(partition: Int, partitionValue: Long): Unit = {
partitionMetricValues.put(partition, partitionValue)
}
+ def setTopicMetricValue(topic: String, topicValue: Long): Unit = {
Review Comment:
Ah, I know what you mean now. Give me some time to fix it. Thanks for pointing it out!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #15133:
URL: https://github.com/apache/kafka/pull/15133#discussion_r1463172726
##########
core/src/test/scala/unit/kafka/utils/TestUtils.scala:
##########
@@ -1396,11 +1396,13 @@ object TestUtils extends Logging {
// Note: Call this method in the test itself, rather than the @AfterEach method.
// Because of the assert, if assertNoNonDaemonThreads fails, nothing after would be executed.
def assertNoNonDaemonThreads(threadNamePrefix: String): Unit = {
- val nonDaemonThreads = Thread.getAllStackTraces.keySet.asScala.filter { t =>
- !t.isDaemon && t.isAlive && t.getName.startsWith(threadNamePrefix)
- }
- val threadCount = nonDaemonThreads.size
- assertEquals(0, threadCount, s"Found unexpected $threadCount NonDaemon threads=${nonDaemonThreads.map(t => t.getName).mkString(", ")}")
+ var nonDemonThreads: mutable.Set[Thread] = mutable.Set.empty[Thread]
+ waitUntilTrue(() => {
+ nonDemonThreads = Thread.getAllStackTraces.keySet.asScala.filter { t =>
+ !t.isDaemon && t.isAlive && t.getName.startsWith(threadNamePrefix)
+ }
+ 0 == nonDemonThreads.size
+ }, s"Found unexpected ${nonDemonThreads.size} NonDaemon threads=${nonDemonThreads.map(t => t.getName).mkString(", ")}", 1000)
Review Comment:
cc @divijvaidya , I found sometimes the [CI](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-15133/9/testReport/junit/kafka.server/ReplicaManagerTest/Build___JDK_11_and_Scala_2_13___testSuccessfulBuildRemoteLogAuxStateMetrics__/) is too sensitive to the non demean threads check. There are some shutdown are in async way. So you can check the failed results [here](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-15133/9/): Basically, if there are some resource not closed, all the following tests should also fail (I verified in my local env). But in the CI results, it only fail 2 of replicaManagertest, and only in jdk11. So I'm going to verify it using `waitUntilTrue` to give it some chance to wait for the threads shutdown.
I also set the wait time as 1 second because if there are really resources leaked, the wait time will be the product of `waitTime` and the number of all the following failed tests.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on code in PR #15133:
URL: https://github.com/apache/kafka/pull/15133#discussion_r1463200794
##########
core/src/test/scala/unit/kafka/utils/TestUtils.scala:
##########
@@ -1396,11 +1396,13 @@ object TestUtils extends Logging {
// Note: Call this method in the test itself, rather than the @AfterEach method.
// Because of the assert, if assertNoNonDaemonThreads fails, nothing after would be executed.
def assertNoNonDaemonThreads(threadNamePrefix: String): Unit = {
- val nonDaemonThreads = Thread.getAllStackTraces.keySet.asScala.filter { t =>
- !t.isDaemon && t.isAlive && t.getName.startsWith(threadNamePrefix)
- }
- val threadCount = nonDaemonThreads.size
- assertEquals(0, threadCount, s"Found unexpected $threadCount NonDaemon threads=${nonDaemonThreads.map(t => t.getName).mkString(", ")}")
+ var nonDemonThreads: mutable.Set[Thread] = mutable.Set.empty[Thread]
+ waitUntilTrue(() => {
+ nonDemonThreads = Thread.getAllStackTraces.keySet.asScala.filter { t =>
+ !t.isDaemon && t.isAlive && t.getName.startsWith(threadNamePrefix)
+ }
+ 0 == nonDemonThreads.size
+ }, s"Found unexpected ${nonDemonThreads.size} NonDaemon threads=${nonDemonThreads.map(t => t.getName).mkString(", ")}", 1000)
Review Comment:
BTW, which threads do you think are closed async during Kafka server shutdown? AFAIK, we have checks in place to ensure that we wait for proper close during shutdown
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #15133:
URL: https://github.com/apache/kafka/pull/15133#issuecomment-1878605296
> Okay, correct me if I am wrong - your idea is that we will use the aggregate object to either store partitions->values or topics->values. For the case where we don't have a topic tag we use the second, in all other cases we use the first.
Correct!
> If this is correct, we need to add the brokerTopicStats.allTopicsStats()... to where we record each metric, no?
Right, Like this: https://github.com/apache/kafka/pull/15133/files#diff-380e4d8859ea9148f21794c09039425c82d9012a392c2dbbe1ce2ec8677a1970R792
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on PR #15133:
URL: https://github.com/apache/kafka/pull/15133#issuecomment-1886780194
I will aim to provide another review today!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #15133:
URL: https://github.com/apache/kafka/pull/15133#discussion_r1448610306
##########
core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java:
##########
@@ -860,28 +871,39 @@ void testRemoteLogManagerRemoteMetrics() throws Exception {
when(mockLog.onlyLocalLogSegmentsCount()).thenReturn(1L);
// before running tasks, the metric should not be registered
- assertThrows(NoSuchElementException.class, () -> yammerMetricValue("RemoteCopyLagBytes"));
- assertThrows(NoSuchElementException.class, () -> yammerMetricValue("RemoteCopyLagSegments"));
- assertThrows(NoSuchElementException.class, () -> yammerMetricValue("RemoteLogSizeComputationTime"));
+ assertThrows(NoSuchElementException.class, () -> yammerMetricValue("RemoteCopyLagBytes,topic=" + leaderTopic));
+ assertThrows(NoSuchElementException.class, () -> yammerMetricValue("RemoteCopyLagSegments,topic=" + leaderTopic));
+ assertThrows(NoSuchElementException.class, () -> yammerMetricValue("RemoteLogSizeComputationTime,topic=" + leaderTopic));
+ // all topic metrics should be 0
+ assertEquals(0L, yammerMetricValue("RemoteCopyLagBytes"));
+ assertEquals(0L, yammerMetricValue("RemoteCopyLagSegments"));
+ assertEquals(0L, yammerMetricValue("RemoteLogSizeComputationTime"));
+
remoteLogManager.onLeadershipChange(Collections.singleton(mockLeaderPartition), Collections.emptySet(), topicIds);
TestUtils.waitForCondition(
- () -> 75 == safeLongYammerMetricValue("RemoteCopyLagBytes"),
- String.format("Expected to find 75 for RemoteCopyLagBytes metric value, but found %d", safeLongYammerMetricValue("RemoteCopyLagBytes")));
+ () -> 75 == safeLongYammerMetricValue("RemoteCopyLagBytes") && 75 == safeLongYammerMetricValue("RemoteCopyLagBytes,topic=" + leaderTopic),
+ String.format("Expected to find 75 for RemoteCopyLagBytes metric value, but found %d for topic 'Leader' and %d for all topics.",
+ safeLongYammerMetricValue("RemoteCopyLagBytes,topic=" + leaderTopic),
+ safeLongYammerMetricValue("RemoteCopyLagBytes")));
TestUtils.waitForCondition(
- () -> 1 == safeLongYammerMetricValue("RemoteCopyLagSegments"),
- String.format("Expected to find 1 for RemoteCopyLagSegments metric value, but found %d", safeLongYammerMetricValue("RemoteCopyLagSegments")));
+ () -> 1 == safeLongYammerMetricValue("RemoteCopyLagSegments") && 1 == safeLongYammerMetricValue("RemoteCopyLagSegments,topic=" + leaderTopic),
+ String.format("Expected to find 1 for RemoteCopyLagSegments metric value, but found %d for topic 'Leader' and %d for all topics.",
+ safeLongYammerMetricValue("RemoteCopyLagSegments,topic=" + leaderTopic),
+ safeLongYammerMetricValue("RemoteCopyLagSegments")));
// unlock copyLogSegmentData
latch.countDown();
TestUtils.waitForCondition(
- () -> safeLongYammerMetricValue("RemoteLogSizeComputationTime") >= 1000,
- String.format("Expected to find 1000 for RemoteLogSizeComputationTime metric value, but found %d", safeLongYammerMetricValue("RemoteLogSizeComputationTime")));
+ () -> safeLongYammerMetricValue("RemoteLogSizeComputationTime") >= 1000 && safeLongYammerMetricValue("RemoteLogSizeComputationTime,topic=" + leaderTopic) >= 1000,
+ String.format("Expected to find 1000 for RemoteLogSizeComputationTime metric value, but found %d for topic 'Leader' and %d for all topics.",
+ safeLongYammerMetricValue("RemoteLogSizeComputationTime,topic=" + leaderTopic),
+ safeLongYammerMetricValue("RemoteLogSizeComputationTime")));
remoteLogSizeComputationTimeLatch.countDown();
}
private Object yammerMetricValue(String name) {
Gauge gauge = (Gauge) KafkaYammerMetrics.defaultRegistry().allMetrics().entrySet().stream()
- .filter(e -> e.getKey().getMBeanName().contains(name))
+ .filter(e -> e.getKey().getMBeanName().endsWith(name))
Review Comment:
Using `endsWith`, so that we can test the all topic metrics and the specific topic metrics. Ex:
`kafka.server:type=BrokerTopicMetrics,name=RemoteCopyLagSegments` and
`kafka.server:type=BrokerTopicMetrics,name=RemoteCopyLagSegments,topic=Leader`
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #15133:
URL: https://github.com/apache/kafka/pull/15133#discussion_r1461855180
##########
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:
When topic deleted, replicaManager will remove all metrics for the topic. All we need to do is to update the allTopics metrics value by removing the topic key from the internal map
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #15133:
URL: https://github.com/apache/kafka/pull/15133#discussion_r1463172726
##########
core/src/test/scala/unit/kafka/utils/TestUtils.scala:
##########
@@ -1396,11 +1396,13 @@ object TestUtils extends Logging {
// Note: Call this method in the test itself, rather than the @AfterEach method.
// Because of the assert, if assertNoNonDaemonThreads fails, nothing after would be executed.
def assertNoNonDaemonThreads(threadNamePrefix: String): Unit = {
- val nonDaemonThreads = Thread.getAllStackTraces.keySet.asScala.filter { t =>
- !t.isDaemon && t.isAlive && t.getName.startsWith(threadNamePrefix)
- }
- val threadCount = nonDaemonThreads.size
- assertEquals(0, threadCount, s"Found unexpected $threadCount NonDaemon threads=${nonDaemonThreads.map(t => t.getName).mkString(", ")}")
+ var nonDemonThreads: mutable.Set[Thread] = mutable.Set.empty[Thread]
+ waitUntilTrue(() => {
+ nonDemonThreads = Thread.getAllStackTraces.keySet.asScala.filter { t =>
+ !t.isDaemon && t.isAlive && t.getName.startsWith(threadNamePrefix)
+ }
+ 0 == nonDemonThreads.size
+ }, s"Found unexpected ${nonDemonThreads.size} NonDaemon threads=${nonDemonThreads.map(t => t.getName).mkString(", ")}", 1000)
Review Comment:
cc @divijvaidya , I found sometimes the [CI](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-15133/9/testReport/junit/kafka.server/ReplicaManagerTest/Build___JDK_11_and_Scala_2_13___testSuccessfulBuildRemoteLogAuxStateMetrics__/) is too sensitive to the non demean threads check. There are some shutdown are in async way. So you can check the failed results [here](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-15133/9/): Basically, if there are some resource not closed, all the following tests should also fail (I verified in my local env). But in the CI results, it only fail 2 of replicaManagertest, and only in jdk11. So I'm going to verify it using `waitUntilTrue` to give it some chance to wait for the threads shutdown.
I also set the wait time as 1 second because if there are really resources leaked, the total wait time will be the product of `waitTime` and the number of all the following failed tests.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #15133:
URL: https://github.com/apache/kafka/pull/15133#discussion_r1462970366
##########
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:
@kamalcph , when running `ReplicaManager#stopPartitions`, it'll call [maybeRemoveTopicMetrics](https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/server/ReplicaManager.scala#L604), and the specific topic metrics will be removed if all partitions are offline (i.e. topics deletion). And the metrics removal will also remove the topic key from brokerTopicStats [here](https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/server/KafkaRequestHandler.scala#L646). That's when it will be deleted.
So, it the topic metrics are all deleted, we also need to remove the topic key from broker-level allTopics metrics. Hope that's clear.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #15133:
URL: https://github.com/apache/kafka/pull/15133#discussion_r1463221572
##########
core/src/test/scala/unit/kafka/utils/TestUtils.scala:
##########
@@ -1396,11 +1396,13 @@ object TestUtils extends Logging {
// Note: Call this method in the test itself, rather than the @AfterEach method.
// Because of the assert, if assertNoNonDaemonThreads fails, nothing after would be executed.
def assertNoNonDaemonThreads(threadNamePrefix: String): Unit = {
- val nonDaemonThreads = Thread.getAllStackTraces.keySet.asScala.filter { t =>
- !t.isDaemon && t.isAlive && t.getName.startsWith(threadNamePrefix)
- }
- val threadCount = nonDaemonThreads.size
- assertEquals(0, threadCount, s"Found unexpected $threadCount NonDaemon threads=${nonDaemonThreads.map(t => t.getName).mkString(", ")}")
+ var nonDemonThreads: mutable.Set[Thread] = mutable.Set.empty[Thread]
+ waitUntilTrue(() => {
+ nonDemonThreads = Thread.getAllStackTraces.keySet.asScala.filter { t =>
+ !t.isDaemon && t.isAlive && t.getName.startsWith(threadNamePrefix)
+ }
+ 0 == nonDemonThreads.size
+ }, s"Found unexpected ${nonDemonThreads.size} NonDaemon threads=${nonDemonThreads.map(t => t.getName).mkString(", ")}", 1000)
Review Comment:
> BTW, which threads do you think are closed async during Kafka server shutdown? AFAIK, we have checks in place to ensure that we wait for proper close during shutdown
Because in the failed test cases, They are `ReplicaFetcherThread` leaked. But had a quick look, yes, we did do `awaitShutdown()` for fetcher threads. Hmm... Let me dig into further tomorrow.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: add all topic metrics [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #15133:
URL: https://github.com/apache/kafka/pull/15133#discussion_r1442611638
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -529,18 +534,28 @@ class BrokerTopicMetrics(name: Option[String], configOpt: java.util.Optional[Kaf
}
}
-class BrokerTopicAggregatedMetric() {
+class BrokerTopicAggregatedMetric(allTopics: Boolean = false) {
private val partitionMetricValues = new ConcurrentHashMap[Int, Long]()
+ private val topicMetricValues = new ConcurrentHashMap[String, Long]()
Review Comment:
add a `topicMetricValues` for allTopics case.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: add all topic metrics [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #15133:
URL: https://github.com/apache/kafka/pull/15133#issuecomment-1878302828
@clolov @satishd @kamalcph , do you think this fix makes sense?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: add all topic metrics [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #15133:
URL: https://github.com/apache/kafka/pull/15133#issuecomment-1878558194
> Should we write a test that could have caught this?
Will do! Just want to make sure the solution makes sense.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #15133:
URL: https://github.com/apache/kafka/pull/15133#discussion_r1449241531
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -473,6 +498,11 @@ class BrokerTopicMetrics(name: Option[String], configOpt: java.util.Optional[Kaf
brokerTopicAggregatedMetric.setPartitionMetricValue(partition, segmentsLag)
}
+ def recordRemoteDeleteLagBytes(topic: String, segmentsLag: Long): Unit = {
Review Comment:
Can we rename `segmentsLag` to `bytesLag`?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #15133:
URL: https://github.com/apache/kafka/pull/15133#issuecomment-1888394290
@satishd , call for review. Thanks.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
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`
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on code in PR #15133:
URL: https://github.com/apache/kafka/pull/15133#discussion_r1463198772
##########
core/src/test/scala/unit/kafka/utils/TestUtils.scala:
##########
@@ -1396,11 +1396,13 @@ object TestUtils extends Logging {
// Note: Call this method in the test itself, rather than the @AfterEach method.
// Because of the assert, if assertNoNonDaemonThreads fails, nothing after would be executed.
def assertNoNonDaemonThreads(threadNamePrefix: String): Unit = {
- val nonDaemonThreads = Thread.getAllStackTraces.keySet.asScala.filter { t =>
- !t.isDaemon && t.isAlive && t.getName.startsWith(threadNamePrefix)
- }
- val threadCount = nonDaemonThreads.size
- assertEquals(0, threadCount, s"Found unexpected $threadCount NonDaemon threads=${nonDaemonThreads.map(t => t.getName).mkString(", ")}")
+ var nonDemonThreads: mutable.Set[Thread] = mutable.Set.empty[Thread]
+ waitUntilTrue(() => {
+ nonDemonThreads = Thread.getAllStackTraces.keySet.asScala.filter { t =>
+ !t.isDaemon && t.isAlive && t.getName.startsWith(threadNamePrefix)
+ }
+ 0 == nonDemonThreads.size
+ }, s"Found unexpected ${nonDemonThreads.size} NonDaemon threads=${nonDemonThreads.map(t => t.getName).mkString(", ")}", 1000)
Review Comment:
This will increase the run-time for test suite when it is leaking threads. Which should be ok, as the happy case (when we are not leaking remains the same).
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #15133:
URL: https://github.com/apache/kafka/pull/15133#issuecomment-1903976486
@divijvaidya , thanks for the comments. I've updated the PR. Please take a look when available. Thanks.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16085: Add metric value consolidated for topics on a broker for tiered storage. [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #15133:
URL: https://github.com/apache/kafka/pull/15133#issuecomment-1900326555
> Most are optional comments but the blocking comments before we merge are the once related to removal of metrics when topic is deleted. I would assume that removal needs to occur from newly introduced `topicMetricValues` map in `BrokerTopicAggregateMetric`?
@divijvaidya , thanks for your review. The metrics removal of topic deletion will be triggered in rlm#stopPartitions, and then `removeRemoteTopicPartitionMetrics` will be invoked. The removal for the "allTopic" metric is done via removing the partition value from the specific topic, ex:
```
# We first remove the partition value from this topic
brokerTopicStats.topicStats(topicIdPartition.topic()).removeRemoteCopyLagBytes(partition);
# Later, we update the topic value in the "allTopic" metric. So, once all partitions are removed, the value of the specific topic will be 0.
brokerTopicStats.allTopicsStats().recordRemoteCopyLagBytes(topic, topicMetrics.remoteCopyLagBytes());
```
I'll address your comments this weekend or next week. Thanks.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org