Skip to content

KAFKA-14953: Add tiered storage related metrics #13944

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
satishd merged 9 commits into from
Jul 18, 2023
Merged

KAFKA-14953: Add tiered storage related metrics #13944

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
b3e3793
KAFKA-14953: Adding RemoteLogManager metrics
abhijeetk88 Jul 1, 2023
925f9cc
Added more tests
abhijeetk88 Jul 2, 2023
fb59242
Address review comments
abhijeetk88 Jul 8, 2023
5663ccb
Undo renaming metrics
abhijeetk88 Jul 16, 2023
bd1104a
Capture all metrics in single place so we wont forget to remove them
abhijeetk88 Jul 16, 2023
7fcdb14
Remove metrics in try/catch
abhijeetk88 Jul 16, 2023
bd46e24
Aggregate metrics
abhijeetk88 Jul 16, 2023
5246478
Fix spotbugs error
abhijeetk88 Jul 17, 2023
4cc354b
Empty commit
abhijeetk88 Jul 17, 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
1 change: 1 addition & 0 deletions build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1704,6 +1704,7 @@ project(':storage') {
implementation libs.caffeine
implementation libs.slf4jApi
implementation libs.jacksonDatabind
implementation libs.metrics

testImplementation project(':clients')
testImplementation project(':clients').sourceSets.test.output
Expand Down
1 change: 1 addition & 0 deletions checkstyle/import-control-core.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,7 @@
<subpackage name="log.remote">
<allow pkg="org.apache.kafka.server.common" />
<allow pkg="org.apache.kafka.server.log.remote" />
<allow pkg="org.apache.kafka.server.metrics" />
<allow pkg="org.apache.kafka.storage.internals" />
<allow pkg="kafka.log" />
<allow pkg="kafka.cluster" />
Expand Down
1 change: 1 addition & 0 deletions checkstyle/import-control.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -266,6 +266,7 @@
</subpackage>

<subpackage name="storage.internals">
<allow pkg="com.yammer.metrics.core" />
<allow pkg="org.apache.kafka.server"/>
<allow pkg="org.apache.kafka.storage.internals"/>
<allow pkg="org.apache.kafka.common" />
Expand Down
2 changes: 1 addition & 1 deletion checkstyle/suppressions.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@
<suppress checks="(NPathComplexity|ClassFanOutComplexity|CyclomaticComplexity|ClassDataAbstractionCoupling|FinalLocalVariable|LocalVariableName|MemberName|ParameterName|MethodLength|JavaNCSS|AvoidStarImport)"
files="core[\\/]src[\\/](generated|generated-test)[\\/].+.java$"/>
<suppress checks="NPathComplexity" files="(ClusterTestExtensions|KafkaApisBuilder).java"/>
<suppress checks="NPathComplexity|ClassFanOutComplexity" files="RemoteLogManager.java"/>
<suppress checks="NPathComplexity|ClassFanOutComplexity" files="(RemoteLogManager|RemoteLogManagerTest).java"/>
<suppress checks="MethodLength"
files="(KafkaClusterTestKit).java"/>

Expand Down
51 changes: 44 additions & 7 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 @@ -16,10 +16,12 @@
*/
package kafka.log.remote;

import com.yammer.metrics.core.Gauge;
import kafka.cluster.EndPoint;
import kafka.cluster.Partition;
import kafka.log.LogSegment;
import kafka.log.UnifiedLog;
import kafka.server.BrokerTopicStats;
import kafka.server.KafkaConfig;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicIdPartition;
Expand Down Expand Up @@ -49,6 +51,7 @@
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentState;
import org.apache.kafka.server.log.remote.storage.RemoteStorageException;
import org.apache.kafka.server.log.remote.storage.RemoteStorageManager;
import org.apache.kafka.server.metrics.KafkaMetricsGroup;
import org.apache.kafka.storage.internals.checkpoint.InMemoryLeaderEpochCheckpoint;
import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache;
import org.apache.kafka.storage.internals.log.AbortedTxn;
Expand Down Expand Up @@ -120,12 +123,14 @@ 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;
private final Time time;
private final Function<TopicPartition, Optional<UnifiedLog>> fetchLog;
private final BrokerTopicStats brokerTopicStats;

private final RemoteStorageManager remoteLogStorageManager;

Expand All @@ -147,6 +152,8 @@ public class RemoteLogManager implements Closeable {
private Optional<EndPoint> endpoint = Optional.empty();
private boolean closed = false;

private KafkaMetricsGroup metricsGroup = new KafkaMetricsGroup(this.getClass());

/**
* Creates RemoteLogManager instance with the given arguments.
*
Expand All @@ -162,26 +169,42 @@ public RemoteLogManager(RemoteLogManagerConfig rlmConfig,
String logDir,
String clusterId,
Time time,
Function<TopicPartition, Optional<UnifiedLog>> fetchLog) throws IOException {
Function<TopicPartition, Optional<UnifiedLog>> fetchLog,
BrokerTopicStats brokerTopicStats) throws IOException {
this.rlmConfig = rlmConfig;
this.brokerId = brokerId;
this.logDir = logDir;
this.clusterId = clusterId;
this.time = time;
this.fetchLog = fetchLog;
this.brokerTopicStats = brokerTopicStats;

remoteLogStorageManager = createRemoteStorageManager();
remoteLogMetadataManager = createRemoteLogMetadataManager();
indexCache = new RemoteIndexCache(1024, remoteLogStorageManager, logDir);
delayInMs = rlmConfig.remoteLogManagerTaskIntervalMs();
rlmScheduledThreadPool = new RLMScheduledThreadPool(rlmConfig.remoteLogManagerThreadPoolSize());

metricsGroup.newGauge(REMOTE_LOG_MANAGER_TASKS_AVG_IDLE_PERCENT, new Gauge<Double>() {
@Override
public Double value() {
return rlmScheduledThreadPool.getIdlePercent();
}
});

remoteStorageReaderThreadPool = new RemoteStorageThreadPool(
REMOTE_LOG_READER_THREAD_NAME_PREFIX,
rlmConfig.remoteLogReaderThreads(),
rlmConfig.remoteLogReaderMaxPendingTasks()
rlmConfig.remoteLogReaderMaxPendingTasks(),
REMOTE_LOG_READER_METRICS_NAME_PREFIX
);
}

private void removeMetrics() {
metricsGroup.removeMetric(REMOTE_LOG_MANAGER_TASKS_AVG_IDLE_PERCENT);
remoteStorageReaderThreadPool.removeMetrics();
}

private <T> T createDelegate(ClassLoader classLoader, String className) {
try {
return (T) classLoader.loadClass(className)
Expand Down Expand Up @@ -553,6 +576,8 @@ public void copyLogSegmentsToRemote(UnifiedLog log) throws InterruptedException
throw ex;
} catch (Exception ex) {
if (!isCancelled()) {
brokerTopicStats.topicStats(log.topicPartition().topic()).failedRemoteWriteRequestRate().mark();
brokerTopicStats.allTopicsStats().failedRemoteWriteRequestRate().mark();
logger.error("Error occurred while copying log segments of partition: {}", topicIdPartition, ex);
}
}
Expand Down Expand Up @@ -594,13 +619,17 @@ private void copyLogSegment(UnifiedLog log, LogSegment segment, long nextSegment
LogSegmentData segmentData = new LogSegmentData(logFile.toPath(), toPathIfExists(segment.lazyOffsetIndex().get().file()),
toPathIfExists(segment.lazyTimeIndex().get().file()), Optional.ofNullable(toPathIfExists(segment.txnIndex().file())),
producerStateSnapshotFile.toPath(), leaderEpochsIndex);
brokerTopicStats.topicStats(log.topicPartition().topic()).remoteWriteRequestRate().mark();
brokerTopicStats.allTopicsStats().remoteWriteRequestRate().mark();
remoteLogStorageManager.copyLogSegmentData(copySegmentStartedRlsm, segmentData);

RemoteLogSegmentMetadataUpdate copySegmentFinishedRlsm = new RemoteLogSegmentMetadataUpdate(id, time.milliseconds(),
RemoteLogSegmentState.COPY_SEGMENT_FINISHED, brokerId);

remoteLogMetadataManager.updateRemoteLogSegmentMetadata(copySegmentFinishedRlsm).get();

brokerTopicStats.topicStats(log.topicPartition().topic())
.remoteBytesOutRate().mark(copySegmentStartedRlsm.segmentSizeInBytes());
brokerTopicStats.allTopicsStats().remoteBytesOutRate().mark(copySegmentStartedRlsm.segmentSizeInBytes());
copiedOffsetOption = OptionalLong.of(endOffset);
log.updateHighestOffsetInRemoteStorage(endOffset);
logger.info("Copied {} to remote storage with segment-id: {}", logFileName, copySegmentFinishedRlsm.remoteLogSegmentId());
Expand Down Expand Up @@ -845,7 +874,7 @@ long findHighestRemoteOffset(TopicIdPartition topicIdPartition, UnifiedLog log)
* @throws java.util.concurrent.RejectedExecutionException if the task cannot be accepted for execution (task queue is full)
*/
public Future<Void> asyncRead(RemoteStorageFetchInfo fetchInfo, Consumer<RemoteLogReadResult> callback) {
return remoteStorageReaderThreadPool.submit(new RemoteLogReader(fetchInfo, this, callback));
return remoteStorageReaderThreadPool.submit(new RemoteLogReader(fetchInfo, this, callback, brokerTopicStats));
}

void doHandleLeaderOrFollowerPartitions(TopicIdPartition topicPartition,
Expand Down Expand Up @@ -896,7 +925,11 @@ public void close() {
Utils.closeQuietly(indexCache, "RemoteIndexCache");

rlmScheduledThreadPool.close();
shutdownAndAwaitTermination(remoteStorageReaderThreadPool, "RemoteStorageReaderThreadPool", 10, TimeUnit.SECONDS);
try {
shutdownAndAwaitTermination(remoteStorageReaderThreadPool, "RemoteStorageReaderThreadPool", 10, TimeUnit.SECONDS);
} finally {
removeMetrics();
}

leaderOrFollowerTasks.clear();
closed = true;
Expand Down Expand Up @@ -955,6 +988,10 @@ public Thread newThread(Runnable r) {
return threadPool;
}

public Double getIdlePercent() {
return 1 - (double) scheduledThreadPool.getActiveCount() / (double) scheduledThreadPool.getCorePoolSize();
Comment thread
divijvaidya marked this conversation as resolved.
Outdated
}

public ScheduledFuture<?> scheduleWithFixedDelay(Runnable runnable, long initialDelay, long delay, TimeUnit timeUnit) {
LOGGER.info("Scheduling runnable {} with initial delay: {}, fixed delay: {}", runnable, initialDelay, delay);
return scheduledThreadPool.scheduleWithFixedDelay(runnable, initialDelay, delay, timeUnit);
Expand All @@ -965,4 +1002,4 @@ public void close() {
}
}

}
}
12 changes: 11 additions & 1 deletion core/src/main/java/kafka/log/remote/RemoteLogReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
*/
package kafka.log.remote;

import kafka.server.BrokerTopicStats;
import org.apache.kafka.common.errors.OffsetOutOfRangeException;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.storage.internals.log.FetchDataInfo;
Expand All @@ -31,14 +32,19 @@ public class RemoteLogReader implements Callable<Void> {
private final Logger logger;
private final RemoteStorageFetchInfo fetchInfo;
private final RemoteLogManager rlm;
private final BrokerTopicStats brokerTopicStats;
private final Consumer<RemoteLogReadResult> callback;

public RemoteLogReader(RemoteStorageFetchInfo fetchInfo,
RemoteLogManager rlm,
Consumer<RemoteLogReadResult> callback) {
Consumer<RemoteLogReadResult> callback,
BrokerTopicStats brokerTopicStats) {
this.fetchInfo = fetchInfo;
this.rlm = rlm;
this.brokerTopicStats = brokerTopicStats;
this.callback = callback;
this.brokerTopicStats.topicStats(fetchInfo.topicPartition.topic()).remoteReadRequestRate().mark();
this.brokerTopicStats.allTopicsStats().remoteReadRequestRate().mark();
logger = new LogContext() {
@Override
public String logPrefix() {
Expand All @@ -54,10 +60,14 @@ public Void call() {
logger.debug("Reading records from remote storage for topic partition {}", fetchInfo.topicPartition);

FetchDataInfo fetchDataInfo = rlm.read(fetchInfo);
brokerTopicStats.topicStats(fetchInfo.topicPartition.topic()).remoteBytesInRate().mark(fetchDataInfo.records.sizeInBytes());
brokerTopicStats.allTopicsStats().remoteBytesInRate().mark(fetchDataInfo.records.sizeInBytes());
result = new RemoteLogReadResult(Optional.of(fetchDataInfo), Optional.empty());
} catch (OffsetOutOfRangeException e) {
result = new RemoteLogReadResult(Optional.empty(), Optional.of(e));

@Staniel Staniel Jul 17, 2023

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Hi, came across this PR and have a qq, why we don't include the OffsetOutOfRangeException result into the failed remote read request rate metrics? If for some reason we can not, can we have a separate metrics to capture that error in metrics?

@abhijeetk88 abhijeetk88 Jul 18, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

We do not want to include OffsetOutOfRangeException because it is not really a failure in remote reads.

We can have a separate discussion on a different metric since that change is unrelated to this PR.

@Staniel Staniel Jul 18, 2023

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Thank you. I will create a JIRA to track that discussion.

@Staniel Staniel Jul 18, 2023

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

https://issues.apache.org/jira/browse/KAFKA-15214

} catch (Exception e) {
brokerTopicStats.topicStats(fetchInfo.topicPartition.topic()).failedRemoteReadRequestRate().mark();
brokerTopicStats.allTopicsStats().failedRemoteReadRequestRate().mark();
logger.error("Error occurred while reading the remote data for {}", fetchInfo.topicPartition, e);
result = new RemoteLogReadResult(Optional.empty(), Optional.of(e));
}
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/kafka/log/UnifiedLog.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,7 @@ import scala.jdk.CollectionConverters._
@threadsafe
class UnifiedLog(@volatile var logStartOffset: Long,
private val localLog: LocalLog,
brokerTopicStats: BrokerTopicStats,
val brokerTopicStats: BrokerTopicStats,
val producerIdExpirationCheckIntervalMs: Int,
@volatile var leaderEpochCache: Option[LeaderEpochFileCache],
val producerStateManager: ProducerStateManager,
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/kafka/server/BrokerServer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -571,7 +571,7 @@ class BrokerServer(
}

Some(new RemoteLogManager(config.remoteLogManagerConfig, config.brokerId, config.logDirs.head, clusterId, time,
(tp: TopicPartition) => logManager.getLog(tp).asJava));
(tp: TopicPartition) => logManager.getLog(tp).asJava, brokerTopicStats));
} else {
None
}
Expand Down
30 changes: 30 additions & 0 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 @@ -277,6 +277,12 @@ class BrokerTopicMetrics(name: Option[String]) {
BrokerTopicStats.TotalFetchRequestsPerSec -> MeterWrapper(BrokerTopicStats.TotalFetchRequestsPerSec, "requests"),
BrokerTopicStats.FetchMessageConversionsPerSec -> MeterWrapper(BrokerTopicStats.FetchMessageConversionsPerSec, "requests"),
BrokerTopicStats.ProduceMessageConversionsPerSec -> MeterWrapper(BrokerTopicStats.ProduceMessageConversionsPerSec, "requests"),
BrokerTopicStats.RemoteBytesOutPerSec -> MeterWrapper(BrokerTopicStats.RemoteBytesOutPerSec, "bytes"),

@divijvaidya divijvaidya Jul 3, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

We should probably add Remote metrics only when RemoteStorage is enabled on the cluster. Otherwise these these per-topic metrics are useless.

@showuon showuon Jul 5, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Good point!

@abhijeetk88 abhijeetk88 Jul 8, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

This requires a significant amount of changes. We need to know in this class if remote storage is enabled and for that to happen BrokerTopicStats needs to know that. There are many instances of object creation for BrokerTopicStats all of which will need to supply whether remote storage is enabled.

Should we address this in a separate PR? Let me know if you have better ideas to do this.

@showuon showuon Jul 10, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Fine. Please open a JIRA ticket for this improvement. Thanks.

@divijvaidya divijvaidya Jul 14, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

I wasn't able to find the JIRA, so when ahead created on https://issues.apache.org/jira/browse/KAFKA-15189 . This JIRA is a blocker for 3.6 launch otherwise it will leak to regression for customers not using the Remote Storage feature,

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"),
Comment on lines +282 to +285

@jeqo jeqo Jul 10, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

few more naming suggestions:

Suggested change
BrokerTopicStats.RemoteReadRequestsPerSec -> MeterWrapper(BrokerTopicStats.RemoteReadRequestsPerSec, "requests"),
BrokerTopicStats.RemoteWriteRequestsPerSec -> MeterWrapper(BrokerTopicStats.RemoteWriteRequestsPerSec, "requests"),
BrokerTopicStats.FailedRemoteReadRequestsPerSec -> MeterWrapper(BrokerTopicStats.FailedRemoteReadRequestsPerSec, "requests"),
BrokerTopicStats.FailedRemoteWriteRequestsPerSec -> MeterWrapper(BrokerTopicStats.FailedRemoteWriteRequestsPerSec, "requests"),
BrokerTopicStats.RemoteFetchRequestsPerSec -> MeterWrapper(BrokerTopicStats.RemoteFetchRequestsPerSec, "requests"),
BrokerTopicStats.RemoteCopyRequestsPerSec -> MeterWrapper(BrokerTopicStats.RemoteCopyRequestsPerSec, "requests"),
BrokerTopicStats.FailedRemoteCopyRequestsPerSec -> MeterWrapper(BrokerTopicStats.FailedRemoteFetchRequestsPerSec, "requests"),
BrokerTopicStats.FailedRemoteCopyRequestsPerSec -> MeterWrapper(BrokerTopicStats.FailedRemoteCopyRequestsPerSec, "requests"),

@divijvaidya divijvaidya Jul 12, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Note that the KIP version was voted with the earlier names and now changing the KIP would be equivalent to breaking the agreed contract when the KIP was accepted. I will propose to keep the metric names as they were in the KIP. If we want to introduce new ones, we can start a new KIP which deprecates these metrics and uses new ones.

@jeqo jeqo Jul 12, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I was suspecting this would be the case. Though as metrics haven't been implemented yet, would it be possible to amend the KIP and ask for feedback?

@divijvaidya divijvaidya Jul 12, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

We at (Amazon MSK) have an implementation against the KIP-405 interfaces [1] and we explicitly tell the customers that the metric such as RemoteBytesInPerSec are as per KIP-405 contract. If we change the KIP now, it would be a break in contract for our customers. Hence, I would not be in favour of amending the accepted KIP at this time.

[1] see: RemoteBytesInPerSec at https://docs.aws.amazon.com/msk/latest/developerguide/metrics-details.html

@satishd satishd Jul 13, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

We should avoid having multiple versions of metrics here. As it is not yet released in Apache, we should have better naming agreeable for the majority. We do not need to have two versions here when it is released in Apache. It avoids confusion for users too. We can discuss whether that can be done by updating KIP-405(I guess I did update d a couple of them) or we can have a minor KIP for them.

@divijvaidya This small change can be handled by organizations that adopted KIP-405.

BrokerTopicStats.NoKeyCompactedTopicRecordsPerSec -> MeterWrapper(BrokerTopicStats.NoKeyCompactedTopicRecordsPerSec, "requests"),
BrokerTopicStats.InvalidMagicNumberRecordsPerSec -> MeterWrapper(BrokerTopicStats.InvalidMagicNumberRecordsPerSec, "requests"),
BrokerTopicStats.InvalidMessageCrcRecordsPerSec -> MeterWrapper(BrokerTopicStats.InvalidMessageCrcRecordsPerSec, "requests"),
Expand Down Expand Up @@ -336,6 +342,18 @@ class BrokerTopicMetrics(name: Option[String]) {

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

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

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

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

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

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

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

def closeMetric(metricType: String): Unit = {
val meter = metricTypeMap.get(metricType)
if (meter != null)
Expand All @@ -360,6 +378,12 @@ 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 @@ -415,6 +439,12 @@ class BrokerTopicStats extends Logging {
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)
}
}

Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/kafka/server/KafkaServer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -614,7 +614,7 @@ class KafkaServer(
}

Some(new RemoteLogManager(config.remoteLogManagerConfig, config.brokerId, config.logDirs.head, clusterId, time,
(tp: TopicPartition) => logManager.getLog(tp).asJava));
(tp: TopicPartition) => logManager.getLog(tp).asJava, brokerTopicStats));
} else {
None
}
Expand Down
Loading