You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mi...@apache.org on 2024/08/27 09:39:44 UTC
(kafka) branch trunk updated: KAFKA-16972: Move BrokerTopicStats to storage module (#17003)
This is an automated email from the ASF dual-hosted git repository.
mimaison pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new b9fe9f532fe KAFKA-16972: Move BrokerTopicStats to storage module (#17003)
b9fe9f532fe is described below
commit b9fe9f532fe11c658d07b11543e874526cb837b8
Author: Mickael Maison <mi...@users.noreply.github.com>
AuthorDate: Tue Aug 27 11:39:37 2024 +0200
KAFKA-16972: Move BrokerTopicStats to storage module (#17003)
Reviewers: Luke Chen <sh...@gmail.com>
---
checkstyle/import-control-core.xml | 1 +
.../java/kafka/log/remote/RemoteLogManager.java | 2 +-
.../java/kafka/log/remote/RemoteLogReader.java | 2 +-
.../kafka/server/builders/KafkaApisBuilder.java | 2 +-
.../kafka/server/builders/LogManagerBuilder.java | 2 +-
.../server/builders/ReplicaManagerBuilder.java | 2 +-
core/src/main/scala/kafka/log/LogManager.scala | 3 +-
core/src/main/scala/kafka/log/UnifiedLog.scala | 4 +-
.../main/scala/kafka/raft/KafkaMetadataLog.scala | 3 +-
.../scala/kafka/server/AbstractFetcherThread.scala | 1 +
.../src/main/scala/kafka/server/BrokerServer.scala | 3 +-
core/src/main/scala/kafka/server/KafkaApis.scala | 1 +
core/src/main/scala/kafka/server/KafkaBroker.scala | 1 +
.../scala/kafka/server/KafkaRequestHandler.scala | 233 +-----------------
core/src/main/scala/kafka/server/KafkaServer.scala | 1 +
.../kafka/server/ReplicaAlterLogDirsManager.scala | 1 +
.../kafka/server/ReplicaAlterLogDirsThread.scala | 1 +
.../main/scala/kafka/server/ReplicaManager.scala | 3 +-
.../kafka/log/remote/RemoteLogManagerTest.java | 2 +-
.../java/kafka/log/remote/RemoteLogReaderTest.java | 2 +-
.../kafka/server/KafkaRequestHandlerTest.scala | 2 +-
.../src/test/scala/other/kafka/StressTestLog.scala | 2 +-
.../scala/other/kafka/TestLinearWriteSpeed.scala | 2 +-
.../unit/kafka/cluster/PartitionLockTest.scala | 1 +
.../scala/unit/kafka/cluster/PartitionTest.scala | 1 +
.../log/AbstractLogCleanerIntegrationTest.scala | 2 +-
.../unit/kafka/log/BrokerCompressionTest.scala | 2 +-
.../unit/kafka/log/LogCleanerManagerTest.scala | 2 +-
.../test/scala/unit/kafka/log/LogCleanerTest.scala | 3 +-
.../scala/unit/kafka/log/LogConcurrencyTest.scala | 2 +-
.../test/scala/unit/kafka/log/LogLoaderTest.scala | 3 +-
.../test/scala/unit/kafka/log/LogManagerTest.scala | 2 +-
.../test/scala/unit/kafka/log/LogTestUtils.scala | 2 +-
.../test/scala/unit/kafka/log/UnifiedLogTest.scala | 4 +-
.../kafka/server/AbstractFetcherManagerTest.scala | 1 +
.../scala/unit/kafka/server/KafkaApisTest.scala | 1 +
.../unit/kafka/server/MockFetcherThread.scala | 1 +
.../server/ReplicaAlterLogDirsThreadTest.scala | 1 +
.../kafka/server/ReplicaFetcherThreadTest.scala | 1 +
.../unit/kafka/server/ReplicaManagerTest.scala | 1 +
.../unit/kafka/tools/DumpLogSegmentsTest.scala | 3 +-
.../scala/unit/kafka/utils/SchedulerTest.scala | 2 +-
.../test/scala/unit/kafka/utils/TestUtils.scala | 1 +
.../jmh/fetcher/ReplicaFetcherThreadBenchmark.java | 2 +-
.../metadata/KRaftMetadataRequestBenchmark.java | 2 +-
.../jmh/metadata/MetadataRequestBenchmark.java | 2 +-
.../partition/PartitionMakeFollowerBenchmark.java | 2 +-
.../UpdateFollowerFetchStateBenchmark.java | 2 +-
.../kafka/jmh/record/BaseRecordBatchBenchmark.java | 2 +-
.../apache/kafka/jmh/server/CheckpointBench.java | 2 +-
.../kafka/jmh/server/PartitionCreationBench.java | 2 +-
.../storage/log/metrics/BrokerTopicStats.java | 265 +++++++++++++++++++++
52 files changed, 325 insertions(+), 268 deletions(-)
diff --git a/checkstyle/import-control-core.xml b/checkstyle/import-control-core.xml
index 1d9f2029c8e..3159dd5c81e 100644
--- a/checkstyle/import-control-core.xml
+++ b/checkstyle/import-control-core.xml
@@ -82,6 +82,7 @@
<allow pkg="org.apache.kafka.server.log.remote" />
<allow pkg="org.apache.kafka.server.metrics" />
<allow pkg="org.apache.kafka.storage.internals" />
+ <allow pkg="org.apache.kafka.storage.log.metrics" />
<allow pkg="kafka.log" />
<allow pkg="kafka.cluster" />
<allow pkg="kafka.server" />
diff --git a/core/src/main/java/kafka/log/remote/RemoteLogManager.java b/core/src/main/java/kafka/log/remote/RemoteLogManager.java
index a1714211af5..049ad529eee 100644
--- a/core/src/main/java/kafka/log/remote/RemoteLogManager.java
+++ b/core/src/main/java/kafka/log/remote/RemoteLogManager.java
@@ -22,7 +22,6 @@ import kafka.log.UnifiedLog;
import kafka.log.remote.quota.RLMQuotaManager;
import kafka.log.remote.quota.RLMQuotaManagerConfig;
import kafka.log.remote.quota.RLMQuotaMetrics;
-import kafka.server.BrokerTopicStats;
import kafka.server.QuotaType;
import kafka.server.StopPartition;
@@ -81,6 +80,7 @@ import org.apache.kafka.storage.internals.log.RemoteStorageFetchInfo;
import org.apache.kafka.storage.internals.log.RemoteStorageThreadPool;
import org.apache.kafka.storage.internals.log.TransactionIndex;
import org.apache.kafka.storage.internals.log.TxnIndexSearchResult;
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
import com.yammer.metrics.core.Timer;
diff --git a/core/src/main/java/kafka/log/remote/RemoteLogReader.java b/core/src/main/java/kafka/log/remote/RemoteLogReader.java
index 93ef2f0ee5e..79ff51cfd2f 100644
--- a/core/src/main/java/kafka/log/remote/RemoteLogReader.java
+++ b/core/src/main/java/kafka/log/remote/RemoteLogReader.java
@@ -17,12 +17,12 @@
package kafka.log.remote;
import kafka.log.remote.quota.RLMQuotaManager;
-import kafka.server.BrokerTopicStats;
import org.apache.kafka.common.errors.OffsetOutOfRangeException;
import org.apache.kafka.storage.internals.log.FetchDataInfo;
import org.apache.kafka.storage.internals.log.RemoteLogReadResult;
import org.apache.kafka.storage.internals.log.RemoteStorageFetchInfo;
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
import com.yammer.metrics.core.Timer;
diff --git a/core/src/main/java/kafka/server/builders/KafkaApisBuilder.java b/core/src/main/java/kafka/server/builders/KafkaApisBuilder.java
index 128d5796006..33e13e03b23 100644
--- a/core/src/main/java/kafka/server/builders/KafkaApisBuilder.java
+++ b/core/src/main/java/kafka/server/builders/KafkaApisBuilder.java
@@ -21,7 +21,6 @@ import kafka.coordinator.transaction.TransactionCoordinator;
import kafka.network.RequestChannel;
import kafka.server.ApiVersionManager;
import kafka.server.AutoTopicCreationManager;
-import kafka.server.BrokerTopicStats;
import kafka.server.DelegationTokenManager;
import kafka.server.FetchManager;
import kafka.server.KafkaApis;
@@ -38,6 +37,7 @@ import org.apache.kafka.common.utils.Time;
import org.apache.kafka.coordinator.group.GroupCoordinator;
import org.apache.kafka.server.ClientMetricsManager;
import org.apache.kafka.server.authorizer.Authorizer;
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
import java.util.Collections;
import java.util.Optional;
diff --git a/core/src/main/java/kafka/server/builders/LogManagerBuilder.java b/core/src/main/java/kafka/server/builders/LogManagerBuilder.java
index 977e914bc05..671321cb20f 100644
--- a/core/src/main/java/kafka/server/builders/LogManagerBuilder.java
+++ b/core/src/main/java/kafka/server/builders/LogManagerBuilder.java
@@ -18,7 +18,6 @@
package kafka.server.builders;
import kafka.log.LogManager;
-import kafka.server.BrokerTopicStats;
import kafka.server.metadata.ConfigRepository;
import org.apache.kafka.common.utils.Time;
@@ -29,6 +28,7 @@ import org.apache.kafka.storage.internals.log.CleanerConfig;
import org.apache.kafka.storage.internals.log.LogConfig;
import org.apache.kafka.storage.internals.log.LogDirFailureChannel;
import org.apache.kafka.storage.internals.log.ProducerStateManagerConfig;
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
import java.io.File;
import java.util.Collections;
diff --git a/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java b/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java
index 16397819719..944ef908c7e 100644
--- a/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java
+++ b/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java
@@ -21,7 +21,6 @@ import kafka.log.LogManager;
import kafka.log.remote.RemoteLogManager;
import kafka.server.AddPartitionsToTxnManager;
import kafka.server.AlterPartitionManager;
-import kafka.server.BrokerTopicStats;
import kafka.server.DelayedDeleteRecords;
import kafka.server.DelayedElectLeader;
import kafka.server.DelayedFetch;
@@ -39,6 +38,7 @@ import org.apache.kafka.common.utils.Time;
import org.apache.kafka.server.common.DirectoryEventHandler;
import org.apache.kafka.server.util.Scheduler;
import org.apache.kafka.storage.internals.log.LogDirFailureChannel;
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
import java.util.Collections;
import java.util.Optional;
diff --git a/core/src/main/scala/kafka/log/LogManager.scala b/core/src/main/scala/kafka/log/LogManager.scala
index 7d5c05c015f..e9d782bba9d 100755
--- a/core/src/main/scala/kafka/log/LogManager.scala
+++ b/core/src/main/scala/kafka/log/LogManager.scala
@@ -23,7 +23,7 @@ import java.nio.file.{Files, NoSuchFileException}
import java.util.concurrent._
import java.util.concurrent.atomic.AtomicInteger
import kafka.server.metadata.ConfigRepository
-import kafka.server.{BrokerTopicStats, KafkaConfig, KafkaRaftServer}
+import kafka.server.{KafkaConfig, KafkaRaftServer}
import kafka.server.metadata.BrokerMetadataPublisher.info
import kafka.utils.threadsafe
import kafka.utils.{CoreUtils, Logging, Pool}
@@ -48,6 +48,7 @@ import org.apache.kafka.server.metrics.KafkaMetricsGroup
import org.apache.kafka.server.util.{FileLock, Scheduler}
import org.apache.kafka.storage.internals.log.{CleanerConfig, LogConfig, LogDirFailureChannel, ProducerStateManagerConfig, RemoteIndexCache}
import org.apache.kafka.storage.internals.checkpoint.{CleanShutdownFileHandler, OffsetCheckpointFile}
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import java.util
import scala.annotation.nowarn
diff --git a/core/src/main/scala/kafka/log/UnifiedLog.scala b/core/src/main/scala/kafka/log/UnifiedLog.scala
index a51c06d8b97..e26406f623a 100644
--- a/core/src/main/scala/kafka/log/UnifiedLog.scala
+++ b/core/src/main/scala/kafka/log/UnifiedLog.scala
@@ -20,7 +20,7 @@ package kafka.log
import kafka.common.{OffsetsOutOfOrderException, UnexpectedAppendOffsetException}
import kafka.log.LocalLog.nextOption
import kafka.log.remote.RemoteLogManager
-import kafka.server.{BrokerTopicStats, RequestLocal}
+import kafka.server.RequestLocal
import kafka.utils._
import org.apache.kafka.common.errors._
import org.apache.kafka.common.internals.Topic
@@ -41,7 +41,7 @@ import org.apache.kafka.server.util.Scheduler
import org.apache.kafka.storage.internals.checkpoint.{LeaderEpochCheckpointFile, PartitionMetadataFile}
import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache
import org.apache.kafka.storage.internals.log.{AbortedTxn, AppendOrigin, BatchMetadata, CompletedTxn, FetchDataInfo, FetchIsolation, LastRecord, LeaderHwChange, LogAppendInfo, LogConfig, LogDirFailureChannel, LogFileUtils, LogOffsetMetadata, LogOffsetSnapshot, LogOffsetsListener, LogSegment, LogSegments, LogStartOffsetIncrementReason, LogValidator, ProducerAppendInfo, ProducerStateManager, ProducerStateManagerConfig, RollParams, VerificationGuard}
-import org.apache.kafka.storage.log.metrics.BrokerTopicMetrics
+import org.apache.kafka.storage.log.metrics.{BrokerTopicMetrics, BrokerTopicStats}
import java.io.{File, IOException}
import java.nio.file.{Files, Path}
diff --git a/core/src/main/scala/kafka/raft/KafkaMetadataLog.scala b/core/src/main/scala/kafka/raft/KafkaMetadataLog.scala
index 5e107aa1487..dae720ef118 100644
--- a/core/src/main/scala/kafka/raft/KafkaMetadataLog.scala
+++ b/core/src/main/scala/kafka/raft/KafkaMetadataLog.scala
@@ -22,7 +22,7 @@ import kafka.raft.KafkaMetadataLog.RetentionMsBreach
import kafka.raft.KafkaMetadataLog.RetentionSizeBreach
import kafka.raft.KafkaMetadataLog.SnapshotDeletionReason
import kafka.raft.KafkaMetadataLog.UnknownReason
-import kafka.server.{BrokerTopicStats, RequestLocal}
+import kafka.server.RequestLocal
import kafka.utils.{CoreUtils, Logging}
import org.apache.kafka.common.config.TopicConfig
import org.apache.kafka.common.errors.InvalidConfigurationException
@@ -41,6 +41,7 @@ import org.apache.kafka.snapshot.SnapshotPath
import org.apache.kafka.snapshot.Snapshots
import org.apache.kafka.storage.internals
import org.apache.kafka.storage.internals.log.{AppendOrigin, FetchIsolation, LogConfig, LogDirFailureChannel, LogStartOffsetIncrementReason, ProducerStateManagerConfig}
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import java.io.File
import java.nio.file.{Files, NoSuchFileException, Path}
diff --git a/core/src/main/scala/kafka/server/AbstractFetcherThread.scala b/core/src/main/scala/kafka/server/AbstractFetcherThread.scala
index 989030f4439..ec5e1d0306c 100755
--- a/core/src/main/scala/kafka/server/AbstractFetcherThread.scala
+++ b/core/src/main/scala/kafka/server/AbstractFetcherThread.scala
@@ -37,6 +37,7 @@ import org.apache.kafka.server.common.OffsetAndEpoch
import org.apache.kafka.server.metrics.KafkaMetricsGroup
import org.apache.kafka.server.util.ShutdownableThread
import org.apache.kafka.storage.internals.log.LogAppendInfo
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import java.nio.ByteBuffer
import java.util
diff --git a/core/src/main/scala/kafka/server/BrokerServer.scala b/core/src/main/scala/kafka/server/BrokerServer.scala
index 08bf5591ca7..ddf74f2ef1f 100644
--- a/core/src/main/scala/kafka/server/BrokerServer.scala
+++ b/core/src/main/scala/kafka/server/BrokerServer.scala
@@ -37,7 +37,7 @@ import org.apache.kafka.common.utils.{LogContext, Time}
import org.apache.kafka.common.{ClusterResource, TopicPartition, Uuid}
import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord
import org.apache.kafka.coordinator.group.metrics.{GroupCoordinatorMetrics, GroupCoordinatorRuntimeMetrics}
-import org.apache.kafka.coordinator.group.{GroupCoordinatorRecordSerde, GroupConfigManager, GroupCoordinator, GroupCoordinatorService}
+import org.apache.kafka.coordinator.group.{GroupConfigManager, GroupCoordinator, GroupCoordinatorRecordSerde, GroupCoordinatorService}
import org.apache.kafka.image.publisher.{BrokerRegistrationTracker, MetadataPublisher}
import org.apache.kafka.metadata.{BrokerState, ListenerInfo}
import org.apache.kafka.security.CredentialProvider
@@ -53,6 +53,7 @@ import org.apache.kafka.server.share.ShareSessionCache
import org.apache.kafka.server.util.timer.{SystemTimer, SystemTimerReaper}
import org.apache.kafka.server.util.{Deadline, FutureUtils, KafkaScheduler}
import org.apache.kafka.storage.internals.log.LogDirFailureChannel
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import java.time.Duration
import java.util
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
index c02e010f17c..8d6b328a434 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -77,6 +77,7 @@ import org.apache.kafka.server.common.MetadataVersion.{IBP_0_11_0_IV0, IBP_2_3_I
import org.apache.kafka.server.record.BrokerCompressionType
import org.apache.kafka.server.share.{ErroneousAndValidPartitionData, ShareAcknowledgementBatch, ShareFetchContext}
import org.apache.kafka.storage.internals.log.{AppendOrigin, FetchIsolation, FetchParams, FetchPartitionData}
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import java.lang.{Long => JLong}
import java.nio.ByteBuffer
diff --git a/core/src/main/scala/kafka/server/KafkaBroker.scala b/core/src/main/scala/kafka/server/KafkaBroker.scala
index bcdf6593b2e..2d0ccb21054 100644
--- a/core/src/main/scala/kafka/server/KafkaBroker.scala
+++ b/core/src/main/scala/kafka/server/KafkaBroker.scala
@@ -34,6 +34,7 @@ import org.apache.kafka.server.NodeToControllerChannelManager
import org.apache.kafka.server.authorizer.Authorizer
import org.apache.kafka.server.metrics.{KafkaMetricsGroup, KafkaYammerMetrics, LinuxIoMetricsCollector}
import org.apache.kafka.server.util.Scheduler
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import java.time.Duration
import scala.collection.Seq
diff --git a/core/src/main/scala/kafka/server/KafkaRequestHandler.scala b/core/src/main/scala/kafka/server/KafkaRequestHandler.scala
index 454552a08e4..b2792fb6fbc 100755
--- a/core/src/main/scala/kafka/server/KafkaRequestHandler.scala
+++ b/core/src/main/scala/kafka/server/KafkaRequestHandler.scala
@@ -18,7 +18,7 @@
package kafka.server
import kafka.network.RequestChannel
-import kafka.utils.{Logging, Pool}
+import kafka.utils.Logging
import kafka.server.KafkaRequestHandler.{threadCurrentRequest, threadRequestChannel}
import java.util.concurrent.{CountDownLatch, TimeUnit}
@@ -26,9 +26,7 @@ import java.util.concurrent.atomic.AtomicInteger
import com.yammer.metrics.core.Meter
import org.apache.kafka.common.internals.FatalExitError
import org.apache.kafka.common.utils.{Exit, KafkaThread, Time}
-import org.apache.kafka.server.log.remote.storage.RemoteStorageMetrics
import org.apache.kafka.server.metrics.KafkaMetricsGroup
-import org.apache.kafka.storage.log.metrics.BrokerTopicMetrics
import scala.collection.mutable
@@ -244,232 +242,3 @@ class KafkaRequestHandlerPool(
info("shut down completely")
}
}
-
-class BrokerTopicStats(remoteStorageEnabled: Boolean = false) extends Logging {
-
- private val valueFactory = (k: String) => new BrokerTopicMetrics(k, remoteStorageEnabled)
- private val stats = new Pool[String, BrokerTopicMetrics](Some(valueFactory))
- val allTopicsStats = new BrokerTopicMetrics(remoteStorageEnabled)
-
- def isTopicStatsExisted(topic: String): Boolean =
- stats.contains(topic)
-
- def topicStats(topic: String): BrokerTopicMetrics =
- stats.getAndMaybePut(topic)
-
- def updateReplicationBytesIn(value: Long): Unit = {
- allTopicsStats.replicationBytesInRate.ifPresent { metric =>
- metric.mark(value)
- }
- }
-
- private def updateReplicationBytesOut(value: Long): Unit = {
- allTopicsStats.replicationBytesOutRate.ifPresent { metric =>
- metric.mark(value)
- }
- }
-
- def updateReassignmentBytesIn(value: Long): Unit = {
- allTopicsStats.reassignmentBytesInPerSec.ifPresent { metric =>
- metric.mark(value)
- }
- }
-
- private def updateReassignmentBytesOut(value: Long): Unit = {
- allTopicsStats.reassignmentBytesOutPerSec.ifPresent { metric =>
- metric.mark(value)
- }
- }
-
- // This method only removes metrics only used for leader
- def removeOldLeaderMetrics(topic: String): Unit = {
- val topicMetrics = topicStats(topic)
- if (topicMetrics != null) {
- topicMetrics.closeMetric(BrokerTopicMetrics.MESSAGE_IN_PER_SEC)
- topicMetrics.closeMetric(BrokerTopicMetrics.BYTES_IN_PER_SEC)
- topicMetrics.closeMetric(BrokerTopicMetrics.BYTES_REJECTED_PER_SEC)
- topicMetrics.closeMetric(BrokerTopicMetrics.FAILED_PRODUCE_REQUESTS_PER_SEC)
- topicMetrics.closeMetric(BrokerTopicMetrics.TOTAL_PRODUCE_REQUESTS_PER_SEC)
- topicMetrics.closeMetric(BrokerTopicMetrics.PRODUCE_MESSAGE_CONVERSIONS_PER_SEC)
- topicMetrics.closeMetric(BrokerTopicMetrics.REPLICATION_BYTES_OUT_PER_SEC)
- topicMetrics.closeMetric(BrokerTopicMetrics.REASSIGNMENT_BYTES_OUT_PER_SEC)
- topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_COPY_BYTES_PER_SEC_METRIC.getName)
- topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_FETCH_BYTES_PER_SEC_METRIC.getName)
- topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_FETCH_REQUESTS_PER_SEC_METRIC.getName)
- topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_COPY_REQUESTS_PER_SEC_METRIC.getName)
- topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_DELETE_REQUESTS_PER_SEC_METRIC.getName)
- topicMetrics.closeMetric(RemoteStorageMetrics.BUILD_REMOTE_LOG_AUX_STATE_REQUESTS_PER_SEC_METRIC.getName)
- topicMetrics.closeMetric(RemoteStorageMetrics.FAILED_REMOTE_FETCH_PER_SEC_METRIC.getName)
- topicMetrics.closeMetric(RemoteStorageMetrics.FAILED_REMOTE_COPY_PER_SEC_METRIC.getName)
- topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_COPY_LAG_BYTES_METRIC.getName)
- topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_LOG_METADATA_COUNT_METRIC.getName)
- topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_LOG_SIZE_COMPUTATION_TIME_METRIC.getName)
- topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_LOG_SIZE_BYTES_METRIC.getName)
- topicMetrics.closeMetric(RemoteStorageMetrics.FAILED_REMOTE_DELETE_PER_SEC_METRIC.getName)
- topicMetrics.closeMetric(RemoteStorageMetrics.FAILED_BUILD_REMOTE_LOG_AUX_STATE_PER_SEC_METRIC.getName)
- topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_COPY_LAG_BYTES_METRIC.getName)
- topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_COPY_LAG_SEGMENTS_METRIC.getName)
- topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_DELETE_LAG_BYTES_METRIC.getName)
- topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_DELETE_LAG_SEGMENTS_METRIC.getName)
- }
- }
-
- // This method only removes metrics only used for follower
- def removeOldFollowerMetrics(topic: String): Unit = {
- val topicMetrics = topicStats(topic)
- if (topicMetrics != null) {
- topicMetrics.closeMetric(BrokerTopicMetrics.REPLICATION_BYTES_IN_PER_SEC)
- topicMetrics.closeMetric(BrokerTopicMetrics.REASSIGNMENT_BYTES_IN_PER_SEC)
- }
- }
-
- def removeMetrics(topic: String): Unit = {
- val metrics = stats.remove(topic)
- if (metrics != null)
- metrics.close()
- }
-
- def updateBytesOut(topic: String, isFollower: Boolean, isReassignment: Boolean, value: Long): Unit = {
- if (isFollower) {
- if (isReassignment)
- updateReassignmentBytesOut(value)
- updateReplicationBytesOut(value)
- } else {
- topicStats(topic).bytesOutRate.mark(value)
- allTopicsStats.bytesOutRate.mark(value)
- }
- }
-
- // 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)
- }
-
- def removeBrokerLevelRemoteCopyLagBytes(topic: String): Unit = {
- allTopicsStats.remoteCopyLagBytesAggrMetric().removeKey(topic)
- }
-
- // Update the broker-level all topic metric values so that we have a sample right for all topics metric after update of partition.
- def recordRemoteCopyLagSegments(topic: String, partition: Int, value: Long): Unit = {
- val topicMetric = topicStats(topic)
- topicMetric.remoteCopyLagSegmentsAggrMetric().setValue(String.valueOf(partition), value)
- allTopicsStats.remoteCopyLagSegmentsAggrMetric().setValue(topic, topicMetric.remoteCopyLagSegments)
- }
-
- // Update the broker-level all topic metric values so that we have a sample right for all topics metric after removal of partition.
- def removeRemoteCopyLagSegments(topic: String, partition: Int): Unit = {
- val topicMetric = topicStats(topic)
- topicMetric.remoteCopyLagSegmentsAggrMetric().removeKey(String.valueOf(partition))
- allTopicsStats.remoteCopyLagSegmentsAggrMetric().setValue(topic, topicMetric.remoteCopyLagSegments)
- }
-
- def removeBrokerLevelRemoteCopyLagSegments(topic: String): Unit = {
- allTopicsStats.remoteCopyLagSegmentsAggrMetric().removeKey(topic)
- }
-
- // Update the broker-level all topic metric values so that we have a sample right for all topics metric after update of partition.
- def recordRemoteDeleteLagBytes(topic: String, partition: Int, value: Long): Unit = {
- val topicMetric = topicStats(topic)
- topicMetric.remoteDeleteLagBytesAggrMetric().setValue(String.valueOf(partition), value)
- allTopicsStats.remoteDeleteLagBytesAggrMetric().setValue(topic, topicMetric.remoteDeleteLagBytes)
- }
-
- // Update the broker-level all topic metric values so that we have a sample right for all topics metric after removal of partition.
- def removeRemoteDeleteLagBytes(topic: String, partition: Int): Unit = {
- val topicMetric = topicStats(topic)
- topicMetric.remoteDeleteLagBytesAggrMetric().removeKey(String.valueOf(partition))
- allTopicsStats.remoteDeleteLagBytesAggrMetric().setValue(topic, topicMetric.remoteDeleteLagBytes)
- }
-
- def removeBrokerLevelRemoteDeleteLagBytes(topic: String): Unit = {
- allTopicsStats.remoteDeleteLagBytesAggrMetric().removeKey(topic)
- }
-
- // Update the broker-level all topic metric values so that we have a sample right for all topics metric after update of partition.
- def recordRemoteDeleteLagSegments(topic: String, partition: Int, value: Long): Unit = {
- val topicMetric = topicStats(topic)
- topicMetric.remoteDeleteLagSegmentsAggrMetric().setValue(String.valueOf(partition), value)
- allTopicsStats.remoteDeleteLagSegmentsAggrMetric().setValue(topic, topicMetric.remoteDeleteLagSegments)
- }
-
- // Update the broker-level all topic metric values so that we have a sample right for all topics metric after removal of partition.
- def removeRemoteDeleteLagSegments(topic: String, partition: Int): Unit = {
- val topicMetric = topicStats(topic)
- topicMetric.remoteDeleteLagSegmentsAggrMetric().removeKey(String.valueOf(partition))
- allTopicsStats.remoteDeleteLagSegmentsAggrMetric().setValue(topic, topicMetric.remoteDeleteLagSegments)
- }
-
- def removeBrokerLevelRemoteDeleteLagSegments(topic: String): Unit = {
- allTopicsStats.remoteDeleteLagSegmentsAggrMetric().removeKey(topic)
- }
-
- // Update the broker-level all topic metric values so that we have a sample right for all topics metric after update of partition.
- def recordRemoteLogMetadataCount(topic: String, partition: Int, value: Long): Unit = {
- val topicMetric = topicStats(topic)
- topicMetric.remoteLogMetadataCountAggrMetric().setValue(String.valueOf(partition), value)
- allTopicsStats.remoteLogMetadataCountAggrMetric().setValue(topic, topicMetric.remoteLogMetadataCount)
- }
-
- // Update the broker-level all topic metric values so that we have a sample right for all topics metric after removal of partition.
- def removeRemoteLogMetadataCount(topic: String, partition: Int): Unit = {
- val topicMetric = topicStats(topic)
- topicMetric.remoteLogMetadataCountAggrMetric().removeKey(String.valueOf(partition))
- allTopicsStats.remoteLogMetadataCountAggrMetric().setValue(topic, topicMetric.remoteLogMetadataCount)
- }
-
- def removeBrokerLevelRemoteLogMetadataCount(topic: String): Unit = {
- allTopicsStats.remoteLogMetadataCountAggrMetric().removeKey(topic)
- }
-
- // Update the broker-level all topic metric values so that we have a sample right for all topics metric after update of partition.
- def recordRemoteLogSizeComputationTime(topic: String, partition: Int, value: Long): Unit = {
- val topicMetric = topicStats(topic)
- topicMetric.remoteLogSizeComputationTimeAggrMetric().setValue(String.valueOf(partition), value)
- allTopicsStats.remoteLogSizeComputationTimeAggrMetric().setValue(topic, topicMetric.remoteLogSizeComputationTime)
- }
-
- // Update the broker-level all topic metric values so that we have a sample right for all topics metric after removal of partition.
- def removeRemoteLogSizeComputationTime(topic: String, partition: Int): Unit = {
- val topicMetric = topicStats(topic)
- topicMetric.remoteLogSizeComputationTimeAggrMetric().removeKey(String.valueOf(partition))
- allTopicsStats.remoteLogSizeComputationTimeAggrMetric().setValue(topic, topicMetric.remoteLogSizeComputationTime)
- }
-
- def removeBrokerLevelRemoteLogSizeComputationTime(topic: String): Unit = {
- allTopicsStats.remoteLogSizeComputationTimeAggrMetric().removeKey(topic)
- }
-
- // Update the broker-level all topic metric values so that we have a sample right for all topics metric after update of partition.
- def recordRemoteLogSizeBytes(topic: String, partition: Int, value: Long): Unit = {
- val topicMetric = topicStats(topic)
- topicMetric.remoteLogSizeBytesAggrMetric().setValue(String.valueOf(partition), value)
- allTopicsStats.remoteLogSizeBytesAggrMetric().setValue(topic, topicMetric.remoteLogSizeBytes)
- }
-
- // Update the broker-level all topic metric values so that we have a sample right for all topics metric after removal of partition.
- def removeRemoteLogSizeBytes(topic: String, partition: Int): Unit = {
- val topicMetric = topicStats(topic)
- topicMetric.remoteLogSizeBytesAggrMetric().removeKey(String.valueOf(partition))
- allTopicsStats.remoteLogSizeBytesAggrMetric().setValue(topic, topicMetric.remoteLogSizeBytes)
- }
-
- def removeBrokerLevelRemoteLogSizeBytes(topic: String): Unit = {
- allTopicsStats.remoteLogSizeBytesAggrMetric().removeKey(topic)
- }
-
- def close(): Unit = {
- allTopicsStats.close()
- stats.values.foreach(_.close())
-
- info("Broker and topic stats closed")
- }
-}
diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala
index 84f169d12cb..80f901d03bf 100755
--- a/core/src/main/scala/kafka/server/KafkaServer.scala
+++ b/core/src/main/scala/kafka/server/KafkaServer.scala
@@ -64,6 +64,7 @@ import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig
import org.apache.kafka.server.metrics.KafkaYammerMetrics
import org.apache.kafka.server.util.KafkaScheduler
import org.apache.kafka.storage.internals.log.LogDirFailureChannel
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import org.apache.zookeeper.client.ZKClientConfig
import java.io.{File, IOException}
diff --git a/core/src/main/scala/kafka/server/ReplicaAlterLogDirsManager.scala b/core/src/main/scala/kafka/server/ReplicaAlterLogDirsManager.scala
index 4dbdbeaed15..6335f7db635 100644
--- a/core/src/main/scala/kafka/server/ReplicaAlterLogDirsManager.scala
+++ b/core/src/main/scala/kafka/server/ReplicaAlterLogDirsManager.scala
@@ -20,6 +20,7 @@ package kafka.server
import kafka.cluster.BrokerEndPoint
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.server.common.DirectoryEventHandler
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
class ReplicaAlterLogDirsManager(brokerConfig: KafkaConfig,
replicaManager: ReplicaManager,
diff --git a/core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala b/core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala
index 95c7a5ac3d4..403cb34180d 100644
--- a/core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala
+++ b/core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala
@@ -23,6 +23,7 @@ import org.apache.kafka.common.{TopicPartition, Uuid}
import org.apache.kafka.common.requests.FetchResponse
import org.apache.kafka.server.common.{DirectoryEventHandler, OffsetAndEpoch, TopicIdPartition}
import org.apache.kafka.storage.internals.log.{LogAppendInfo, LogStartOffsetIncrementReason}
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import java.util.concurrent.ConcurrentHashMap
import scala.collection.{Map, Set}
diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala
index 9c2a2913d99..09c3b1dc4d4 100644
--- a/core/src/main/scala/kafka/server/ReplicaManager.scala
+++ b/core/src/main/scala/kafka/server/ReplicaManager.scala
@@ -59,8 +59,9 @@ import org.apache.kafka.server.common.DirectoryEventHandler
import org.apache.kafka.server.common.MetadataVersion._
import org.apache.kafka.server.metrics.KafkaMetricsGroup
import org.apache.kafka.server.util.{Scheduler, ShutdownableThread}
-import org.apache.kafka.storage.internals.checkpoint.{LazyOffsetCheckpoints, OffsetCheckpoints, OffsetCheckpointFile}
+import org.apache.kafka.storage.internals.checkpoint.{LazyOffsetCheckpoints, OffsetCheckpointFile, OffsetCheckpoints}
import org.apache.kafka.storage.internals.log.{AppendOrigin, FetchDataInfo, FetchParams, FetchPartitionData, LeaderHwChange, LogAppendInfo, LogConfig, LogDirFailureChannel, LogOffsetMetadata, LogReadInfo, RecordValidationException, RemoteLogReadResult, RemoteStorageFetchInfo, VerificationGuard}
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import java.io.File
import java.lang.{Long => JLong}
diff --git a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
index 302b6adfe48..66d5d1726ac 100644
--- a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
+++ b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
@@ -21,7 +21,6 @@ import kafka.cluster.Partition;
import kafka.log.UnifiedLog;
import kafka.log.remote.quota.RLMQuotaManager;
import kafka.log.remote.quota.RLMQuotaManagerConfig;
-import kafka.server.BrokerTopicStats;
import kafka.server.KafkaConfig;
import kafka.server.StopPartition;
@@ -77,6 +76,7 @@ import org.apache.kafka.storage.internals.log.ProducerStateManager;
import org.apache.kafka.storage.internals.log.RemoteStorageFetchInfo;
import org.apache.kafka.storage.internals.log.TimeIndex;
import org.apache.kafka.storage.internals.log.TransactionIndex;
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
import org.apache.kafka.test.TestUtils;
import com.yammer.metrics.core.Gauge;
diff --git a/core/src/test/java/kafka/log/remote/RemoteLogReaderTest.java b/core/src/test/java/kafka/log/remote/RemoteLogReaderTest.java
index befc7a56bb5..9bf2b8b019a 100644
--- a/core/src/test/java/kafka/log/remote/RemoteLogReaderTest.java
+++ b/core/src/test/java/kafka/log/remote/RemoteLogReaderTest.java
@@ -17,7 +17,6 @@
package kafka.log.remote;
import kafka.log.remote.quota.RLMQuotaManager;
-import kafka.server.BrokerTopicStats;
import kafka.utils.TestUtils;
import org.apache.kafka.common.TopicPartition;
@@ -27,6 +26,7 @@ import org.apache.kafka.storage.internals.log.FetchDataInfo;
import org.apache.kafka.storage.internals.log.LogOffsetMetadata;
import org.apache.kafka.storage.internals.log.RemoteLogReadResult;
import org.apache.kafka.storage.internals.log.RemoteStorageFetchInfo;
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
import com.yammer.metrics.core.Timer;
diff --git a/core/src/test/scala/kafka/server/KafkaRequestHandlerTest.scala b/core/src/test/scala/kafka/server/KafkaRequestHandlerTest.scala
index adcd8a13ef3..8b9bca94c3d 100644
--- a/core/src/test/scala/kafka/server/KafkaRequestHandlerTest.scala
+++ b/core/src/test/scala/kafka/server/KafkaRequestHandlerTest.scala
@@ -27,7 +27,7 @@ import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol}
import org.apache.kafka.common.utils.{BufferSupplier, MockTime, Time}
import org.apache.kafka.server.log.remote.storage.RemoteStorageMetrics
import org.apache.kafka.server.metrics.KafkaYammerMetrics
-import org.apache.kafka.storage.log.metrics.BrokerTopicMetrics
+import org.apache.kafka.storage.log.metrics.{BrokerTopicMetrics, BrokerTopicStats}
import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue}
import org.junit.jupiter.api.Test
import org.junit.jupiter.params.ParameterizedTest
diff --git a/core/src/test/scala/other/kafka/StressTestLog.scala b/core/src/test/scala/other/kafka/StressTestLog.scala
index da3a680ef89..78fda569a36 100755
--- a/core/src/test/scala/other/kafka/StressTestLog.scala
+++ b/core/src/test/scala/other/kafka/StressTestLog.scala
@@ -20,7 +20,6 @@ package kafka
import java.util.Properties
import java.util.concurrent.atomic._
import kafka.log._
-import kafka.server.BrokerTopicStats
import kafka.utils._
import org.apache.kafka.clients.consumer.OffsetOutOfRangeException
import org.apache.kafka.common.config.TopicConfig
@@ -29,6 +28,7 @@ import org.apache.kafka.common.utils.{Exit, Utils}
import org.apache.kafka.coordinator.transaction.TransactionLogConfigs
import org.apache.kafka.server.util.MockTime
import org.apache.kafka.storage.internals.log.{FetchIsolation, LogConfig, LogDirFailureChannel, ProducerStateManagerConfig}
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
/**
* A stress test that instantiates a log and then runs continual appends against it from one thread and continual reads against it
diff --git a/core/src/test/scala/other/kafka/TestLinearWriteSpeed.scala b/core/src/test/scala/other/kafka/TestLinearWriteSpeed.scala
index 9dc8a16774b..0811cec64c5 100755
--- a/core/src/test/scala/other/kafka/TestLinearWriteSpeed.scala
+++ b/core/src/test/scala/other/kafka/TestLinearWriteSpeed.scala
@@ -24,7 +24,6 @@ import java.nio.file.StandardOpenOption
import java.util.{Properties, Random}
import joptsimple._
import kafka.log._
-import kafka.server.BrokerTopicStats
import org.apache.kafka.common.compress.{Compression, GzipCompression, Lz4Compression, ZstdCompression}
import org.apache.kafka.common.config.TopicConfig
import org.apache.kafka.common.record._
@@ -33,6 +32,7 @@ import org.apache.kafka.coordinator.transaction.TransactionLogConfigs
import org.apache.kafka.server.util.{KafkaScheduler, Scheduler}
import org.apache.kafka.server.util.CommandLineUtils
import org.apache.kafka.storage.internals.log.{LogConfig, LogDirFailureChannel, ProducerStateManagerConfig}
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import scala.math.max
diff --git a/core/src/test/scala/unit/kafka/cluster/PartitionLockTest.scala b/core/src/test/scala/unit/kafka/cluster/PartitionLockTest.scala
index 2cad8b47828..1f70ccfccd3 100644
--- a/core/src/test/scala/unit/kafka/cluster/PartitionLockTest.scala
+++ b/core/src/test/scala/unit/kafka/cluster/PartitionLockTest.scala
@@ -41,6 +41,7 @@ import org.apache.kafka.server.util.MockTime
import org.apache.kafka.storage.internals.checkpoint.OffsetCheckpoints
import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache
import org.apache.kafka.storage.internals.log.{AppendOrigin, CleanerConfig, FetchIsolation, FetchParams, LogAppendInfo, LogConfig, LogDirFailureChannel, LogSegments, ProducerStateManager, ProducerStateManagerConfig, VerificationGuard}
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue}
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
import org.mockito.ArgumentMatchers
diff --git a/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala b/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala
index 22894f8d308..1d5e2677c67 100644
--- a/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala
+++ b/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala
@@ -62,6 +62,7 @@ import org.apache.kafka.server.util.{KafkaScheduler, MockTime}
import org.apache.kafka.storage.internals.checkpoint.OffsetCheckpoints
import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache
import org.apache.kafka.storage.internals.log.{AppendOrigin, CleanerConfig, EpochEntry, FetchIsolation, FetchParams, LogAppendInfo, LogDirFailureChannel, LogOffsetMetadata, LogReadInfo, LogSegments, LogStartOffsetIncrementReason, ProducerStateManager, ProducerStateManagerConfig, VerificationGuard}
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.ValueSource
diff --git a/core/src/test/scala/unit/kafka/log/AbstractLogCleanerIntegrationTest.scala b/core/src/test/scala/unit/kafka/log/AbstractLogCleanerIntegrationTest.scala
index ae35d4a58a8..ff94377b2f1 100644
--- a/core/src/test/scala/unit/kafka/log/AbstractLogCleanerIntegrationTest.scala
+++ b/core/src/test/scala/unit/kafka/log/AbstractLogCleanerIntegrationTest.scala
@@ -19,7 +19,6 @@ package kafka.log
import java.io.File
import java.nio.file.Files
import java.util.Properties
-import kafka.server.BrokerTopicStats
import kafka.utils.{Pool, TestUtils}
import kafka.utils.Implicits._
import org.apache.kafka.common.TopicPartition
@@ -30,6 +29,7 @@ import org.apache.kafka.common.utils.Utils
import org.apache.kafka.coordinator.transaction.TransactionLogConfigs
import org.apache.kafka.server.util.MockTime
import org.apache.kafka.storage.internals.log.{CleanerConfig, LogConfig, LogDirFailureChannel, ProducerStateManagerConfig}
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import org.junit.jupiter.api.{AfterEach, Tag}
import scala.collection.Seq
diff --git a/core/src/test/scala/unit/kafka/log/BrokerCompressionTest.scala b/core/src/test/scala/unit/kafka/log/BrokerCompressionTest.scala
index 07e70e97f0b..1495d688d5c 100755
--- a/core/src/test/scala/unit/kafka/log/BrokerCompressionTest.scala
+++ b/core/src/test/scala/unit/kafka/log/BrokerCompressionTest.scala
@@ -17,7 +17,6 @@
package kafka.log
-import kafka.server.BrokerTopicStats
import kafka.utils._
import org.apache.kafka.common.compress.Compression
import org.apache.kafka.common.config.TopicConfig
@@ -27,6 +26,7 @@ import org.apache.kafka.coordinator.transaction.TransactionLogConfigs
import org.apache.kafka.server.record.BrokerCompressionType
import org.apache.kafka.server.util.MockTime
import org.apache.kafka.storage.internals.log.{FetchIsolation, LogConfig, LogDirFailureChannel, ProducerStateManagerConfig}
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api._
import org.junit.jupiter.params.ParameterizedTest
diff --git a/core/src/test/scala/unit/kafka/log/LogCleanerManagerTest.scala b/core/src/test/scala/unit/kafka/log/LogCleanerManagerTest.scala
index 9fc5629ed86..978b856a9fa 100644
--- a/core/src/test/scala/unit/kafka/log/LogCleanerManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogCleanerManagerTest.scala
@@ -20,7 +20,6 @@ package kafka.log
import java.io.File
import java.nio.file.Files
import java.util.Properties
-import kafka.server.BrokerTopicStats
import kafka.utils._
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.compress.Compression
@@ -30,6 +29,7 @@ import org.apache.kafka.common.utils.Utils
import org.apache.kafka.coordinator.transaction.TransactionLogConfigs
import org.apache.kafka.server.util.MockTime
import org.apache.kafka.storage.internals.log.{AppendOrigin, LogConfig, LogDirFailureChannel, LogSegment, LogSegments, LogStartOffsetIncrementReason, ProducerStateManager, ProducerStateManagerConfig}
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, Test}
diff --git a/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala b/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
index ee62b3f0f21..f8350cf7427 100644
--- a/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
@@ -18,7 +18,7 @@
package kafka.log
import kafka.common._
-import kafka.server.{BrokerTopicStats, KafkaConfig}
+import kafka.server.KafkaConfig
import kafka.utils.{CoreUtils, Logging, Pool, TestUtils}
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.compress.Compression
@@ -31,6 +31,7 @@ import org.apache.kafka.server.metrics.{KafkaMetricsGroup, KafkaYammerMetrics}
import org.apache.kafka.server.util.MockTime
import org.apache.kafka.storage.internals.log.{AbortedTxn, AppendOrigin, CleanerConfig, LogAppendInfo, LogConfig, LogDirFailureChannel, LogFileUtils, LogSegment, LogSegments, LogStartOffsetIncrementReason, OffsetMap, ProducerStateManager, ProducerStateManagerConfig}
import org.apache.kafka.storage.internals.utils.Throttler
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, Test}
import org.mockito.ArgumentMatchers
diff --git a/core/src/test/scala/unit/kafka/log/LogConcurrencyTest.scala b/core/src/test/scala/unit/kafka/log/LogConcurrencyTest.scala
index 65187b707ad..0793768bfbd 100644
--- a/core/src/test/scala/unit/kafka/log/LogConcurrencyTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogConcurrencyTest.scala
@@ -19,7 +19,6 @@ package kafka.log
import java.util.Properties
import java.util.concurrent.{Callable, Executors}
-import kafka.server.BrokerTopicStats
import kafka.utils.TestUtils
import org.apache.kafka.common.config.TopicConfig
import org.apache.kafka.common.record.SimpleRecord
@@ -27,6 +26,7 @@ import org.apache.kafka.common.utils.{Time, Utils}
import org.apache.kafka.coordinator.transaction.TransactionLogConfigs
import org.apache.kafka.server.util.KafkaScheduler
import org.apache.kafka.storage.internals.log.{FetchIsolation, LogConfig, LogDirFailureChannel, ProducerStateManagerConfig}
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
diff --git a/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala b/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala
index 35cb8b79b29..5c0bcd3b4d7 100644
--- a/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala
@@ -21,7 +21,7 @@ import java.io.{BufferedWriter, File, FileWriter, IOException}
import java.nio.ByteBuffer
import java.nio.file.{Files, NoSuchFileException, Paths}
import java.util.{Optional, OptionalLong, Properties}
-import kafka.server.{BrokerTopicStats, KafkaConfig}
+import kafka.server.KafkaConfig
import kafka.server.metadata.MockConfigRepository
import kafka.utils.TestUtils
import org.apache.kafka.common.TopicPartition
@@ -37,6 +37,7 @@ import org.apache.kafka.server.util.{MockTime, Scheduler}
import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache
import org.apache.kafka.storage.internals.log.{AbortedTxn, CleanerConfig, EpochEntry, LogConfig, LogDirFailureChannel, LogFileUtils, LogOffsetMetadata, LogSegment, LogSegments, LogStartOffsetIncrementReason, OffsetIndex, ProducerStateManager, ProducerStateManagerConfig, SnapshotFile}
import org.apache.kafka.storage.internals.checkpoint.CleanShutdownFileHandler
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import org.junit.jupiter.api.Assertions.{assertDoesNotThrow, assertEquals, assertFalse, assertNotEquals, assertThrows, assertTrue}
import org.junit.jupiter.api.function.Executable
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
diff --git a/core/src/test/scala/unit/kafka/log/LogManagerTest.scala b/core/src/test/scala/unit/kafka/log/LogManagerTest.scala
index 7bf04874f78..9e96ec16bdc 100755
--- a/core/src/test/scala/unit/kafka/log/LogManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogManagerTest.scala
@@ -19,7 +19,6 @@ package kafka.log
import com.yammer.metrics.core.{Gauge, MetricName}
import kafka.server.metadata.{ConfigRepository, MockConfigRepository}
-import kafka.server.BrokerTopicStats
import kafka.utils._
import org.apache.directory.api.util.FileUtils
import org.apache.kafka.common.config.TopicConfig
@@ -50,6 +49,7 @@ import org.apache.kafka.server.metrics.KafkaYammerMetrics
import org.apache.kafka.server.util.{FileLock, KafkaScheduler, MockTime, Scheduler}
import org.apache.kafka.storage.internals.log.{CleanerConfig, FetchDataInfo, FetchIsolation, LogConfig, LogDirFailureChannel, LogStartOffsetIncrementReason, ProducerStateManagerConfig, RemoteIndexCache}
import org.apache.kafka.storage.internals.checkpoint.{CleanShutdownFileHandler, OffsetCheckpointFile}
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import org.junit.jupiter.api.function.Executable
import java.time.Duration
diff --git a/core/src/test/scala/unit/kafka/log/LogTestUtils.scala b/core/src/test/scala/unit/kafka/log/LogTestUtils.scala
index 264dfb36cfe..28037861253 100644
--- a/core/src/test/scala/unit/kafka/log/LogTestUtils.scala
+++ b/core/src/test/scala/unit/kafka/log/LogTestUtils.scala
@@ -21,7 +21,6 @@ import kafka.log.remote.RemoteLogManager
import java.io.File
import java.util.Properties
-import kafka.server.BrokerTopicStats
import kafka.utils.TestUtils
import org.apache.kafka.common.Uuid
import org.apache.kafka.common.compress.Compression
@@ -38,6 +37,7 @@ import org.apache.kafka.server.util.Scheduler
import org.apache.kafka.storage.internals.checkpoint.LeaderEpochCheckpointFile
import org.apache.kafka.storage.internals.log.LogConfig.{DEFAULT_REMOTE_LOG_COPY_DISABLE_CONFIG, DEFAULT_REMOTE_LOG_DELETE_ON_DISABLE_CONFIG}
import org.apache.kafka.storage.internals.log.{AbortedTxn, AppendOrigin, FetchDataInfo, FetchIsolation, LazyIndex, LogAppendInfo, LogConfig, LogDirFailureChannel, LogFileUtils, LogOffsetsListener, LogSegment, ProducerStateManager, ProducerStateManagerConfig, TransactionIndex}
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import scala.jdk.CollectionConverters._
diff --git a/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala b/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala
index 448f017a2d1..3c5a8ab468b 100755
--- a/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala
+++ b/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala
@@ -19,7 +19,7 @@ package kafka.log
import kafka.common.{OffsetsOutOfOrderException, UnexpectedAppendOffsetException}
import kafka.log.remote.RemoteLogManager
-import kafka.server.{BrokerTopicStats, KafkaConfig}
+import kafka.server.KafkaConfig
import kafka.utils.TestUtils
import org.apache.kafka.common.compress.Compression
import org.apache.kafka.common.config.TopicConfig
@@ -40,7 +40,7 @@ import org.apache.kafka.storage.internals.checkpoint.{LeaderEpochCheckpointFile,
import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache
import org.apache.kafka.storage.internals.log.{AbortedTxn, AppendOrigin, EpochEntry, FetchIsolation, LogConfig, LogFileUtils, LogOffsetMetadata, LogOffsetSnapshot, LogOffsetsListener, LogSegment, LogSegments, LogStartOffsetIncrementReason, ProducerStateManager, ProducerStateManagerConfig, RecordValidationException, VerificationGuard}
import org.apache.kafka.storage.internals.utils.Throttler
-import org.apache.kafka.storage.log.metrics.BrokerTopicMetrics
+import org.apache.kafka.storage.log.metrics.{BrokerTopicMetrics, BrokerTopicStats}
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
import org.junit.jupiter.params.ParameterizedTest
diff --git a/core/src/test/scala/unit/kafka/server/AbstractFetcherManagerTest.scala b/core/src/test/scala/unit/kafka/server/AbstractFetcherManagerTest.scala
index f9750618ae0..45086961342 100644
--- a/core/src/test/scala/unit/kafka/server/AbstractFetcherManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/AbstractFetcherManagerTest.scala
@@ -29,6 +29,7 @@ import org.apache.kafka.common.{TopicPartition, Uuid}
import org.apache.kafka.server.common.OffsetAndEpoch
import org.apache.kafka.server.metrics.KafkaYammerMetrics
import org.apache.kafka.storage.internals.log.LogAppendInfo
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{BeforeEach, Test}
import org.mockito.Mockito.{mock, verify, when}
diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
index c6666630cde..87d824ee4f5 100644
--- a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
@@ -90,6 +90,7 @@ import org.apache.kafka.server.share.{CachedSharePartition, ErroneousAndValidPar
import org.apache.kafka.server.quota.ThrottleCallback
import org.apache.kafka.server.util.{FutureUtils, MockTime}
import org.apache.kafka.storage.internals.log.{AppendOrigin, FetchParams, FetchPartitionData, LogConfig}
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, Test}
import org.junit.jupiter.params.ParameterizedTest
diff --git a/core/src/test/scala/unit/kafka/server/MockFetcherThread.scala b/core/src/test/scala/unit/kafka/server/MockFetcherThread.scala
index 30528a6729d..304fd0de970 100644
--- a/core/src/test/scala/unit/kafka/server/MockFetcherThread.scala
+++ b/core/src/test/scala/unit/kafka/server/MockFetcherThread.scala
@@ -24,6 +24,7 @@ import org.apache.kafka.common.utils.Time
import org.apache.kafka.server.common.OffsetAndEpoch
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.storage.internals.log.LogAppendInfo
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import org.junit.jupiter.api.Assertions._
import java.util.OptionalInt
diff --git a/core/src/test/scala/unit/kafka/server/ReplicaAlterLogDirsThreadTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaAlterLogDirsThreadTest.scala
index f648095271d..cfdf37ef50f 100644
--- a/core/src/test/scala/unit/kafka/server/ReplicaAlterLogDirsThreadTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ReplicaAlterLogDirsThreadTest.scala
@@ -34,6 +34,7 @@ import org.apache.kafka.common.{TopicIdPartition, TopicPartition, Uuid}
import org.apache.kafka.server.common
import org.apache.kafka.server.common.{DirectoryEventHandler, MetadataVersion, OffsetAndEpoch}
import org.apache.kafka.storage.internals.log.{FetchIsolation, FetchParams, FetchPartitionData}
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.Test
import org.mockito.ArgumentMatchers.{any, anyBoolean}
diff --git a/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala
index 4e2f3285fb8..81fc5ca422b 100644
--- a/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala
@@ -39,6 +39,7 @@ import org.apache.kafka.server.config.ReplicationConfigs
import org.apache.kafka.server.common.{MetadataVersion, OffsetAndEpoch}
import org.apache.kafka.server.common.MetadataVersion.IBP_2_6_IV0
import org.apache.kafka.storage.internals.log.LogAppendInfo
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, Test}
import org.junit.jupiter.params.ParameterizedTest
diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
index 857e657b602..29c4c602c17 100644
--- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
@@ -68,6 +68,7 @@ import org.apache.kafka.server.util.timer.MockTimer
import org.apache.kafka.server.util.{MockScheduler, MockTime}
import org.apache.kafka.storage.internals.checkpoint.{LazyOffsetCheckpoints, OffsetCheckpointFile, PartitionMetadataFile}
import org.apache.kafka.storage.internals.log._
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterAll, AfterEach, BeforeEach, Test}
import org.junit.jupiter.params.ParameterizedTest
diff --git a/core/src/test/scala/unit/kafka/tools/DumpLogSegmentsTest.scala b/core/src/test/scala/unit/kafka/tools/DumpLogSegmentsTest.scala
index b0c9015f57e..04d42dadd75 100644
--- a/core/src/test/scala/unit/kafka/tools/DumpLogSegmentsTest.scala
+++ b/core/src/test/scala/unit/kafka/tools/DumpLogSegmentsTest.scala
@@ -26,7 +26,7 @@ import java.util.Properties
import java.util.stream.IntStream
import kafka.log.{LogTestUtils, UnifiedLog}
import kafka.raft.{KafkaMetadataLog, MetadataLogConfig}
-import kafka.server.{BrokerTopicStats, KafkaRaftServer}
+import kafka.server.KafkaRaftServer
import kafka.tools.DumpLogSegments.{OffsetsMessageParser, TimeIndexDumpErrors}
import kafka.utils.{TestUtils, VerifiableProperties}
import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor.{Assignment, Subscription}
@@ -51,6 +51,7 @@ import org.apache.kafka.server.log.remote.storage.{RemoteLogSegmentId, RemoteLog
import org.apache.kafka.server.util.MockTime
import org.apache.kafka.snapshot.RecordsSnapshotWriter
import org.apache.kafka.storage.internals.log.{AppendOrigin, FetchIsolation, LogConfig, LogDirFailureChannel, ProducerStateManagerConfig}
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, Test}
diff --git a/core/src/test/scala/unit/kafka/utils/SchedulerTest.scala b/core/src/test/scala/unit/kafka/utils/SchedulerTest.scala
index 6280318af5d..917f53eed60 100644
--- a/core/src/test/scala/unit/kafka/utils/SchedulerTest.scala
+++ b/core/src/test/scala/unit/kafka/utils/SchedulerTest.scala
@@ -20,11 +20,11 @@ import java.util.Properties
import java.util.concurrent.atomic._
import java.util.concurrent.{CountDownLatch, Executors, TimeUnit}
import kafka.log.{LocalLog, LogLoader, UnifiedLog}
-import kafka.server.BrokerTopicStats
import kafka.utils.TestUtils.retry
import org.apache.kafka.coordinator.transaction.TransactionLogConfigs
import org.apache.kafka.server.util.{KafkaScheduler, MockTime}
import org.apache.kafka.storage.internals.log.{LogConfig, LogDirFailureChannel, LogSegments, ProducerStateManager, ProducerStateManagerConfig}
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test, Timeout}
diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
index 546844fd4b3..c29769e99ed 100755
--- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala
+++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
@@ -65,6 +65,7 @@ import org.apache.kafka.server.util.timer.SystemTimer
import org.apache.kafka.server.{ClientMetricsManager, ControllerRequestCompletionHandler}
import org.apache.kafka.storage.internals.checkpoint.OffsetCheckpointFile
import org.apache.kafka.storage.internals.log.{CleanerConfig, LogConfig, LogDirFailureChannel, ProducerStateManagerConfig}
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats
import org.apache.kafka.test.{TestSslUtils, TestUtils => JTestUtils}
import org.apache.zookeeper.KeeperException.SessionExpiredException
import org.apache.zookeeper.ZooDefs._
diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/fetcher/ReplicaFetcherThreadBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/fetcher/ReplicaFetcherThreadBenchmark.java
index af08d7d5540..2f8f762145f 100644
--- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/fetcher/ReplicaFetcherThreadBenchmark.java
+++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/fetcher/ReplicaFetcherThreadBenchmark.java
@@ -25,7 +25,6 @@ import kafka.log.LogManager;
import kafka.server.AlterPartitionManager;
import kafka.server.BrokerBlockingSender;
import kafka.server.BrokerFeatures;
-import kafka.server.BrokerTopicStats;
import kafka.server.FailedPartitions;
import kafka.server.InitialFetchState;
import kafka.server.KafkaConfig;
@@ -73,6 +72,7 @@ import org.apache.kafka.storage.internals.log.CleanerConfig;
import org.apache.kafka.storage.internals.log.LogAppendInfo;
import org.apache.kafka.storage.internals.log.LogConfig;
import org.apache.kafka.storage.internals.log.LogDirFailureChannel;
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
import org.mockito.Mockito;
import org.openjdk.jmh.annotations.Benchmark;
diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/metadata/KRaftMetadataRequestBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/metadata/KRaftMetadataRequestBenchmark.java
index d8fd1720699..11eb714886f 100644
--- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/metadata/KRaftMetadataRequestBenchmark.java
+++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/metadata/KRaftMetadataRequestBenchmark.java
@@ -21,7 +21,6 @@ import kafka.coordinator.transaction.TransactionCoordinator;
import kafka.network.RequestChannel;
import kafka.network.RequestConvertToJson;
import kafka.server.AutoTopicCreationManager;
-import kafka.server.BrokerTopicStats;
import kafka.server.ClientQuotaManager;
import kafka.server.ClientRequestQuotaManager;
import kafka.server.ControllerMutationQuotaManager;
@@ -64,6 +63,7 @@ import org.apache.kafka.server.common.FinalizedFeatures;
import org.apache.kafka.server.common.KRaftVersion;
import org.apache.kafka.server.common.MetadataVersion;
import org.apache.kafka.server.config.KRaftConfigs;
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
import org.mockito.Mockito;
import org.openjdk.jmh.annotations.Benchmark;
diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/metadata/MetadataRequestBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/metadata/MetadataRequestBenchmark.java
index 0a0bcfc6705..939ecf3e192 100644
--- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/metadata/MetadataRequestBenchmark.java
+++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/metadata/MetadataRequestBenchmark.java
@@ -23,7 +23,6 @@ import kafka.network.RequestChannel;
import kafka.network.RequestConvertToJson;
import kafka.server.AutoTopicCreationManager;
import kafka.server.BrokerFeatures;
-import kafka.server.BrokerTopicStats;
import kafka.server.ClientQuotaManager;
import kafka.server.ClientRequestQuotaManager;
import kafka.server.ControllerMutationQuotaManager;
@@ -64,6 +63,7 @@ import org.apache.kafka.server.common.FinalizedFeatures;
import org.apache.kafka.server.common.MetadataVersion;
import org.apache.kafka.server.config.ServerConfigs;
import org.apache.kafka.server.config.ZkConfigs;
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
import org.mockito.Mockito;
import org.openjdk.jmh.annotations.Benchmark;
diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/PartitionMakeFollowerBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/PartitionMakeFollowerBenchmark.java
index 6184bb20b94..be5fae09335 100644
--- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/PartitionMakeFollowerBenchmark.java
+++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/PartitionMakeFollowerBenchmark.java
@@ -22,7 +22,6 @@ import kafka.cluster.DelayedOperations;
import kafka.cluster.Partition;
import kafka.log.LogManager;
import kafka.server.AlterPartitionManager;
-import kafka.server.BrokerTopicStats;
import kafka.server.MetadataCache;
import kafka.server.builders.LogManagerBuilder;
import kafka.server.metadata.MockConfigRepository;
@@ -41,6 +40,7 @@ import org.apache.kafka.storage.internals.checkpoint.OffsetCheckpoints;
import org.apache.kafka.storage.internals.log.CleanerConfig;
import org.apache.kafka.storage.internals.log.LogConfig;
import org.apache.kafka.storage.internals.log.LogDirFailureChannel;
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
import org.mockito.Mockito;
import org.openjdk.jmh.annotations.Benchmark;
diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/UpdateFollowerFetchStateBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/UpdateFollowerFetchStateBenchmark.java
index 917018cfc82..88d7263e806 100644
--- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/UpdateFollowerFetchStateBenchmark.java
+++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/partition/UpdateFollowerFetchStateBenchmark.java
@@ -23,7 +23,6 @@ import kafka.cluster.Partition;
import kafka.cluster.Replica;
import kafka.log.LogManager;
import kafka.server.AlterPartitionManager;
-import kafka.server.BrokerTopicStats;
import kafka.server.MetadataCache;
import kafka.server.builders.LogManagerBuilder;
import kafka.server.metadata.MockConfigRepository;
@@ -39,6 +38,7 @@ import org.apache.kafka.storage.internals.log.CleanerConfig;
import org.apache.kafka.storage.internals.log.LogConfig;
import org.apache.kafka.storage.internals.log.LogDirFailureChannel;
import org.apache.kafka.storage.internals.log.LogOffsetMetadata;
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
import org.mockito.Mockito;
import org.openjdk.jmh.annotations.Benchmark;
diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/record/BaseRecordBatchBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/record/BaseRecordBatchBenchmark.java
index 44d72720684..01e085d5866 100644
--- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/record/BaseRecordBatchBenchmark.java
+++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/record/BaseRecordBatchBenchmark.java
@@ -17,7 +17,6 @@
package org.apache.kafka.jmh.record;
import kafka.log.UnifiedLog;
-import kafka.server.BrokerTopicStats;
import kafka.server.RequestLocal;
import org.apache.kafka.common.compress.Compression;
@@ -29,6 +28,7 @@ import org.apache.kafka.common.record.Record;
import org.apache.kafka.common.record.RecordBatch;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.storage.internals.log.LogValidator;
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
import org.openjdk.jmh.annotations.Param;
import org.openjdk.jmh.annotations.Scope;
diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/CheckpointBench.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/CheckpointBench.java
index d78b48302f1..1edbca20eaa 100644
--- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/CheckpointBench.java
+++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/CheckpointBench.java
@@ -20,7 +20,6 @@ import kafka.cluster.Partition;
import kafka.log.LogManager;
import kafka.server.AlterPartitionManager;
import kafka.server.BrokerFeatures;
-import kafka.server.BrokerTopicStats;
import kafka.server.KafkaConfig;
import kafka.server.MetadataCache;
import kafka.server.QuotaFactory;
@@ -42,6 +41,7 @@ import org.apache.kafka.storage.internals.checkpoint.OffsetCheckpoints;
import org.apache.kafka.storage.internals.log.CleanerConfig;
import org.apache.kafka.storage.internals.log.LogConfig;
import org.apache.kafka.storage.internals.log.LogDirFailureChannel;
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.Fork;
diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/PartitionCreationBench.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/PartitionCreationBench.java
index 5ca9f707c4b..c2411124bb7 100644
--- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/PartitionCreationBench.java
+++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/PartitionCreationBench.java
@@ -20,7 +20,6 @@ import kafka.cluster.Partition;
import kafka.log.LogManager;
import kafka.server.AlterPartitionManager;
import kafka.server.BrokerFeatures;
-import kafka.server.BrokerTopicStats;
import kafka.server.KafkaConfig;
import kafka.server.MetadataCache;
import kafka.server.QuotaFactory;
@@ -45,6 +44,7 @@ import org.apache.kafka.storage.internals.checkpoint.OffsetCheckpoints;
import org.apache.kafka.storage.internals.log.CleanerConfig;
import org.apache.kafka.storage.internals.log.LogConfig;
import org.apache.kafka.storage.internals.log.LogDirFailureChannel;
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
diff --git a/storage/src/main/java/org/apache/kafka/storage/log/metrics/BrokerTopicStats.java b/storage/src/main/java/org/apache/kafka/storage/log/metrics/BrokerTopicStats.java
new file mode 100644
index 00000000000..3b5f8c900e3
--- /dev/null
+++ b/storage/src/main/java/org/apache/kafka/storage/log/metrics/BrokerTopicStats.java
@@ -0,0 +1,265 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.storage.log.metrics;
+
+import org.apache.kafka.server.log.remote.storage.RemoteStorageMetrics;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+public class BrokerTopicStats {
+
+ private static final Logger LOG = LoggerFactory.getLogger(BrokerTopicStats.class);
+
+ private final boolean remoteStorageEnabled;
+ private final BrokerTopicMetrics allTopicsStats;
+ private final ConcurrentMap<String, BrokerTopicMetrics> stats;
+
+ public BrokerTopicStats() {
+ this(false);
+ }
+
+ public BrokerTopicStats(boolean remoteStorageEnabled) {
+ this.remoteStorageEnabled = remoteStorageEnabled;
+ this.allTopicsStats = new BrokerTopicMetrics(remoteStorageEnabled);
+ this.stats = new ConcurrentHashMap<>();
+ }
+
+ public boolean isTopicStatsExisted(String topic) {
+ return stats.containsKey(topic);
+ }
+
+ public BrokerTopicMetrics topicStats(String topic) {
+ return stats.computeIfAbsent(topic, k -> new BrokerTopicMetrics(k, remoteStorageEnabled));
+ }
+
+ public void updateReplicationBytesIn(long value) {
+ allTopicsStats.replicationBytesInRate().ifPresent(metric -> metric.mark(value));
+ }
+
+ private void updateReplicationBytesOut(long value) {
+ allTopicsStats.replicationBytesOutRate().ifPresent(metric -> metric.mark(value));
+ }
+
+ public void updateReassignmentBytesIn(long value) {
+ allTopicsStats.reassignmentBytesInPerSec().ifPresent(metric -> metric.mark(value));
+ }
+
+ private void updateReassignmentBytesOut(long value) {
+ allTopicsStats.reassignmentBytesOutPerSec().ifPresent(metric -> metric.mark(value));
+ }
+
+ // This method only removes metrics only used for leader
+ public void removeOldLeaderMetrics(String topic) {
+ BrokerTopicMetrics topicMetrics = topicStats(topic);
+ if (topicMetrics != null) {
+ topicMetrics.closeMetric(BrokerTopicMetrics.MESSAGE_IN_PER_SEC);
+ topicMetrics.closeMetric(BrokerTopicMetrics.BYTES_IN_PER_SEC);
+ topicMetrics.closeMetric(BrokerTopicMetrics.BYTES_REJECTED_PER_SEC);
+ topicMetrics.closeMetric(BrokerTopicMetrics.FAILED_PRODUCE_REQUESTS_PER_SEC);
+ topicMetrics.closeMetric(BrokerTopicMetrics.TOTAL_PRODUCE_REQUESTS_PER_SEC);
+ topicMetrics.closeMetric(BrokerTopicMetrics.PRODUCE_MESSAGE_CONVERSIONS_PER_SEC);
+ topicMetrics.closeMetric(BrokerTopicMetrics.REPLICATION_BYTES_OUT_PER_SEC);
+ topicMetrics.closeMetric(BrokerTopicMetrics.REASSIGNMENT_BYTES_OUT_PER_SEC);
+ topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_COPY_BYTES_PER_SEC_METRIC.getName());
+ topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_FETCH_BYTES_PER_SEC_METRIC.getName());
+ topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_FETCH_REQUESTS_PER_SEC_METRIC.getName());
+ topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_COPY_REQUESTS_PER_SEC_METRIC.getName());
+ topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_DELETE_REQUESTS_PER_SEC_METRIC.getName());
+ topicMetrics.closeMetric(RemoteStorageMetrics.BUILD_REMOTE_LOG_AUX_STATE_REQUESTS_PER_SEC_METRIC.getName());
+ topicMetrics.closeMetric(RemoteStorageMetrics.FAILED_REMOTE_FETCH_PER_SEC_METRIC.getName());
+ topicMetrics.closeMetric(RemoteStorageMetrics.FAILED_REMOTE_COPY_PER_SEC_METRIC.getName());
+ topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_COPY_LAG_BYTES_METRIC.getName());
+ topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_LOG_METADATA_COUNT_METRIC.getName());
+ topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_LOG_SIZE_COMPUTATION_TIME_METRIC.getName());
+ topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_LOG_SIZE_BYTES_METRIC.getName());
+ topicMetrics.closeMetric(RemoteStorageMetrics.FAILED_REMOTE_DELETE_PER_SEC_METRIC.getName());
+ topicMetrics.closeMetric(RemoteStorageMetrics.FAILED_BUILD_REMOTE_LOG_AUX_STATE_PER_SEC_METRIC.getName());
+ topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_COPY_LAG_BYTES_METRIC.getName());
+ topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_COPY_LAG_SEGMENTS_METRIC.getName());
+ topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_DELETE_LAG_BYTES_METRIC.getName());
+ topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_DELETE_LAG_SEGMENTS_METRIC.getName());
+ }
+ }
+
+ // This method only removes metrics only used for follower
+ public void removeOldFollowerMetrics(String topic) {
+ BrokerTopicMetrics topicMetrics = topicStats(topic);
+ if (topicMetrics != null) {
+ topicMetrics.closeMetric(BrokerTopicMetrics.REPLICATION_BYTES_IN_PER_SEC);
+ topicMetrics.closeMetric(BrokerTopicMetrics.REASSIGNMENT_BYTES_IN_PER_SEC);
+ }
+ }
+
+ public void removeMetrics(String topic) {
+ BrokerTopicMetrics metrics = stats.remove(topic);
+ if (metrics != null) {
+ metrics.close();
+ }
+ }
+
+ public void updateBytesOut(String topic, boolean isFollower, boolean isReassignment, long value) {
+ if (isFollower) {
+ if (isReassignment) {
+ updateReassignmentBytesOut(value);
+ }
+ updateReplicationBytesOut(value);
+ } else {
+ topicStats(topic).bytesOutRate().mark(value);
+ allTopicsStats.bytesOutRate().mark(value);
+ }
+ }
+
+ // Update the broker-level all topic metric values so that we have a sample right for all topics metric after update of partition.
+ public void recordRemoteCopyLagBytes(String topic, int partition, long value) {
+ BrokerTopicMetrics 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.
+ public void removeRemoteCopyLagBytes(String topic, int partition) {
+ BrokerTopicMetrics topicMetric = topicStats(topic);
+ topicMetric.remoteCopyLagBytesAggrMetric().removeKey(String.valueOf(partition));
+ allTopicsStats.remoteCopyLagBytesAggrMetric().setValue(topic, topicMetric.remoteCopyLagBytes());
+ }
+
+ public void removeBrokerLevelRemoteCopyLagBytes(String topic) {
+ allTopicsStats.remoteCopyLagBytesAggrMetric().removeKey(topic);
+ }
+
+ // Update the broker-level all topic metric values so that we have a sample right for all topics metric after update of partition.
+ public void recordRemoteCopyLagSegments(String topic, int partition, long value) {
+ BrokerTopicMetrics topicMetric = topicStats(topic);
+ topicMetric.remoteCopyLagSegmentsAggrMetric().setValue(String.valueOf(partition), value);
+ allTopicsStats.remoteCopyLagSegmentsAggrMetric().setValue(topic, topicMetric.remoteCopyLagSegments());
+ }
+
+ // Update the broker-level all topic metric values so that we have a sample right for all topics metric after removal of partition.
+ public void removeRemoteCopyLagSegments(String topic, int partition) {
+ BrokerTopicMetrics topicMetric = topicStats(topic);
+ topicMetric.remoteCopyLagSegmentsAggrMetric().removeKey(String.valueOf(partition));
+ allTopicsStats.remoteCopyLagSegmentsAggrMetric().setValue(topic, topicMetric.remoteCopyLagSegments());
+ }
+
+ public void removeBrokerLevelRemoteCopyLagSegments(String topic) {
+ allTopicsStats.remoteCopyLagSegmentsAggrMetric().removeKey(topic);
+ }
+
+ // Update the broker-level all topic metric values so that we have a sample right for all topics metric after update of partition.
+ public void recordRemoteDeleteLagBytes(String topic, int partition, long value) {
+ BrokerTopicMetrics topicMetric = topicStats(topic);
+ topicMetric.remoteDeleteLagBytesAggrMetric().setValue(String.valueOf(partition), value);
+ allTopicsStats.remoteDeleteLagBytesAggrMetric().setValue(topic, topicMetric.remoteDeleteLagBytes());
+ }
+
+ // Update the broker-level all topic metric values so that we have a sample right for all topics metric after removal of partition.
+ public void removeRemoteDeleteLagBytes(String topic, int partition) {
+ BrokerTopicMetrics topicMetric = topicStats(topic);
+ topicMetric.remoteDeleteLagBytesAggrMetric().removeKey(String.valueOf(partition));
+ allTopicsStats.remoteDeleteLagBytesAggrMetric().setValue(topic, topicMetric.remoteDeleteLagBytes());
+ }
+
+ public void removeBrokerLevelRemoteDeleteLagBytes(String topic) {
+ allTopicsStats.remoteDeleteLagBytesAggrMetric().removeKey(topic);
+ }
+
+ // Update the broker-level all topic metric values so that we have a sample right for all topics metric after update of partition.
+ public void recordRemoteDeleteLagSegments(String topic, int partition, long value) {
+ BrokerTopicMetrics topicMetric = topicStats(topic);
+ topicMetric.remoteDeleteLagSegmentsAggrMetric().setValue(String.valueOf(partition), value);
+ allTopicsStats.remoteDeleteLagSegmentsAggrMetric().setValue(topic, topicMetric.remoteDeleteLagSegments());
+ }
+
+ // Update the broker-level all topic metric values so that we have a sample right for all topics metric after removal of partition.
+ public void removeRemoteDeleteLagSegments(String topic, int partition) {
+ BrokerTopicMetrics topicMetric = topicStats(topic);
+ topicMetric.remoteDeleteLagSegmentsAggrMetric().removeKey(String.valueOf(partition));
+ allTopicsStats.remoteDeleteLagSegmentsAggrMetric().setValue(topic, topicMetric.remoteDeleteLagSegments());
+ }
+
+ public void removeBrokerLevelRemoteDeleteLagSegments(String topic) {
+ allTopicsStats.remoteDeleteLagSegmentsAggrMetric().removeKey(topic);
+ }
+
+ // Update the broker-level all topic metric values so that we have a sample right for all topics metric after update of partition.
+ public void recordRemoteLogMetadataCount(String topic, int partition, long value) {
+ BrokerTopicMetrics topicMetric = topicStats(topic);
+ topicMetric.remoteLogMetadataCountAggrMetric().setValue(String.valueOf(partition), value);
+ allTopicsStats.remoteLogMetadataCountAggrMetric().setValue(topic, topicMetric.remoteLogMetadataCount());
+ }
+
+ // Update the broker-level all topic metric values so that we have a sample right for all topics metric after removal of partition.
+ public void removeRemoteLogMetadataCount(String topic, int partition) {
+ BrokerTopicMetrics topicMetric = topicStats(topic);
+ topicMetric.remoteLogMetadataCountAggrMetric().removeKey(String.valueOf(partition));
+ allTopicsStats.remoteLogMetadataCountAggrMetric().setValue(topic, topicMetric.remoteLogMetadataCount());
+ }
+
+ public void removeBrokerLevelRemoteLogMetadataCount(String topic) {
+ allTopicsStats.remoteLogMetadataCountAggrMetric().removeKey(topic);
+ }
+
+ // Update the broker-level all topic metric values so that we have a sample right for all topics metric after update of partition.
+ public void recordRemoteLogSizeComputationTime(String topic, int partition, long value) {
+ BrokerTopicMetrics topicMetric = topicStats(topic);
+ topicMetric.remoteLogSizeComputationTimeAggrMetric().setValue(String.valueOf(partition), value);
+ allTopicsStats.remoteLogSizeComputationTimeAggrMetric().setValue(topic, topicMetric.remoteLogSizeComputationTime());
+ }
+
+ // Update the broker-level all topic metric values so that we have a sample right for all topics metric after removal of partition.
+ public void removeRemoteLogSizeComputationTime(String topic, int partition) {
+ BrokerTopicMetrics topicMetric = topicStats(topic);
+ topicMetric.remoteLogSizeComputationTimeAggrMetric().removeKey(String.valueOf(partition));
+ allTopicsStats.remoteLogSizeComputationTimeAggrMetric().setValue(topic, topicMetric.remoteLogSizeComputationTime());
+ }
+
+ public void removeBrokerLevelRemoteLogSizeComputationTime(String topic) {
+ allTopicsStats.remoteLogSizeComputationTimeAggrMetric().removeKey(topic);
+ }
+
+ // Update the broker-level all topic metric values so that we have a sample right for all topics metric after update of partition.
+ public void recordRemoteLogSizeBytes(String topic, int partition, long value) {
+ BrokerTopicMetrics topicMetric = topicStats(topic);
+ topicMetric.remoteLogSizeBytesAggrMetric().setValue(String.valueOf(partition), value);
+ allTopicsStats.remoteLogSizeBytesAggrMetric().setValue(topic, topicMetric.remoteLogSizeBytes());
+ }
+
+ // Update the broker-level all topic metric values so that we have a sample right for all topics metric after removal of partition.
+ public void removeRemoteLogSizeBytes(String topic, int partition) {
+ BrokerTopicMetrics topicMetric = topicStats(topic);
+ topicMetric.remoteLogSizeBytesAggrMetric().removeKey(String.valueOf(partition));
+ allTopicsStats.remoteLogSizeBytesAggrMetric().setValue(topic, topicMetric.remoteLogSizeBytes());
+ }
+
+ public void removeBrokerLevelRemoteLogSizeBytes(String topic) {
+ allTopicsStats.remoteLogSizeBytesAggrMetric().removeKey(topic);
+ }
+
+ public void close() {
+ allTopicsStats.close();
+ stats.values().forEach(BrokerTopicMetrics::close);
+ LOG.info("Broker and topic stats closed");
+ }
+
+ public BrokerTopicMetrics allTopicsStats() {
+ return allTopicsStats;
+ }
+}