MINOR: fix bug where we weren't registering SnapshotEmitterMetrics (a…

…pache#14918) Fix a bug where we weren't properly exposing SnapshotEmitterMetrics. Add a test. Reviewers: David Arthur <[email protected]>
hyk001 · Dec 5, 2023 · ebae7b2 · ebae7b2
1 parent 463ed09
commit ebae7b2
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 1 deletion.
diff --git a/core/src/main/scala/kafka/server/SharedServer.scala b/core/src/main/scala/kafka/server/SharedServer.scala
@@ -29,6 +29,7 @@ import org.apache.kafka.image.MetadataProvenance
 import org.apache.kafka.image.loader.MetadataLoader
 import org.apache.kafka.image.loader.metrics.MetadataLoaderMetrics
 import org.apache.kafka.image.publisher.{SnapshotEmitter, SnapshotGenerator}
+import org.apache.kafka.image.publisher.metrics.SnapshotEmitterMetrics
 import org.apache.kafka.metadata.MetadataRecordSerde
 import org.apache.kafka.metadata.properties.MetaPropertiesEnsemble
 import org.apache.kafka.raft.RaftConfig.AddressSpec
@@ -289,6 +290,8 @@ class SharedServer(
  snapshotEmitter = new SnapshotEmitter.Builder().
  setNodeId(nodeId).
  setRaftClient(_raftManager.client).
+ setMetrics(new SnapshotEmitterMetrics(
+ Optional.of(KafkaYammerMetrics.defaultRegistry()), time)).
  build()
  snapshotGenerator = new SnapshotGenerator.Builder(snapshotEmitter).
  setNodeId(nodeId).

diff --git a/core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala b/core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala
@@ -984,6 +984,11 @@ class KRaftClusterTest {
  val countAfterTenIntervals = snapshotCounter(metaLog)
  assertTrue(countAfterTenIntervals > 1, s"Expected to see at least one more snapshot, saw $countAfterTenIntervals")
  assertTrue(countAfterTenIntervals < 20, s"Did not expect to see more than twice as many snapshots as snapshot intervals, saw $countAfterTenIntervals")
+ TestUtils.waitUntilTrue(() => {
+ val emitterMetrics = cluster.controllers().values().iterator().next().
+ sharedServer.snapshotEmitter.metrics()
+ emitterMetrics.latestSnapshotGeneratedBytes() > 0
+ }, "Failed to see latestSnapshotGeneratedBytes > 0")
  } finally {
  cluster.close()
  }

diff --git a/metadata/src/main/java/org/apache/kafka/image/publisher/SnapshotEmitter.java b/metadata/src/main/java/org/apache/kafka/image/publisher/SnapshotEmitter.java
@@ -128,7 +128,7 @@ private SnapshotEmitter(
  this.metrics = metrics;
  }
 
- SnapshotEmitterMetrics metrics() {
+ public SnapshotEmitterMetrics metrics() {
  return metrics;
  }