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/05/30 09:51:33 UTC
(kafka) branch trunk updated: MINOR: Remove KafkaConfig dependency in KafkaRequestHandler (#16108)
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 8068a086a3f MINOR: Remove KafkaConfig dependency in KafkaRequestHandler (#16108)
8068a086a3f is described below
commit 8068a086a3f41b9f7e4d4a1dab3338f029089f23
Author: Mickael Maison <mi...@users.noreply.github.com>
AuthorDate: Thu May 30 11:51:24 2024 +0200
MINOR: Remove KafkaConfig dependency in KafkaRequestHandler (#16108)
Reviewers: Luke Chen <sh...@gmail.com>, Apoorv Mittal <am...@confluent.io>
---
.../kafka/server/builders/KafkaApisBuilder.java | 2 +-
.../server/builders/ReplicaManagerBuilder.java | 2 +-
.../src/main/scala/kafka/server/BrokerServer.scala | 2 +-
.../scala/kafka/server/KafkaRequestHandler.scala | 63 +++++++++++-----------
core/src/main/scala/kafka/server/KafkaServer.scala | 2 +-
.../kafka/log/remote/RemoteLogManagerTest.java | 2 +-
.../java/kafka/log/remote/RemoteLogReaderTest.java | 8 +--
.../kafka/server/KafkaRequestHandlerTest.scala | 34 +++++-------
.../unit/kafka/server/ReplicaManagerTest.scala | 6 +--
.../jmh/fetcher/ReplicaFetcherThreadBenchmark.java | 5 +-
.../metadata/KRaftMetadataRequestBenchmark.java | 2 +-
.../jmh/metadata/MetadataRequestBenchmark.java | 2 +-
.../partition/PartitionMakeFollowerBenchmark.java | 2 +-
.../UpdateFollowerFetchStateBenchmark.java | 2 +-
.../kafka/jmh/record/BaseRecordBatchBenchmark.java | 3 +-
.../apache/kafka/jmh/server/CheckpointBench.java | 3 +-
.../kafka/jmh/server/PartitionCreationBench.java | 3 +-
17 files changed, 61 insertions(+), 82 deletions(-)
diff --git a/core/src/main/java/kafka/server/builders/KafkaApisBuilder.java b/core/src/main/java/kafka/server/builders/KafkaApisBuilder.java
index 9182f9c4bcd..6ffd741f4fc 100644
--- a/core/src/main/java/kafka/server/builders/KafkaApisBuilder.java
+++ b/core/src/main/java/kafka/server/builders/KafkaApisBuilder.java
@@ -179,7 +179,7 @@ public class KafkaApisBuilder {
if (metrics == null) throw new RuntimeException("You must set metrics");
if (quotas == null) throw new RuntimeException("You must set quotas");
if (fetchManager == null) throw new RuntimeException("You must set fetchManager");
- if (brokerTopicStats == null) brokerTopicStats = new BrokerTopicStats(Optional.of(config));
+ if (brokerTopicStats == null) brokerTopicStats = new BrokerTopicStats(config.isRemoteLogStorageSystemEnabled());
if (apiVersionManager == null) throw new RuntimeException("You must set apiVersionManager");
return new KafkaApis(requestChannel,
diff --git a/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java b/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java
index 13b60a539f9..82aa75909ab 100644
--- a/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java
+++ b/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java
@@ -185,7 +185,7 @@ public class ReplicaManagerBuilder {
if (metadataCache == null) throw new RuntimeException("You must set metadataCache");
if (logDirFailureChannel == null) throw new RuntimeException("You must set logDirFailureChannel");
if (alterPartitionManager == null) throw new RuntimeException("You must set alterIsrManager");
- if (brokerTopicStats == null) brokerTopicStats = new BrokerTopicStats(Optional.of(config));
+ if (brokerTopicStats == null) brokerTopicStats = new BrokerTopicStats(config.isRemoteLogStorageSystemEnabled());
// Initialize metrics in the end just before passing it to ReplicaManager to ensure ReplicaManager closes the
// metrics correctly. There might be a resource leak if it is initialized and an exception occurs between
// its initialization and creation of ReplicaManager.
diff --git a/core/src/main/scala/kafka/server/BrokerServer.scala b/core/src/main/scala/kafka/server/BrokerServer.scala
index c8ddd25bc5c..cb9a7ee0fbb 100644
--- a/core/src/main/scala/kafka/server/BrokerServer.scala
+++ b/core/src/main/scala/kafka/server/BrokerServer.scala
@@ -184,7 +184,7 @@ class BrokerServer(
kafkaScheduler.startup()
/* register broker metrics */
- brokerTopicStats = new BrokerTopicStats(java.util.Optional.of(config))
+ brokerTopicStats = new BrokerTopicStats(config.isRemoteLogStorageSystemEnabled)
quotaManagers = QuotaFactory.instantiate(config, metrics, time, s"broker-${config.nodeId}-")
diff --git a/core/src/main/scala/kafka/server/KafkaRequestHandler.scala b/core/src/main/scala/kafka/server/KafkaRequestHandler.scala
index d9a8a8315b7..d054d9604c7 100755
--- a/core/src/main/scala/kafka/server/KafkaRequestHandler.scala
+++ b/core/src/main/scala/kafka/server/KafkaRequestHandler.scala
@@ -17,8 +17,8 @@
package kafka.server
-import kafka.network._
-import kafka.utils._
+import kafka.network.RequestChannel
+import kafka.utils.{Exit, Logging, Pool}
import kafka.server.KafkaRequestHandler.{threadCurrentRequest, threadRequestChannel}
import java.util.concurrent.{ConcurrentHashMap, CountDownLatch, TimeUnit}
@@ -246,7 +246,7 @@ class KafkaRequestHandlerPool(
}
}
-class BrokerTopicMetrics(name: Option[String], configOpt: java.util.Optional[KafkaConfig]) {
+class BrokerTopicMetrics(name: Option[String], remoteStorageEnabled: Boolean = false) {
private val metricsGroup = new KafkaMetricsGroup(this.getClass)
val tags: java.util.Map[String, String] = name match {
@@ -333,31 +333,30 @@ class BrokerTopicMetrics(name: Option[String], configOpt: java.util.Optional[Kaf
metricTypeMap.put(BrokerTopicStats.ReassignmentBytesOutPerSec, MeterWrapper(BrokerTopicStats.ReassignmentBytesOutPerSec, "bytes"))
}
- configOpt.ifPresent(config =>
- if (config.remoteLogManagerConfig.enableRemoteStorageSystem()) {
- metricTypeMap.putAll(Map(
- RemoteStorageMetrics.REMOTE_COPY_BYTES_PER_SEC_METRIC.getName -> MeterWrapper(RemoteStorageMetrics.REMOTE_COPY_BYTES_PER_SEC_METRIC.getName, "bytes"),
- RemoteStorageMetrics.REMOTE_FETCH_BYTES_PER_SEC_METRIC.getName -> MeterWrapper(RemoteStorageMetrics.REMOTE_FETCH_BYTES_PER_SEC_METRIC.getName, "bytes"),
- RemoteStorageMetrics.REMOTE_FETCH_REQUESTS_PER_SEC_METRIC.getName -> MeterWrapper(RemoteStorageMetrics.REMOTE_FETCH_REQUESTS_PER_SEC_METRIC.getName, "requests"),
- RemoteStorageMetrics.REMOTE_COPY_REQUESTS_PER_SEC_METRIC.getName -> MeterWrapper(RemoteStorageMetrics.REMOTE_COPY_REQUESTS_PER_SEC_METRIC.getName, "requests"),
- RemoteStorageMetrics.REMOTE_DELETE_REQUESTS_PER_SEC_METRIC.getName -> MeterWrapper(RemoteStorageMetrics.REMOTE_DELETE_REQUESTS_PER_SEC_METRIC.getName, "requests"),
- RemoteStorageMetrics.BUILD_REMOTE_LOG_AUX_STATE_REQUESTS_PER_SEC_METRIC.getName -> MeterWrapper(RemoteStorageMetrics.BUILD_REMOTE_LOG_AUX_STATE_REQUESTS_PER_SEC_METRIC.getName, "requests"),
- RemoteStorageMetrics.FAILED_REMOTE_FETCH_PER_SEC_METRIC.getName -> MeterWrapper(RemoteStorageMetrics.FAILED_REMOTE_FETCH_PER_SEC_METRIC.getName, "requests"),
- RemoteStorageMetrics.FAILED_REMOTE_COPY_PER_SEC_METRIC.getName -> MeterWrapper(RemoteStorageMetrics.FAILED_REMOTE_COPY_PER_SEC_METRIC.getName, "requests"),
- RemoteStorageMetrics.FAILED_REMOTE_DELETE_PER_SEC_METRIC.getName -> MeterWrapper(RemoteStorageMetrics.FAILED_REMOTE_DELETE_PER_SEC_METRIC.getName, "requests"),
- RemoteStorageMetrics.FAILED_BUILD_REMOTE_LOG_AUX_STATE_PER_SEC_METRIC.getName -> MeterWrapper(RemoteStorageMetrics.FAILED_BUILD_REMOTE_LOG_AUX_STATE_PER_SEC_METRIC.getName, "requests")
- ).asJava)
-
- metricGaugeTypeMap.putAll(Map(
- RemoteStorageMetrics.REMOTE_COPY_LAG_BYTES_METRIC.getName -> GaugeWrapper(RemoteStorageMetrics.REMOTE_COPY_LAG_BYTES_METRIC.getName),
- RemoteStorageMetrics.REMOTE_COPY_LAG_SEGMENTS_METRIC.getName -> GaugeWrapper(RemoteStorageMetrics.REMOTE_COPY_LAG_SEGMENTS_METRIC.getName),
- RemoteStorageMetrics.REMOTE_DELETE_LAG_BYTES_METRIC.getName -> GaugeWrapper(RemoteStorageMetrics.REMOTE_DELETE_LAG_BYTES_METRIC.getName),
- RemoteStorageMetrics.REMOTE_DELETE_LAG_SEGMENTS_METRIC.getName -> GaugeWrapper(RemoteStorageMetrics.REMOTE_DELETE_LAG_SEGMENTS_METRIC.getName),
- RemoteStorageMetrics.REMOTE_LOG_METADATA_COUNT_METRIC.getName -> GaugeWrapper(RemoteStorageMetrics.REMOTE_LOG_METADATA_COUNT_METRIC.getName),
- RemoteStorageMetrics.REMOTE_LOG_SIZE_COMPUTATION_TIME_METRIC.getName -> GaugeWrapper(RemoteStorageMetrics.REMOTE_LOG_SIZE_COMPUTATION_TIME_METRIC.getName),
- RemoteStorageMetrics.REMOTE_LOG_SIZE_BYTES_METRIC.getName -> GaugeWrapper(RemoteStorageMetrics.REMOTE_LOG_SIZE_BYTES_METRIC.getName)
- ).asJava)
- })
+ if (remoteStorageEnabled) {
+ metricTypeMap.putAll(Map(
+ RemoteStorageMetrics.REMOTE_COPY_BYTES_PER_SEC_METRIC.getName -> MeterWrapper(RemoteStorageMetrics.REMOTE_COPY_BYTES_PER_SEC_METRIC.getName, "bytes"),
+ RemoteStorageMetrics.REMOTE_FETCH_BYTES_PER_SEC_METRIC.getName -> MeterWrapper(RemoteStorageMetrics.REMOTE_FETCH_BYTES_PER_SEC_METRIC.getName, "bytes"),
+ RemoteStorageMetrics.REMOTE_FETCH_REQUESTS_PER_SEC_METRIC.getName -> MeterWrapper(RemoteStorageMetrics.REMOTE_FETCH_REQUESTS_PER_SEC_METRIC.getName, "requests"),
+ RemoteStorageMetrics.REMOTE_COPY_REQUESTS_PER_SEC_METRIC.getName -> MeterWrapper(RemoteStorageMetrics.REMOTE_COPY_REQUESTS_PER_SEC_METRIC.getName, "requests"),
+ RemoteStorageMetrics.REMOTE_DELETE_REQUESTS_PER_SEC_METRIC.getName -> MeterWrapper(RemoteStorageMetrics.REMOTE_DELETE_REQUESTS_PER_SEC_METRIC.getName, "requests"),
+ RemoteStorageMetrics.BUILD_REMOTE_LOG_AUX_STATE_REQUESTS_PER_SEC_METRIC.getName -> MeterWrapper(RemoteStorageMetrics.BUILD_REMOTE_LOG_AUX_STATE_REQUESTS_PER_SEC_METRIC.getName, "requests"),
+ RemoteStorageMetrics.FAILED_REMOTE_FETCH_PER_SEC_METRIC.getName -> MeterWrapper(RemoteStorageMetrics.FAILED_REMOTE_FETCH_PER_SEC_METRIC.getName, "requests"),
+ RemoteStorageMetrics.FAILED_REMOTE_COPY_PER_SEC_METRIC.getName -> MeterWrapper(RemoteStorageMetrics.FAILED_REMOTE_COPY_PER_SEC_METRIC.getName, "requests"),
+ RemoteStorageMetrics.FAILED_REMOTE_DELETE_PER_SEC_METRIC.getName -> MeterWrapper(RemoteStorageMetrics.FAILED_REMOTE_DELETE_PER_SEC_METRIC.getName, "requests"),
+ RemoteStorageMetrics.FAILED_BUILD_REMOTE_LOG_AUX_STATE_PER_SEC_METRIC.getName -> MeterWrapper(RemoteStorageMetrics.FAILED_BUILD_REMOTE_LOG_AUX_STATE_PER_SEC_METRIC.getName, "requests")
+ ).asJava)
+
+ metricGaugeTypeMap.putAll(Map(
+ RemoteStorageMetrics.REMOTE_COPY_LAG_BYTES_METRIC.getName -> GaugeWrapper(RemoteStorageMetrics.REMOTE_COPY_LAG_BYTES_METRIC.getName),
+ RemoteStorageMetrics.REMOTE_COPY_LAG_SEGMENTS_METRIC.getName -> GaugeWrapper(RemoteStorageMetrics.REMOTE_COPY_LAG_SEGMENTS_METRIC.getName),
+ RemoteStorageMetrics.REMOTE_DELETE_LAG_BYTES_METRIC.getName -> GaugeWrapper(RemoteStorageMetrics.REMOTE_DELETE_LAG_BYTES_METRIC.getName),
+ RemoteStorageMetrics.REMOTE_DELETE_LAG_SEGMENTS_METRIC.getName -> GaugeWrapper(RemoteStorageMetrics.REMOTE_DELETE_LAG_SEGMENTS_METRIC.getName),
+ RemoteStorageMetrics.REMOTE_LOG_METADATA_COUNT_METRIC.getName -> GaugeWrapper(RemoteStorageMetrics.REMOTE_LOG_METADATA_COUNT_METRIC.getName),
+ RemoteStorageMetrics.REMOTE_LOG_SIZE_COMPUTATION_TIME_METRIC.getName -> GaugeWrapper(RemoteStorageMetrics.REMOTE_LOG_SIZE_COMPUTATION_TIME_METRIC.getName),
+ RemoteStorageMetrics.REMOTE_LOG_SIZE_BYTES_METRIC.getName -> GaugeWrapper(RemoteStorageMetrics.REMOTE_LOG_SIZE_BYTES_METRIC.getName)
+ ).asJava)
+ }
// used for testing only
def metricMap: Map[String, MeterWrapper] = metricTypeMap.toMap
@@ -523,11 +522,11 @@ object BrokerTopicStats {
val InvalidOffsetOrSequenceRecordsPerSec = "InvalidOffsetOrSequenceRecordsPerSec"
}
-class BrokerTopicStats(configOpt: java.util.Optional[KafkaConfig] = java.util.Optional.empty()) extends Logging {
+class BrokerTopicStats(remoteStorageEnabled: Boolean = false) extends Logging {
- private val valueFactory = (k: String) => new BrokerTopicMetrics(Some(k), configOpt)
+ private val valueFactory = (k: String) => new BrokerTopicMetrics(Some(k), remoteStorageEnabled)
private val stats = new Pool[String, BrokerTopicMetrics](Some(valueFactory))
- val allTopicsStats = new BrokerTopicMetrics(None, configOpt)
+ val allTopicsStats = new BrokerTopicMetrics(None, remoteStorageEnabled)
def isTopicStatsExisted(topic: String): Boolean =
stats.contains(topic)
@@ -553,7 +552,7 @@ class BrokerTopicStats(configOpt: java.util.Optional[KafkaConfig] = java.util.Op
}
}
- def updateReassignmentBytesOut(value: Long): Unit = {
+ private def updateReassignmentBytesOut(value: Long): Unit = {
allTopicsStats.reassignmentBytesOutPerSec.foreach { metric =>
metric.mark(value)
}
diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala
index ce155e92ffd..1d076bd5ea7 100755
--- a/core/src/main/scala/kafka/server/KafkaServer.scala
+++ b/core/src/main/scala/kafka/server/KafkaServer.scala
@@ -276,7 +276,7 @@ class KafkaServer(
createCurrentControllerIdMetric()
/* register broker metrics */
- _brokerTopicStats = new BrokerTopicStats(java.util.Optional.of(config))
+ _brokerTopicStats = new BrokerTopicStats(config.isRemoteLogStorageSystemEnabled)
quotaManagers = QuotaFactory.instantiate(config, metrics, time, threadNamePrefix.getOrElse(""))
KafkaBroker.notifyClusterListeners(clusterId, kafkaMetricsReporters ++ metrics.reporters.asScala)
diff --git a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
index 8a3cefa0405..ae2691a2d45 100644
--- a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
+++ b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
@@ -213,7 +213,7 @@ public class RemoteLogManagerTest {
props.setProperty(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, "true");
props.setProperty(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_PROP, "100");
remoteLogManagerConfig = createRLMConfig(props);
- brokerTopicStats = new BrokerTopicStats(Optional.of(KafkaConfig.fromProps(props)));
+ brokerTopicStats = new BrokerTopicStats(KafkaConfig.fromProps(props).isRemoteLogStorageSystemEnabled());
remoteLogManager = new RemoteLogManager(remoteLogManagerConfig, brokerId, logDir, clusterId, time,
tp -> Optional.of(mockLog),
diff --git a/core/src/test/java/kafka/log/remote/RemoteLogReaderTest.java b/core/src/test/java/kafka/log/remote/RemoteLogReaderTest.java
index d1d35b5e5df..bff58364b9d 100644
--- a/core/src/test/java/kafka/log/remote/RemoteLogReaderTest.java
+++ b/core/src/test/java/kafka/log/remote/RemoteLogReaderTest.java
@@ -17,11 +17,9 @@
package kafka.log.remote;
import kafka.server.BrokerTopicStats;
-import kafka.server.KafkaConfig;
import kafka.utils.TestUtils;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.record.Records;
-import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig;
import org.apache.kafka.server.log.remote.storage.RemoteStorageException;
import org.apache.kafka.storage.internals.log.FetchDataInfo;
import org.apache.kafka.storage.internals.log.LogOffsetMetadata;
@@ -32,8 +30,6 @@ import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import java.io.IOException;
-import java.util.Optional;
-import java.util.Properties;
import java.util.function.Consumer;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -55,9 +51,7 @@ public class RemoteLogReaderTest {
@BeforeEach
public void setUp() {
TestUtils.clearYammerMetrics();
- Properties props = kafka.utils.TestUtils.createDummyBrokerConfig();
- props.setProperty(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, "true");
- brokerTopicStats = new BrokerTopicStats(Optional.of(KafkaConfig.fromProps(props)));
+ brokerTopicStats = new BrokerTopicStats(true);
}
@Test
diff --git a/core/src/test/scala/kafka/server/KafkaRequestHandlerTest.scala b/core/src/test/scala/kafka/server/KafkaRequestHandlerTest.scala
index ba584c9a6e7..7be9b47c124 100644
--- a/core/src/test/scala/kafka/server/KafkaRequestHandlerTest.scala
+++ b/core/src/test/scala/kafka/server/KafkaRequestHandlerTest.scala
@@ -25,7 +25,7 @@ import org.apache.kafka.common.protocol.ApiKeys
import org.apache.kafka.common.requests.{RequestContext, RequestHeader}
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.{RemoteLogManagerConfig, RemoteStorageMetrics}
+import org.apache.kafka.server.log.remote.storage.RemoteStorageMetrics
import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue}
import org.junit.jupiter.api.Test
import org.junit.jupiter.params.ParameterizedTest
@@ -40,13 +40,11 @@ import java.util.concurrent.CompletableFuture
import java.util.concurrent.atomic.AtomicInteger
class KafkaRequestHandlerTest {
- val props = kafka.utils.TestUtils.createDummyBrokerConfig()
- props.setProperty(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, true.toString)
- val brokerTopicStats = new BrokerTopicStats(java.util.Optional.of(KafkaConfig.fromProps(props)))
+ val brokerTopicStats = new BrokerTopicStats(true)
val topic = "topic"
val topic2 = "topic2"
- val brokerTopicMetrics = brokerTopicStats.topicStats(topic)
- val allTopicMetrics = brokerTopicStats.allTopicsStats
+ val brokerTopicMetrics: BrokerTopicMetrics = brokerTopicStats.topicStats(topic)
+ val allTopicMetrics: BrokerTopicMetrics = brokerTopicStats.allTopicsStats
@Test
def testCallbackTiming(): Unit = {
@@ -65,7 +63,7 @@ class KafkaRequestHandlerTest {
time.sleep(2)
// Prepare the callback.
val callback = KafkaRequestHandler.wrapAsyncCallback(
- (reqLocal: RequestLocal, ms: Int) => {
+ (_: RequestLocal, ms: Int) => {
time.sleep(ms)
handler.stop()
},
@@ -104,7 +102,7 @@ class KafkaRequestHandlerTest {
handledCount = handledCount + 1
// Prepare the callback.
val callback = KafkaRequestHandler.wrapAsyncCallback(
- (reqLocal: RequestLocal, ms: Int) => {
+ (_: RequestLocal, _: Int) => {
handler.stop()
},
RequestLocal.NoCaching)
@@ -140,7 +138,7 @@ class KafkaRequestHandlerTest {
when(apiHandler.handle(ArgumentMatchers.eq(request), any())).thenAnswer { _ =>
// Prepare the callback.
val callback = KafkaRequestHandler.wrapAsyncCallback(
- (reqLocal: RequestLocal, ms: Int) => {
+ (reqLocal: RequestLocal, _: Int) => {
reqLocal.bufferSupplier.close()
handledCount = handledCount + 1
handler.stop()
@@ -175,7 +173,7 @@ class KafkaRequestHandlerTest {
when(apiHandler.handle(ArgumentMatchers.eq(request), any())).thenAnswer { _ =>
// Prepare the callback.
val callback = KafkaRequestHandler.wrapAsyncCallback(
- (reqLocal: RequestLocal, ms: Int) => {
+ (reqLocal: RequestLocal, _: Int) => {
reqLocal.bufferSupplier.close()
handledCount = handledCount + 1
handler.stop()
@@ -195,9 +193,7 @@ class KafkaRequestHandlerTest {
@ParameterizedTest
@ValueSource(booleans = Array(true, false))
def testTopicStats(systemRemoteStorageEnabled: Boolean): Unit = {
- 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 brokerTopicStats = new BrokerTopicStats(systemRemoteStorageEnabled)
val brokerTopicMetrics = brokerTopicStats.topicStats(topic)
val gaugeMetrics = Set(
RemoteStorageMetrics.REMOTE_LOG_SIZE_COMPUTATION_TIME_METRIC.getName,
@@ -243,17 +239,13 @@ class KafkaRequestHandlerTest {
def setupBrokerTopicMetrics(systemRemoteStorageEnabled: Boolean = true): BrokerTopicMetrics = {
val topic = "topic"
- val props = kafka.utils.TestUtils.createDummyBrokerConfig()
- props.setProperty(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, systemRemoteStorageEnabled.toString)
- new BrokerTopicMetrics(Option.apply(topic), java.util.Optional.of(KafkaConfig.fromProps(props)))
+ new BrokerTopicMetrics(Option.apply(topic), systemRemoteStorageEnabled)
}
@ParameterizedTest
@ValueSource(booleans = Array(true, false))
def testSingularCopyLagBytesMetric(systemRemoteStorageEnabled: Boolean): Unit = {
- 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 brokerTopicStats = new BrokerTopicStats(systemRemoteStorageEnabled)
val brokerTopicMetrics = brokerTopicStats.topicStats(topic)
if (systemRemoteStorageEnabled) {
@@ -593,9 +585,7 @@ class KafkaRequestHandlerTest {
@ParameterizedTest
@ValueSource(booleans = Array(true, false))
def testSingularLogSizeBytesMetric(systemRemoteStorageEnabled: Boolean): Unit = {
- 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 brokerTopicStats = new BrokerTopicStats(systemRemoteStorageEnabled)
val brokerTopicMetrics = brokerTopicStats.topicStats(topic)
if (systemRemoteStorageEnabled) {
brokerTopicStats.recordRemoteLogSizeBytes(topic, 0, 100)
diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
index eebc6e99a31..a2dbc9cc0c3 100644
--- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
@@ -3389,7 +3389,7 @@ class ReplicaManagerTest {
when(mockLog.remoteLogEnabled()).thenReturn(enableRemoteStorage)
when(mockLog.remoteStorageSystemEnable).thenReturn(enableRemoteStorage)
val aliveBrokers = aliveBrokerIds.map(brokerId => new Node(brokerId, s"host$brokerId", brokerId))
- brokerTopicStats = new BrokerTopicStats(java.util.Optional.of(KafkaConfig.fromProps(props)))
+ brokerTopicStats = new BrokerTopicStats(KafkaConfig.fromProps(props).remoteLogManagerConfig.enableRemoteStorageSystem)
val metadataCache: MetadataCache = mock(classOf[MetadataCache])
when(metadataCache.topicIdInfo()).thenReturn((topicIds.asJava, topicNames.asJava))
@@ -4095,7 +4095,7 @@ class ReplicaManagerTest {
val config = new AbstractConfig(RemoteLogManagerConfig.CONFIG_DEF, props)
val remoteLogManagerConfig = new RemoteLogManagerConfig(config)
val mockLog = mock(classOf[UnifiedLog])
- val brokerTopicStats = new BrokerTopicStats(java.util.Optional.of(KafkaConfig.fromProps(props)))
+ val brokerTopicStats = new BrokerTopicStats(KafkaConfig.fromProps(props).isRemoteLogStorageSystemEnabled)
val remoteLogManager = new RemoteLogManager(
remoteLogManagerConfig,
0,
@@ -4194,7 +4194,7 @@ class ReplicaManagerTest {
val config = new AbstractConfig(RemoteLogManagerConfig.CONFIG_DEF, props)
val remoteLogManagerConfig = new RemoteLogManagerConfig(config)
val dummyLog = mock(classOf[UnifiedLog])
- val brokerTopicStats = new BrokerTopicStats(java.util.Optional.of(KafkaConfig.fromProps(props)))
+ val brokerTopicStats = new BrokerTopicStats(KafkaConfig.fromProps(props).isRemoteLogStorageSystemEnabled)
val remoteLogManager = new RemoteLogManager(
remoteLogManagerConfig,
0,
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 48ff022a79a..c1c36447c29 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
@@ -95,7 +95,6 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
-import java.util.Optional;
import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
@@ -112,7 +111,7 @@ import scala.collection.Map;
public class ReplicaFetcherThreadBenchmark {
private final File logDir = new File(System.getProperty("java.io.tmpdir"), UUID.randomUUID().toString());
private final KafkaScheduler scheduler = new KafkaScheduler(1, true, "scheduler");
- private final Pool<TopicPartition, Partition> pool = new Pool<TopicPartition, Partition>(Option.empty());
+ private final Pool<TopicPartition, Partition> pool = new Pool<>(Option.empty());
private final Metrics metrics = new Metrics();
private final Option<Uuid> topicId = Option.apply(Uuid.randomUuid());
@Param({"100", "500", "1000", "5000"})
@@ -133,7 +132,7 @@ public class ReplicaFetcherThreadBenchmark {
KafkaConfig config = new KafkaConfig(props);
LogConfig logConfig = createLogConfig();
- BrokerTopicStats brokerTopicStats = new BrokerTopicStats(Optional.empty());
+ BrokerTopicStats brokerTopicStats = new BrokerTopicStats(false);
LogDirFailureChannel logDirFailureChannel = Mockito.mock(LogDirFailureChannel.class);
List<File> logDirs = Collections.singletonList(logDir);
logManager = new LogManagerBuilder().
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 d3c606b3fd4..19b2f25d5a4 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
@@ -114,7 +114,7 @@ public class KRaftMetadataRequestBenchmark {
clientQuotaManager, clientRequestQuotaManager, controllerMutationQuotaManager, replicaQuotaManager,
replicaQuotaManager, replicaQuotaManager, Option.empty());
private final FetchManager fetchManager = Mockito.mock(FetchManager.class);
- private final BrokerTopicStats brokerTopicStats = new BrokerTopicStats(Optional.empty());
+ private final BrokerTopicStats brokerTopicStats = new BrokerTopicStats(false);
private final KafkaPrincipal principal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "test-user");
@Param({"500", "1000", "5000"})
private int topicCount;
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 abca07a71af..5d3591454bd 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
@@ -118,7 +118,7 @@ public class MetadataRequestBenchmark {
clientQuotaManager, clientRequestQuotaManager, controllerMutationQuotaManager, replicaQuotaManager,
replicaQuotaManager, replicaQuotaManager, Option.empty());
private final FetchManager fetchManager = Mockito.mock(FetchManager.class);
- private final BrokerTopicStats brokerTopicStats = new BrokerTopicStats(Optional.empty());
+ private final BrokerTopicStats brokerTopicStats = new BrokerTopicStats(false);
private final KafkaPrincipal principal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "test-user");
@Param({"500", "1000", "5000"})
private int topicCount;
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 8d016b504e8..62840ef9b86 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
@@ -94,7 +94,7 @@ public class PartitionMakeFollowerBenchmark {
scheduler.startup();
LogConfig logConfig = new LogConfig(new Properties());
- BrokerTopicStats brokerTopicStats = new BrokerTopicStats(Optional.empty());
+ BrokerTopicStats brokerTopicStats = new BrokerTopicStats(false);
LogDirFailureChannel logDirFailureChannel = Mockito.mock(LogDirFailureChannel.class);
logManager = new LogManagerBuilder().
setLogDirs(Collections.singletonList(logDir)).
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 7495dd37d2a..686ee7acf67 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
@@ -74,7 +74,7 @@ public class UpdateFollowerFetchStateBenchmark {
private final Option<Uuid> topicId = OptionConverters.toScala(Optional.of(Uuid.randomUuid()));
private final File logDir = new File(System.getProperty("java.io.tmpdir"), topicPartition.toString());
private final KafkaScheduler scheduler = new KafkaScheduler(1, true, "scheduler");
- private final BrokerTopicStats brokerTopicStats = new BrokerTopicStats(Optional.empty());
+ private final BrokerTopicStats brokerTopicStats = new BrokerTopicStats(false);
private final LogDirFailureChannel logDirFailureChannel = Mockito.mock(LogDirFailureChannel.class);
private long nextOffset = 0;
private LogManager logManager;
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 0a9c5ff5d44..1190854a339 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
@@ -36,7 +36,6 @@ import org.openjdk.jmh.annotations.TearDown;
import java.nio.ByteBuffer;
import java.util.Arrays;
-import java.util.Optional;
import java.util.Random;
import java.util.stream.IntStream;
@@ -80,7 +79,7 @@ public abstract class BaseRecordBatchBenchmark {
ByteBuffer[] batchBuffers;
RequestLocal requestLocal;
LogValidator.MetricsRecorder validatorMetricsRecorder = UnifiedLog.newValidatorMetricsRecorder(
- new BrokerTopicStats(Optional.empty()).allTopicsStats());
+ new BrokerTopicStats(false).allTopicsStats());
@Setup
public void init() {
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 3c4e84c6a6d..059e41dadec 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
@@ -57,7 +57,6 @@ import org.openjdk.jmh.annotations.Warmup;
import java.io.File;
import java.util.ArrayList;
import java.util.List;
-import java.util.Optional;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@@ -112,7 +111,7 @@ public class CheckpointBench {
new LogConfig(new Properties()), new MockConfigRepository(), new CleanerConfig(1, 4 * 1024 * 1024L, 0.9d,
1024 * 1024, 32 * 1024 * 1024, Double.MAX_VALUE, 15 * 1000, true), time, MetadataVersion.latestTesting(), 4, false, Option.empty(), false, ServerLogConfigs.LOG_INITIAL_TASK_DELAY_MS_DEFAULT);
scheduler.startup();
- final BrokerTopicStats brokerTopicStats = new BrokerTopicStats(Optional.empty());
+ final BrokerTopicStats brokerTopicStats = new BrokerTopicStats(false);
final MetadataCache metadataCache =
MetadataCache.zkMetadataCache(this.brokerProperties.brokerId(),
this.brokerProperties.interBrokerProtocolVersion(),
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 c38e4b489e2..24261ba1071 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
@@ -63,7 +63,6 @@ import java.io.File;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-import java.util.Optional;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@@ -114,7 +113,7 @@ public class PartitionCreationBench {
this.metrics = new Metrics();
this.time = Time.SYSTEM;
this.failureChannel = new LogDirFailureChannel(brokerProperties.logDirs().size());
- final BrokerTopicStats brokerTopicStats = new BrokerTopicStats(Optional.empty());
+ final BrokerTopicStats brokerTopicStats = new BrokerTopicStats(false);
final List<File> files =
JavaConverters.seqAsJavaList(brokerProperties.logDirs()).stream().map(File::new).collect(Collectors.toList());
CleanerConfig cleanerConfig = new CleanerConfig(1,