Skip to content

KAFKA-15189: only init remote topic metrics when enabled #14133

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
showuon merged 14 commits into from
Aug 5, 2023
Merged

KAFKA-15189: only init remote topic metrics when enabled #14133

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
e7fa43e
KAFKA-15189: only init remote topic metrics when enabled
showuon Jul 31, 2023
6f7a39c
KAFKA-15189: pass config into brokerTopicStats
showuon Aug 1, 2023
f7696e8
KAFKA-15189: add integration tests
showuon Aug 1, 2023
6b76e3b
KAFKA-15189: verify all remote storage related metric
showuon Aug 1, 2023
9efb0e2
KAFKA-15189: add class description
showuon Aug 1, 2023
aa89b79
KAFKA-15189: use RemoteStorageMetrics as source of truth
showuon Aug 2, 2023
7ba060c
Merge branch 'trunk' of https://github.com/apache/kafka into KAFKA-15189
showuon Aug 3, 2023
3241b60
KAFKA-15189: resolve conflict
showuon Aug 3, 2023
5dc462b
KAFKA-15189: use metricName as source of truth
showuon Aug 3, 2023
c0536c8
KAFKA-15189: add test name into setup filter
showuon Aug 3, 2023
4987727
KAFKA-15189: fix checkstyle error
showuon Aug 4, 2023
4f37d9c
KAFKA-15189: fix scala 2.12 compilation error
showuon Aug 4, 2023
4c2d335
KAFKA-15189: fix broken test
showuon Aug 4, 2023
86ed825
Merge branch 'KAFKA-15189' of https://github.com/showuon/kafka into K…
showuon Aug 4, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1634,6 +1634,8 @@ project(':storage:api') {

dependencies {
implementation project(':clients')
implementation project(':server-common')
implementation libs.metrics
Comment on lines +1641 to +1642

@showuon showuon Aug 2, 2023 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Need these 2 packages in storage module to access KafkaYammerMetrics class and yammer metrics

implementation libs.slf4jApi

testImplementation project(':clients')
Expand Down
4 changes: 2 additions & 2 deletions core/src/main/java/kafka/log/remote/RemoteLogManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,6 +111,8 @@
import java.util.stream.Stream;

import static org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManagerConfig.REMOTE_LOG_METADATA_COMMON_CLIENT_PREFIX;
import static org.apache.kafka.server.log.remote.storage.RemoteStorageMetrics.REMOTE_LOG_MANAGER_TASKS_AVG_IDLE_PERCENT;
import static org.apache.kafka.server.log.remote.storage.RemoteStorageMetrics.REMOTE_LOG_READER_METRICS_NAME_PREFIX;

/**
* This class is responsible for
Expand All @@ -123,8 +125,6 @@ public class RemoteLogManager implements Closeable {

private static final Logger LOGGER = LoggerFactory.getLogger(RemoteLogManager.class);
private static final String REMOTE_LOG_READER_THREAD_NAME_PREFIX = "remote-log-reader";
public static final String REMOTE_LOG_READER_METRICS_NAME_PREFIX = "RemoteLogReader";
public static final String REMOTE_LOG_MANAGER_TASKS_AVG_IDLE_PERCENT = "RemoteLogManagerTasksAvgIdlePercent";
private final RemoteLogManagerConfig rlmConfig;
private final int brokerId;
private final String logDir;
Expand Down
72 changes: 0 additions & 72 deletions core/src/main/java/kafka/log/remote/RemoteStorageMetrics.java
View file
Open in desktop

This file was deleted.

43 changes: 19 additions & 24 deletions core/src/main/scala/kafka/server/KafkaRequestHandler.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +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.{KafkaThread, Time}
import org.apache.kafka.server.log.remote.storage.RemoteStorageMetrics
import org.apache.kafka.server.metrics.KafkaMetricsGroup

import java.util.Collections
Expand Down Expand Up @@ -286,12 +287,12 @@ class BrokerTopicMetrics(name: Option[String], configOpt: java.util.Optional[Kaf
configOpt.ifPresent(config =>
if (config.remoteLogManagerConfig.enableRemoteStorageSystem()) {
metricTypeMap.putAll(Map(
BrokerTopicStats.RemoteBytesOutPerSec -> MeterWrapper(BrokerTopicStats.RemoteBytesOutPerSec, "bytes"),
BrokerTopicStats.RemoteBytesInPerSec -> MeterWrapper(BrokerTopicStats.RemoteBytesInPerSec, "bytes"),
BrokerTopicStats.RemoteReadRequestsPerSec -> MeterWrapper(BrokerTopicStats.RemoteReadRequestsPerSec, "requests"),
BrokerTopicStats.RemoteWriteRequestsPerSec -> MeterWrapper(BrokerTopicStats.RemoteWriteRequestsPerSec, "requests"),
BrokerTopicStats.FailedRemoteReadRequestsPerSec -> MeterWrapper(BrokerTopicStats.FailedRemoteReadRequestsPerSec, "requests"),
BrokerTopicStats.FailedRemoteWriteRequestsPerSec -> MeterWrapper(BrokerTopicStats.FailedRemoteWriteRequestsPerSec, "requests")
RemoteStorageMetrics.REMOTE_BYTES_OUT_PER_SEC -> MeterWrapper(RemoteStorageMetrics.REMOTE_BYTES_OUT_PER_SEC, "bytes"),
RemoteStorageMetrics.REMOTE_BYTES_IN_PER_SEC -> MeterWrapper(RemoteStorageMetrics.REMOTE_BYTES_IN_PER_SEC, "bytes"),
RemoteStorageMetrics.REMOTE_READ_REQUESTS_PER_SEC -> MeterWrapper(RemoteStorageMetrics.REMOTE_READ_REQUESTS_PER_SEC, "requests"),
RemoteStorageMetrics.REMOTE_WRITE_REQUESTS_PER_SEC -> MeterWrapper(RemoteStorageMetrics.REMOTE_WRITE_REQUESTS_PER_SEC, "requests"),
RemoteStorageMetrics.FAILED_REMOTE_READ_REQUESTS_PER_SEC -> MeterWrapper(RemoteStorageMetrics.FAILED_REMOTE_READ_REQUESTS_PER_SEC, "requests"),
RemoteStorageMetrics.FAILED_REMOTE_WRITE_REQUESTS_PER_SEC -> MeterWrapper(RemoteStorageMetrics.FAILED_REMOTE_WRITE_REQUESTS_PER_SEC, "requests")
).asJava)
})

Expand Down Expand Up @@ -349,17 +350,17 @@ class BrokerTopicMetrics(name: Option[String], configOpt: java.util.Optional[Kaf

def invalidOffsetOrSequenceRecordsPerSec: Meter = metricTypeMap.get(BrokerTopicStats.InvalidOffsetOrSequenceRecordsPerSec).meter()

def remoteBytesOutRate: Meter = metricTypeMap.get(BrokerTopicStats.RemoteBytesOutPerSec).meter()
def remoteBytesOutRate: Meter = metricTypeMap.get(RemoteStorageMetrics.REMOTE_BYTES_OUT_PER_SEC).meter()

def remoteBytesInRate: Meter = metricTypeMap.get(BrokerTopicStats.RemoteBytesInPerSec).meter()
def remoteBytesInRate: Meter = metricTypeMap.get(RemoteStorageMetrics.REMOTE_BYTES_IN_PER_SEC).meter()

def remoteReadRequestRate: Meter = metricTypeMap.get(BrokerTopicStats.RemoteReadRequestsPerSec).meter()
def remoteReadRequestRate: Meter = metricTypeMap.get(RemoteStorageMetrics.REMOTE_READ_REQUESTS_PER_SEC).meter()

def remoteWriteRequestRate: Meter = metricTypeMap.get(BrokerTopicStats.RemoteWriteRequestsPerSec).meter()
def remoteWriteRequestRate: Meter = metricTypeMap.get(RemoteStorageMetrics.REMOTE_WRITE_REQUESTS_PER_SEC).meter()

def failedRemoteReadRequestRate: Meter = metricTypeMap.get(BrokerTopicStats.FailedRemoteReadRequestsPerSec).meter()
def failedRemoteReadRequestRate: Meter = metricTypeMap.get(RemoteStorageMetrics.FAILED_REMOTE_READ_REQUESTS_PER_SEC).meter()

def failedRemoteWriteRequestRate: Meter = metricTypeMap.get(BrokerTopicStats.FailedRemoteWriteRequestsPerSec).meter()
def failedRemoteWriteRequestRate: Meter = metricTypeMap.get(RemoteStorageMetrics.FAILED_REMOTE_WRITE_REQUESTS_PER_SEC).meter()

def closeMetric(metricType: String): Unit = {
val meter = metricTypeMap.get(metricType)
Expand All @@ -385,12 +386,6 @@ object BrokerTopicStats {
val ProduceMessageConversionsPerSec = "ProduceMessageConversionsPerSec"
val ReassignmentBytesInPerSec = "ReassignmentBytesInPerSec"
val ReassignmentBytesOutPerSec = "ReassignmentBytesOutPerSec"
val RemoteBytesOutPerSec = "RemoteBytesOutPerSec"
val RemoteBytesInPerSec = "RemoteBytesInPerSec"
val RemoteReadRequestsPerSec = "RemoteReadRequestsPerSec"
val RemoteWriteRequestsPerSec = "RemoteWriteRequestsPerSec"
val FailedRemoteReadRequestsPerSec = "RemoteReadErrorsPerSec"
val FailedRemoteWriteRequestsPerSec = "RemoteWriteErrorsPerSec"

// These following topics are for LogValidator for better debugging on failed records
val NoKeyCompactedTopicRecordsPerSec = "NoKeyCompactedTopicRecordsPerSec"
Expand Down Expand Up @@ -444,12 +439,12 @@ class BrokerTopicStats(configOpt: java.util.Optional[KafkaConfig] = java.util.Op
topicMetrics.closeMetric(BrokerTopicStats.ProduceMessageConversionsPerSec)
topicMetrics.closeMetric(BrokerTopicStats.ReplicationBytesOutPerSec)
topicMetrics.closeMetric(BrokerTopicStats.ReassignmentBytesOutPerSec)
topicMetrics.closeMetric(BrokerTopicStats.RemoteBytesOutPerSec)
topicMetrics.closeMetric(BrokerTopicStats.RemoteBytesInPerSec)
topicMetrics.closeMetric(BrokerTopicStats.RemoteReadRequestsPerSec)
topicMetrics.closeMetric(BrokerTopicStats.RemoteWriteRequestsPerSec)
topicMetrics.closeMetric(BrokerTopicStats.FailedRemoteReadRequestsPerSec)
topicMetrics.closeMetric(BrokerTopicStats.FailedRemoteWriteRequestsPerSec)
topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_BYTES_OUT_PER_SEC)
topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_BYTES_IN_PER_SEC)
topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_READ_REQUESTS_PER_SEC)
topicMetrics.closeMetric(RemoteStorageMetrics.REMOTE_WRITE_REQUESTS_PER_SEC)
topicMetrics.closeMetric(RemoteStorageMetrics.FAILED_REMOTE_READ_REQUESTS_PER_SEC)
topicMetrics.closeMetric(RemoteStorageMetrics.FAILED_REMOTE_WRITE_REQUESTS_PER_SEC)
}
}

Expand Down
4 changes: 2 additions & 2 deletions core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,9 +92,9 @@
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;

import static kafka.log.remote.RemoteLogManager.REMOTE_LOG_MANAGER_TASKS_AVG_IDLE_PERCENT;
import static kafka.log.remote.RemoteLogManager.REMOTE_LOG_READER_METRICS_NAME_PREFIX;
import static org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManagerConfig.REMOTE_LOG_METADATA_COMMON_CLIENT_PREFIX;
import static org.apache.kafka.server.log.remote.storage.RemoteStorageMetrics.REMOTE_LOG_MANAGER_TASKS_AVG_IDLE_PERCENT;
import static org.apache.kafka.server.log.remote.storage.RemoteStorageMetrics.REMOTE_LOG_READER_METRICS_NAME_PREFIX;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
Expand Down
3 changes: 1 addition & 2 deletions core/src/test/scala/integration/kafka/api/MetricsTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@ import java.util.{Locale, Properties}
import kafka.server.{KafkaConfig, KafkaServer}
import kafka.utils.{JaasTestUtils, TestUtils}
import com.yammer.metrics.core.{Gauge, Histogram, Meter}
import kafka.log.remote.RemoteStorageMetrics
import org.apache.kafka.clients.consumer.Consumer
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.{Metric, MetricName, TopicPartition}
Expand All @@ -25,7 +24,7 @@ import org.apache.kafka.common.errors.{InvalidTopicException, UnknownTopicOrPart
import org.apache.kafka.common.network.ListenerName
import org.apache.kafka.common.security.auth.SecurityProtocol
import org.apache.kafka.common.security.authenticator.TestJaasConfig
import org.apache.kafka.server.log.remote.storage.{NoOpRemoteLogMetadataManager, NoOpRemoteStorageManager, RemoteLogManagerConfig}
import org.apache.kafka.server.log.remote.storage.{NoOpRemoteLogMetadataManager, NoOpRemoteStorageManager, RemoteLogManagerConfig, RemoteStorageMetrics}
import org.apache.kafka.server.metrics.KafkaYammerMetrics
import org.junit.jupiter.api.{AfterEach, BeforeEach, TestInfo}
import org.junit.jupiter.api.Assertions._
Expand Down
23 changes: 8 additions & 15 deletions core/src/test/scala/kafka/server/KafkaRequestHandlerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.MockTime
import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig
import org.apache.kafka.server.log.remote.storage.{RemoteLogManagerConfig, RemoteStorageMetrics}
import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue}
import org.junit.jupiter.api.Test
import org.junit.jupiter.params.ParameterizedTest
Expand All @@ -37,6 +37,7 @@ import org.mockito.Mockito.{mock, when}
import java.net.InetAddress
import java.nio.ByteBuffer
import java.util.concurrent.atomic.AtomicInteger
import scala.jdk.CollectionConverters.SetHasAsScala

class KafkaRequestHandlerTest {

Expand Down Expand Up @@ -90,20 +91,12 @@ class KafkaRequestHandlerTest {
props.setProperty(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, systemRemoteStorageEnabled.toString)
val brokerTopicStats = new BrokerTopicStats(java.util.Optional.of(KafkaConfig.fromProps(props)))
brokerTopicStats.topicStats(topic)
if (systemRemoteStorageEnabled) {
assertTrue(brokerTopicStats.topicStats(topic).metricMap.contains(BrokerTopicStats.RemoteBytesInPerSec))
assertTrue(brokerTopicStats.topicStats(topic).metricMap.contains(BrokerTopicStats.RemoteBytesOutPerSec))
assertTrue(brokerTopicStats.topicStats(topic).metricMap.contains(BrokerTopicStats.RemoteReadRequestsPerSec))
assertTrue(brokerTopicStats.topicStats(topic).metricMap.contains(BrokerTopicStats.RemoteWriteRequestsPerSec))
assertTrue(brokerTopicStats.topicStats(topic).metricMap.contains(BrokerTopicStats.FailedRemoteReadRequestsPerSec))
assertTrue(brokerTopicStats.topicStats(topic).metricMap.contains(BrokerTopicStats.FailedRemoteWriteRequestsPerSec))
} else {
assertFalse(brokerTopicStats.topicStats(topic).metricMap.contains(BrokerTopicStats.RemoteBytesInPerSec))
assertFalse(brokerTopicStats.topicStats(topic).metricMap.contains(BrokerTopicStats.RemoteBytesOutPerSec))
assertFalse(brokerTopicStats.topicStats(topic).metricMap.contains(BrokerTopicStats.RemoteReadRequestsPerSec))
assertFalse(brokerTopicStats.topicStats(topic).metricMap.contains(BrokerTopicStats.RemoteWriteRequestsPerSec))
assertFalse(brokerTopicStats.topicStats(topic).metricMap.contains(BrokerTopicStats.FailedRemoteReadRequestsPerSec))
assertFalse(brokerTopicStats.topicStats(topic).metricMap.contains(BrokerTopicStats.FailedRemoteWriteRequestsPerSec))
for (metric <- RemoteStorageMetrics.brokerTopicStatsMetrics.asScala) {
if (systemRemoteStorageEnabled) {
assertTrue(brokerTopicStats.topicStats(topic).metricMap.contains(metric.getName))
} else {
assertFalse(brokerTopicStats.topicStats(topic).metricMap.contains(metric.getName))
}
}
}
}
Loading