You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by cm...@apache.org on 2023/07/24 20:40:13 UTC
[kafka] branch KAFKA-15183 updated: address review comments, fix SnapshotEmitterMetrics
This is an automated email from the ASF dual-hosted git repository.
cmccabe pushed a commit to branch KAFKA-15183
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/KAFKA-15183 by this push:
new 5e858f7f89c address review comments, fix SnapshotEmitterMetrics
5e858f7f89c is described below
commit 5e858f7f89c90915a9f29feb8e8843f313faaa89
Author: Colin P. McCabe <cm...@apache.org>
AuthorDate: Wed Jul 19 14:54:43 2023 -0700
address review comments, fix SnapshotEmitterMetrics
---
.../kafka/server/KRaftClusterTest.scala | 4 +-
.../apache/kafka/controller/QuorumController.java | 13 ---
.../metrics/QuorumControllerMetrics.java | 24 +----
.../kafka/image/publisher/SnapshotEmitter.java | 51 ++++++++--
.../publisher/metrics/SnapshotEmitterMetrics.java | 14 +--
.../kafka/image/writer/RaftSnapshotWriter.java | 11 +-
.../QuorumControllerIntegrationTestUtils.java | 4 +-
.../metrics/QuorumControllerMetricsTest.java | 45 ++++++++-
.../org/apache/kafka/image/FakeSnapshotWriter.java | 93 +++++++++++++++++
.../loader/metrics/MetadataLoaderMetricsTest.java | 5 +-
.../kafka/image/publisher/SnapshotEmitterTest.java | 75 +++-----------
.../metrics/SnapshotEmitterMetricsTest.java | 111 +++++++++++++++++++++
.../kafka/image/writer/RaftSnapshotWriterTest.java | 63 ++----------
.../kafka/snapshot/RecordsSnapshotWriter.java | 3 +-
.../org/apache/kafka/snapshot/SnapshotWriter.java | 4 +-
15 files changed, 346 insertions(+), 174 deletions(-)
diff --git a/core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala b/core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala
index 37bd66cd2ae..242c7548853 100644
--- a/core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala
+++ b/core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala
@@ -35,7 +35,7 @@ import org.apache.kafka.common.quota.{ClientQuotaAlteration, ClientQuotaEntity,
import org.apache.kafka.common.requests.{ApiError, DescribeClusterRequest, DescribeClusterResponse}
import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.kafka.common.{Cluster, Endpoint, Reconfigurable, TopicPartition, TopicPartitionInfo}
-import org.apache.kafka.controller.QuorumController
+import org.apache.kafka.controller.{QuorumController, QuorumControllerIntegrationTestUtils}
import org.apache.kafka.image.ClusterImage
import org.apache.kafka.metadata.BrokerState
import org.apache.kafka.server.authorizer._
@@ -1153,7 +1153,7 @@ class KRaftClusterTest {
val controller = cluster.controllers().values().iterator().next()
controller.controller.waitForReadyBrokers(3).get()
TestUtils.retry(60000) {
- val latch = controller.controller.asInstanceOf[QuorumController].pause()
+ val latch = QuorumControllerIntegrationTestUtils.pause(controller.asInstanceOf[QuorumController])
Thread.sleep(1001)
latch.countDown()
assertEquals(0, controller.sharedServer.controllerServerMetrics.fencedBrokerCount())
diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
index 0ee0106841c..7b80ef5d09e 100644
--- a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
+++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
@@ -2320,19 +2320,6 @@ public final class QuorumController implements Controller {
controllerMetrics.close();
}
- // VisibleForTesting
- public CountDownLatch pause() {
- final CountDownLatch latch = new CountDownLatch(1);
- appendControlEvent("pause", () -> {
- try {
- latch.await();
- } catch (InterruptedException e) {
- log.info("Interrupted while waiting for unpause.", e);
- }
- });
- return latch;
- }
-
// VisibleForTesting
Time time() {
return time;
diff --git a/metadata/src/main/java/org/apache/kafka/controller/metrics/QuorumControllerMetrics.java b/metadata/src/main/java/org/apache/kafka/controller/metrics/QuorumControllerMetrics.java
index 1767874916c..6c7aa581a79 100644
--- a/metadata/src/main/java/org/apache/kafka/controller/metrics/QuorumControllerMetrics.java
+++ b/metadata/src/main/java/org/apache/kafka/controller/metrics/QuorumControllerMetrics.java
@@ -45,7 +45,7 @@ public class QuorumControllerMetrics implements AutoCloseable {
private final static MetricName EVENT_QUEUE_PROCESSING_TIME_MS = getMetricName(
"ControllerEventManager", "EventQueueProcessingTimeMs");
private final static MetricName ZK_WRITE_BEHIND_LAG = getMetricName(
- "KafkaController", "ZKWriteBehindLag");
+ "KafkaController", "ZKWriteBehindLag");
private final static MetricName LAST_APPLIED_RECORD_OFFSET = getMetricName(
"KafkaController", "LastAppliedRecordOffset");
private final static MetricName LAST_COMMITTED_RECORD_OFFSET = getMetricName(
@@ -59,7 +59,7 @@ public class QuorumControllerMetrics implements AutoCloseable {
private final static MetricName EVENT_QUEUE_OPERATIONS_STARTED_COUNT = getMetricName(
"KafkaController", "EventQueueOperationsStartedCount");
private final static MetricName EVENT_QUEUE_OPERATIONS_TIMED_OUT_COUNT = getMetricName(
- "KafkaController", "EventQueueOperationsTimedOutCount");
+ "KafkaController", "EventQueueOperationsTimedOutCount");
private final static MetricName NEW_ACTIVE_CONTROLLERS_COUNT = getMetricName(
"KafkaController", "NewActiveControllersCount");
@@ -133,13 +133,13 @@ public class QuorumControllerMetrics implements AutoCloseable {
registry.ifPresent(r -> r.newGauge(EVENT_QUEUE_OPERATIONS_STARTED_COUNT, new Gauge<Long>() {
@Override
public Long value() {
- return timedOutHeartbeats();
+ return operationsStarted();
}
}));
registry.ifPresent(r -> r.newGauge(EVENT_QUEUE_OPERATIONS_TIMED_OUT_COUNT, new Gauge<Long>() {
@Override
public Long value() {
- return timedOutHeartbeats();
+ return operationsTimedOut();
}
}));
registry.ifPresent(r -> r.newGauge(NEW_ACTIVE_CONTROLLERS_COUNT, new Gauge<Long>() {
@@ -213,10 +213,6 @@ public class QuorumControllerMetrics implements AutoCloseable {
timedOutHeartbeats.incrementAndGet();
}
- public void setTimedOutHeartbeats(long value) {
- timedOutHeartbeats.set(value);
- }
-
public long timedOutHeartbeats() {
return timedOutHeartbeats.get();
}
@@ -225,10 +221,6 @@ public class QuorumControllerMetrics implements AutoCloseable {
operationsStarted.incrementAndGet();
}
- public void setOperationsStarted(long value) {
- operationsStarted.set(value);
- }
-
public long operationsStarted() {
return operationsStarted.get();
}
@@ -237,10 +229,6 @@ public class QuorumControllerMetrics implements AutoCloseable {
operationsTimedOut.incrementAndGet();
}
- public void setOperationsTimedOut(long value) {
- operationsTimedOut.set(value);
- }
-
public long operationsTimedOut() {
return operationsTimedOut.get();
}
@@ -249,10 +237,6 @@ public class QuorumControllerMetrics implements AutoCloseable {
newActiveControllers.incrementAndGet();
}
- public void setNewActiveControllers(long value) {
- newActiveControllers.set(value);
- }
-
public long newActiveControllers() {
return newActiveControllers.get();
}
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
index ec46bcc06fa..8ab224f91f2 100644
--- a/metadata/src/main/java/org/apache/kafka/image/publisher/SnapshotEmitter.java
+++ b/metadata/src/main/java/org/apache/kafka/image/publisher/SnapshotEmitter.java
@@ -18,9 +18,11 @@
package org.apache.kafka.image.publisher;
import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.image.MetadataImage;
import org.apache.kafka.image.MetadataProvenance;
+import org.apache.kafka.image.publisher.metrics.SnapshotEmitterMetrics;
import org.apache.kafka.image.writer.ImageWriterOptions;
import org.apache.kafka.image.writer.RaftSnapshotWriter;
import org.apache.kafka.raft.RaftClient;
@@ -43,9 +45,16 @@ public class SnapshotEmitter implements SnapshotGenerator.Emitter {
private final static int DEFAULT_BATCH_SIZE = 1024;
public static class Builder {
+ private Time time = Time.SYSTEM;
private int nodeId = 0;
private RaftClient<ApiMessageAndVersion> raftClient = null;
private int batchSize = DEFAULT_BATCH_SIZE;
+ private SnapshotEmitterMetrics metrics = null;
+
+ public Builder setTime(Time time) {
+ this.time = time;
+ return this;
+ }
public Builder setNodeId(int nodeId) {
this.nodeId = nodeId;
@@ -62,11 +71,21 @@ public class SnapshotEmitter implements SnapshotGenerator.Emitter {
return this;
}
+ public Builder setMetrics(SnapshotEmitterMetrics metrics) {
+ this.metrics = metrics;
+ return this;
+ }
+
public SnapshotEmitter build() {
if (raftClient == null) throw new RuntimeException("You must set the raftClient.");
- return new SnapshotEmitter(nodeId,
+ if (metrics == null) metrics = new SnapshotEmitterMetrics(
+ Optional.empty(),
+ time);
+ return new SnapshotEmitter(time,
+ nodeId,
raftClient,
- batchSize);
+ batchSize,
+ metrics);
}
}
@@ -75,6 +94,11 @@ public class SnapshotEmitter implements SnapshotGenerator.Emitter {
*/
private final Logger log;
+ /**
+ * The clock object.
+ */
+ private final Time time;
+
/**
* The RaftClient to use.
*/
@@ -85,14 +109,27 @@ public class SnapshotEmitter implements SnapshotGenerator.Emitter {
*/
private final int batchSize;
+ /**
+ * The metrics to use.
+ */
+ private final SnapshotEmitterMetrics metrics;
+
private SnapshotEmitter(
- int nodeId,
- RaftClient<ApiMessageAndVersion> raftClient,
- int batchSize
+ Time time,
+ int nodeId,
+ RaftClient<ApiMessageAndVersion> raftClient,
+ int batchSize,
+ SnapshotEmitterMetrics metrics
) {
+ this.time = time;
this.log = new LogContext("[SnapshotEmitter id=" + nodeId + "] ").logger(SnapshotEmitter.class);
this.raftClient = raftClient;
this.batchSize = batchSize;
+ this.metrics = metrics;
+ }
+
+ SnapshotEmitterMetrics metrics() {
+ return metrics;
}
@Override
@@ -112,6 +149,9 @@ public class SnapshotEmitter implements SnapshotGenerator.Emitter {
setMetadataVersion(image.features().metadataVersion()).
build());
writer.close(true);
+ metrics.setLatestSnapshotGeneratedTimeMs(time.milliseconds());
+ metrics.setLatestSnapshotGeneratedBytes(writer.frozenSize().getAsLong());
+ log.info("Successfully wrote {}", provenance.snapshotName());
} catch (Throwable e) {
log.error("Encountered error while writing {}", provenance.snapshotName(), e);
throw e;
@@ -119,6 +159,5 @@ public class SnapshotEmitter implements SnapshotGenerator.Emitter {
Utils.closeQuietly(writer, "RaftSnapshotWriter");
Utils.closeQuietly(snapshotWriter.get(), "SnapshotWriter");
}
- log.info("Successfully wrote {}", provenance.snapshotName());
}
}
diff --git a/metadata/src/main/java/org/apache/kafka/image/publisher/metrics/SnapshotEmitterMetrics.java b/metadata/src/main/java/org/apache/kafka/image/publisher/metrics/SnapshotEmitterMetrics.java
index 0a2cd308463..5e59942c960 100644
--- a/metadata/src/main/java/org/apache/kafka/image/publisher/metrics/SnapshotEmitterMetrics.java
+++ b/metadata/src/main/java/org/apache/kafka/image/publisher/metrics/SnapshotEmitterMetrics.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.kafka.image.loader.metrics;
+package org.apache.kafka.image.publisher.metrics;
import com.yammer.metrics.core.Gauge;
import com.yammer.metrics.core.MetricName;
@@ -25,7 +25,6 @@ import org.apache.kafka.server.metrics.KafkaYammerMetrics;
import java.util.Arrays;
import java.util.Optional;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
/**
@@ -49,13 +48,12 @@ public final class SnapshotEmitterMetrics implements AutoCloseable {
*/
public SnapshotEmitterMetrics(
Optional<MetricsRegistry> registry,
- Time time,
- long initialLatestSnapshotGeneratedBytes
+ Time time
) {
this.registry = registry;
this.time = time;
- this.latestSnapshotGeneratedBytes = new AtomicLong(initialLatestSnapshotGeneratedBytes);
- this.latestSnapshotGeneratedTimeMs = new AtomicLong(monoTimeInMs());
+ this.latestSnapshotGeneratedBytes = new AtomicLong(0L);
+ this.latestSnapshotGeneratedTimeMs = new AtomicLong(time.milliseconds());
registry.ifPresent(r -> r.newGauge(LATEST_SNAPSHOT_GENERATED_BYTES, new Gauge<Long>() {
@Override
public Long value() {
@@ -70,10 +68,6 @@ public final class SnapshotEmitterMetrics implements AutoCloseable {
}));
}
- long monoTimeInMs() {
- return TimeUnit.NANOSECONDS.toMillis(time.nanoseconds());
- }
-
public void setLatestSnapshotGeneratedBytes(long value) {
this.latestSnapshotGeneratedBytes.set(value);
}
diff --git a/metadata/src/main/java/org/apache/kafka/image/writer/RaftSnapshotWriter.java b/metadata/src/main/java/org/apache/kafka/image/writer/RaftSnapshotWriter.java
index e1faad90dc3..eec5cd90c76 100644
--- a/metadata/src/main/java/org/apache/kafka/image/writer/RaftSnapshotWriter.java
+++ b/metadata/src/main/java/org/apache/kafka/image/writer/RaftSnapshotWriter.java
@@ -22,6 +22,7 @@ import org.apache.kafka.snapshot.SnapshotWriter;
import java.util.ArrayList;
import java.util.List;
+import java.util.OptionalLong;
/**
@@ -31,6 +32,7 @@ public class RaftSnapshotWriter implements ImageWriter {
private final SnapshotWriter<ApiMessageAndVersion> snapshotWriter;
private final int batchSize;
private List<ApiMessageAndVersion> records;
+ private OptionalLong frozenSize = OptionalLong.empty();
public RaftSnapshotWriter(
SnapshotWriter<ApiMessageAndVersion> snapshotWriter,
@@ -59,11 +61,18 @@ public class RaftSnapshotWriter implements ImageWriter {
if (!records.isEmpty()) {
snapshotWriter.append(records);
}
- snapshotWriter.freeze();
+ frozenSize = OptionalLong.of(snapshotWriter.freeze());
}
} finally {
records = null;
snapshotWriter.close();
}
}
+
+ /**
+ * @return the frozen size of the snapshot, or OptionalLong.empty if the snapshot was not frozen.
+ */
+ public OptionalLong frozenSize() {
+ return frozenSize;
+ }
}
diff --git a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerIntegrationTestUtils.java b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerIntegrationTestUtils.java
index 83758c0fff1..aec2a4a513f 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerIntegrationTestUtils.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerIntegrationTestUtils.java
@@ -47,7 +47,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
/**
* Utility functions for use in QuorumController integration tests.
*/
-class QuorumControllerIntegrationTestUtils {
+public class QuorumControllerIntegrationTestUtils {
private final static Logger log = LoggerFactory.getLogger(QuorumControllerIntegrationTestUtils.class);
BrokerRegistrationRequestData.FeatureCollection brokerFeatures() {
@@ -185,7 +185,7 @@ class QuorumControllerIntegrationTestUtils {
* @param controller The controller.
* @return The latch that can be used to unpause the controller.
*/
- static CountDownLatch pause(QuorumController controller) {
+ public static CountDownLatch pause(QuorumController controller) {
final CountDownLatch latch = new CountDownLatch(1);
controller.appendControlEvent("pause", () -> {
try {
diff --git a/metadata/src/test/java/org/apache/kafka/controller/metrics/QuorumControllerMetricsTest.java b/metadata/src/test/java/org/apache/kafka/controller/metrics/QuorumControllerMetricsTest.java
index 8aeb3831617..9258577a0dd 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/metrics/QuorumControllerMetricsTest.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/metrics/QuorumControllerMetricsTest.java
@@ -99,10 +99,23 @@ public class QuorumControllerMetricsTest {
MetricsRegistry registry = new MetricsRegistry();
MockTime time = new MockTime();
time.sleep(1000);
- try (QuorumControllerMetrics metrics = new QuorumControllerMetrics(Optional.of(registry), time, false)) {
+ try (QuorumControllerMetrics metrics = new QuorumControllerMetrics(Optional.of(registry), time, true)) {
metrics.setLastAppliedRecordOffset(100);
metrics.setLastAppliedRecordTimestamp(500);
metrics.setLastCommittedRecordOffset(50);
+ metrics.updateDualWriteOffset(40L);
+ for (int i = 0; i < 2; i++) {
+ metrics.incrementTimedOutHeartbeats();
+ }
+ for (int i = 0; i < 3; i++) {
+ metrics.incrementOperationsStarted();
+ }
+ for (int i = 0; i < 4; i++) {
+ metrics.incrementOperationsTimedOut();
+ }
+ for (int i = 0; i < 5; i++) {
+ metrics.incrementNewActiveControllers();
+ }
@SuppressWarnings("unchecked")
Gauge<Long> lastAppliedRecordOffset = (Gauge<Long>) registry
@@ -127,6 +140,36 @@ public class QuorumControllerMetricsTest {
.allMetrics()
.get(metricName("KafkaController", "LastCommittedRecordOffset"));
assertEquals(50, lastCommittedRecordOffset.value());
+
+ @SuppressWarnings("unchecked")
+ Gauge<Long> zkWriteBehindLag = (Gauge<Long>) registry
+ .allMetrics()
+ .get(metricName("KafkaController", "ZKWriteBehindLag"));
+ assertEquals(10L, zkWriteBehindLag.value());
+
+ @SuppressWarnings("unchecked")
+ Gauge<Long> timedOutBrokerHeartbeats = (Gauge<Long>) registry
+ .allMetrics()
+ .get(metricName("KafkaController", "TimedOutBrokerHeartbeatCount"));
+ assertEquals(2L, timedOutBrokerHeartbeats.value());
+
+ @SuppressWarnings("unchecked")
+ Gauge<Long> operationsStarted = (Gauge<Long>) registry
+ .allMetrics()
+ .get(metricName("KafkaController", "EventQueueOperationsStartedCount"));
+ assertEquals(3L, operationsStarted.value());
+
+ @SuppressWarnings("unchecked")
+ Gauge<Long> operationsTimedOut = (Gauge<Long>) registry
+ .allMetrics()
+ .get(metricName("KafkaController", "EventQueueOperationsTimedOutCount"));
+ assertEquals(4L, operationsTimedOut.value());
+
+ @SuppressWarnings("unchecked")
+ Gauge<Long> newActiveControllers = (Gauge<Long>) registry
+ .allMetrics()
+ .get(metricName("KafkaController", "NewActiveControllersCount"));
+ assertEquals(5L, newActiveControllers.value());
} finally {
registry.shutdown();
}
diff --git a/metadata/src/test/java/org/apache/kafka/image/FakeSnapshotWriter.java b/metadata/src/test/java/org/apache/kafka/image/FakeSnapshotWriter.java
new file mode 100644
index 00000000000..d76b1c81f21
--- /dev/null
+++ b/metadata/src/test/java/org/apache/kafka/image/FakeSnapshotWriter.java
@@ -0,0 +1,93 @@
+/*
+ * 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.image;
+
+import org.apache.kafka.raft.OffsetAndEpoch;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.snapshot.SnapshotWriter;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+
+public class FakeSnapshotWriter implements SnapshotWriter<ApiMessageAndVersion> {
+ private final OffsetAndEpoch snapshotId;
+ private List<List<ApiMessageAndVersion>> batches = new ArrayList<>();
+ private boolean frozen = false;
+ private boolean closed = false;
+
+ public List<List<ApiMessageAndVersion>> batches() {
+ List<List<ApiMessageAndVersion>> result = new ArrayList<>();
+ for (List<ApiMessageAndVersion> batch : batches) {
+ result.add(Collections.unmodifiableList(batch));
+ }
+ return Collections.unmodifiableList(result);
+ }
+
+ public FakeSnapshotWriter() {
+ this(new OffsetAndEpoch(100L, 10));
+ }
+
+ public FakeSnapshotWriter(OffsetAndEpoch snapshotId) {
+ this.snapshotId = snapshotId;
+ }
+
+ @Override
+ public OffsetAndEpoch snapshotId() {
+ return snapshotId;
+ }
+
+ @Override
+ public long lastContainedLogOffset() {
+ return snapshotId().offset() - 1;
+ }
+
+ @Override
+ public int lastContainedLogEpoch() {
+ return snapshotId().epoch();
+ }
+
+ @Override
+ public boolean isFrozen() {
+ return frozen;
+ }
+
+ @Override
+ public void append(List<ApiMessageAndVersion> batch) {
+ if (frozen) {
+ throw new IllegalStateException("Append not supported. Snapshot is already frozen.");
+ }
+ batches.add(batch);
+ }
+
+ @Override
+ public long freeze() {
+ frozen = true;
+ return batches.size() * 100;
+ }
+
+ @Override
+ public void close() {
+ closed = true;
+ }
+
+ public boolean isClosed() {
+ return closed;
+ }
+}
\ No newline at end of file
diff --git a/metadata/src/test/java/org/apache/kafka/image/loader/metrics/MetadataLoaderMetricsTest.java b/metadata/src/test/java/org/apache/kafka/image/loader/metrics/MetadataLoaderMetricsTest.java
index c8d33754106..5e3c6c6f571 100644
--- a/metadata/src/test/java/org/apache/kafka/image/loader/metrics/MetadataLoaderMetricsTest.java
+++ b/metadata/src/test/java/org/apache/kafka/image/loader/metrics/MetadataLoaderMetricsTest.java
@@ -107,8 +107,9 @@ public class MetadataLoaderMetricsTest {
public void testUpdateLastAppliedImageProvenance() {
MetricsRegistry registry = new MetricsRegistry();
try (FakeMetadataLoaderMetrics fakeMetrics = new FakeMetadataLoaderMetrics(registry)) {
- fakeMetrics.metrics.updateLastAppliedImageProvenance(new MetadataProvenance(1L, 2, 3L));
- assertEquals(new MetadataProvenance(1L, 2, 3L), fakeMetrics.provenance.get());
+ MetadataProvenance provenance = new MetadataProvenance(1L, 2, 3L);
+ fakeMetrics.metrics.updateLastAppliedImageProvenance(provenance);
+ assertEquals(provenance, fakeMetrics.provenance.get());
}
}
diff --git a/metadata/src/test/java/org/apache/kafka/image/publisher/SnapshotEmitterTest.java b/metadata/src/test/java/org/apache/kafka/image/publisher/SnapshotEmitterTest.java
index ca72aa058ef..be4285f6bbf 100644
--- a/metadata/src/test/java/org/apache/kafka/image/publisher/SnapshotEmitterTest.java
+++ b/metadata/src/test/java/org/apache/kafka/image/publisher/SnapshotEmitterTest.java
@@ -17,6 +17,8 @@
package org.apache.kafka.image.publisher;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.image.FakeSnapshotWriter;
import org.apache.kafka.image.MetadataImageTest;
import org.apache.kafka.raft.LeaderAndEpoch;
import org.apache.kafka.raft.OffsetAndEpoch;
@@ -26,7 +28,6 @@ import org.apache.kafka.snapshot.SnapshotWriter;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
-import java.util.ArrayList;
import java.util.List;
import java.util.NavigableSet;
import java.util.Optional;
@@ -43,7 +44,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
@Timeout(value = 40)
public class SnapshotEmitterTest {
static class MockRaftClient implements RaftClient<ApiMessageAndVersion> {
- TreeMap<OffsetAndEpoch, MockSnapshotWriter> writers = new TreeMap<>();
+ TreeMap<OffsetAndEpoch, FakeSnapshotWriter> writers = new TreeMap<>();
@Override
public void initialize() {
@@ -103,7 +104,7 @@ public class SnapshotEmitterTest {
if (writers.containsKey(snapshotId)) {
return Optional.empty();
}
- MockSnapshotWriter writer = new MockSnapshotWriter(snapshotId);
+ FakeSnapshotWriter writer = new FakeSnapshotWriter(snapshotId);
writers.put(snapshotId, writer);
return Optional.of(writer);
}
@@ -124,72 +125,24 @@ public class SnapshotEmitterTest {
}
}
- static class MockSnapshotWriter implements SnapshotWriter<ApiMessageAndVersion> {
- private final OffsetAndEpoch snapshotId;
- private boolean frozen = false;
- private boolean closed = false;
- private final List<List<ApiMessageAndVersion>> batches;
-
- MockSnapshotWriter(OffsetAndEpoch snapshotId) {
- this.snapshotId = snapshotId;
- this.batches = new ArrayList<>();
- }
-
- @Override
- public OffsetAndEpoch snapshotId() {
- return snapshotId;
- }
-
- @Override
- public long lastContainedLogOffset() {
- return snapshotId.offset() - 1;
- }
-
- @Override
- public int lastContainedLogEpoch() {
- return snapshotId.epoch();
- }
-
- @Override
- public boolean isFrozen() {
- return frozen;
- }
-
- @Override
- public void append(List<ApiMessageAndVersion> records) {
- batches.add(records);
- }
-
- List<List<ApiMessageAndVersion>> batches() {
- List<List<ApiMessageAndVersion>> results = new ArrayList<>();
- batches.forEach(batch -> results.add(new ArrayList<>(batch)));
- return results;
- }
-
- @Override
- public void freeze() {
- frozen = true;
- }
-
- @Override
- public void close() {
- closed = true;
- }
-
- boolean isClosed() {
- return closed;
- }
- }
-
@Test
public void testEmit() throws Exception {
MockRaftClient mockRaftClient = new MockRaftClient();
+ MockTime time = new MockTime(0, 10000L, 20000L);
SnapshotEmitter emitter = new SnapshotEmitter.Builder().
+ setTime(time).
setBatchSize(2).
setRaftClient(mockRaftClient).
build();
+ assertEquals(0L, emitter.metrics().latestSnapshotGeneratedAgeMs());
+ assertEquals(0L, emitter.metrics().latestSnapshotGeneratedBytes());
+ time.sleep(30000L);
+ assertEquals(30000L, emitter.metrics().latestSnapshotGeneratedAgeMs());
+ assertEquals(0L, emitter.metrics().latestSnapshotGeneratedBytes());
emitter.maybeEmit(MetadataImageTest.IMAGE1);
- MockSnapshotWriter writer = mockRaftClient.writers.get(
+ assertEquals(0L, emitter.metrics().latestSnapshotGeneratedAgeMs());
+ assertEquals(1400L, emitter.metrics().latestSnapshotGeneratedBytes());
+ FakeSnapshotWriter writer = mockRaftClient.writers.get(
MetadataImageTest.IMAGE1.provenance().snapshotId());
assertNotNull(writer);
assertEquals(MetadataImageTest.IMAGE1.highestOffsetAndEpoch().offset(),
diff --git a/metadata/src/test/java/org/apache/kafka/image/publisher/metrics/SnapshotEmitterMetricsTest.java b/metadata/src/test/java/org/apache/kafka/image/publisher/metrics/SnapshotEmitterMetricsTest.java
new file mode 100644
index 00000000000..9f79631eef3
--- /dev/null
+++ b/metadata/src/test/java/org/apache/kafka/image/publisher/metrics/SnapshotEmitterMetricsTest.java
@@ -0,0 +1,111 @@
+/*
+ * 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.image.publisher.metrics;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.MetricsRegistry;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Optional;
+
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.controller.metrics.ControllerMetricsTestUtils;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+
+@Timeout(value = 40)
+public class SnapshotEmitterMetricsTest {
+ private static final Logger log = LoggerFactory.getLogger(SnapshotEmitterMetricsTest.class);
+
+ static class SnapshotEmitterMetricsTestContext implements AutoCloseable {
+ final MetricsRegistry registry;
+ final MockTime time;
+ final SnapshotEmitterMetrics metrics;
+
+ SnapshotEmitterMetricsTestContext() {
+ this.registry = new MetricsRegistry();
+ this.time = new MockTime(0, 10000L, 0L);
+ this.metrics = new SnapshotEmitterMetrics(Optional.of(registry), time);
+ }
+
+ @SuppressWarnings("unchecked") // suppress warning about Gauge typecast
+ long readLongGauge(String name) {
+ MetricName metricName = new MetricName(
+ "kafka.server",
+ "SnapshotEmitter",
+ name,
+ null,
+ "kafka.server:type=SnapshotEmitter,name=" + name
+ );
+ return ((Gauge<Long>) registry.allMetrics().get(metricName)).value();
+ }
+
+ @Override
+ public void close() {
+ try {
+ registry.shutdown();
+ } catch (Exception e) {
+ log.error("Error closing registry", e);
+ }
+ }
+ }
+
+ @Test
+ public void testMetricNames() {
+ try (SnapshotEmitterMetricsTestContext ctx = new SnapshotEmitterMetricsTestContext()) {
+ ControllerMetricsTestUtils.assertMetricsForTypeEqual(ctx.registry, "kafka.server:",
+ new HashSet<>(Arrays.asList(
+ "kafka.server:type=SnapshotEmitter,name=LatestSnapshotGeneratedBytes",
+ "kafka.server:type=SnapshotEmitter,name=LatestSnapshotGeneratedAgeMs"
+ )));
+ ctx.metrics.close();
+ ControllerMetricsTestUtils.assertMetricsForTypeEqual(ctx.registry, "KafkaController",
+ Collections.emptySet());
+ }
+ }
+
+ @Test
+ public void testLatestSnapshotGeneratedBytesMetric() {
+ try (SnapshotEmitterMetricsTestContext ctx = new SnapshotEmitterMetricsTestContext()) {
+ assertEquals(0L, ctx.metrics.latestSnapshotGeneratedBytes());
+ ctx.metrics.setLatestSnapshotGeneratedBytes(12345L);
+ assertEquals(12345L, ctx.metrics.latestSnapshotGeneratedBytes());
+ assertEquals(12345L, ctx.readLongGauge("LatestSnapshotGeneratedBytes"));
+ }
+ }
+
+ @Test
+ public void testLatestSnapshotGeneratedAgeMsMetric() {
+ try (SnapshotEmitterMetricsTestContext ctx = new SnapshotEmitterMetricsTestContext()) {
+ assertEquals(10000L, ctx.metrics.latestSnapshotGeneratedTimeMs());
+ assertEquals(0L, ctx.metrics.latestSnapshotGeneratedAgeMs());
+ ctx.time.sleep(20000L);
+ assertEquals(10000L, ctx.metrics.latestSnapshotGeneratedTimeMs());
+ assertEquals(20000L, ctx.metrics.latestSnapshotGeneratedAgeMs());
+ assertEquals(20000L, ctx.readLongGauge("LatestSnapshotGeneratedAgeMs"));
+ }
+ }
+}
diff --git a/metadata/src/test/java/org/apache/kafka/image/writer/RaftSnapshotWriterTest.java b/metadata/src/test/java/org/apache/kafka/image/writer/RaftSnapshotWriterTest.java
index 4cdacfe7e1e..0137aeb0772 100644
--- a/metadata/src/test/java/org/apache/kafka/image/writer/RaftSnapshotWriterTest.java
+++ b/metadata/src/test/java/org/apache/kafka/image/writer/RaftSnapshotWriterTest.java
@@ -17,15 +17,11 @@
package org.apache.kafka.image.writer;
-import org.apache.kafka.raft.OffsetAndEpoch;
-import org.apache.kafka.server.common.ApiMessageAndVersion;
-import org.apache.kafka.snapshot.SnapshotWriter;
+import org.apache.kafka.image.FakeSnapshotWriter;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
-import java.util.ArrayList;
import java.util.Arrays;
-import java.util.List;
import static java.util.Collections.emptyList;
import static org.apache.kafka.metadata.RecordTestUtils.testRecord;
@@ -36,70 +32,29 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
@Timeout(value = 40)
public class RaftSnapshotWriterTest {
- static class MockSnapshotWriter implements SnapshotWriter<ApiMessageAndVersion> {
- boolean frozen = false;
- boolean closed = false;
- List<List<ApiMessageAndVersion>> batches = new ArrayList<>();
-
- @Override
- public OffsetAndEpoch snapshotId() {
- return new OffsetAndEpoch(100L, 10);
- }
-
- @Override
- public long lastContainedLogOffset() {
- return snapshotId().offset();
- }
-
- @Override
- public int lastContainedLogEpoch() {
- return snapshotId().epoch();
- }
-
- @Override
- public boolean isFrozen() {
- return frozen;
- }
-
- @Override
- public void append(List<ApiMessageAndVersion> batch) {
- batches.add(batch);
- }
-
- @Override
- public void freeze() {
- frozen = true;
- }
-
- @Override
- public void close() {
- closed = true;
- }
- }
-
@Test
public void testFreezeAndClose() {
- MockSnapshotWriter snapshotWriter = new MockSnapshotWriter();
+ FakeSnapshotWriter snapshotWriter = new FakeSnapshotWriter();
RaftSnapshotWriter writer = new RaftSnapshotWriter(snapshotWriter, 2);
writer.write(testRecord(0));
writer.write(testRecord(1));
writer.write(testRecord(2));
writer.close(true);
- assertTrue(snapshotWriter.frozen);
- assertTrue(snapshotWriter.closed);
+ assertTrue(snapshotWriter.isFrozen());
+ assertTrue(snapshotWriter.isClosed());
assertEquals(Arrays.asList(
Arrays.asList(testRecord(0), testRecord(1)),
- Arrays.asList(testRecord(2))), snapshotWriter.batches);
+ Arrays.asList(testRecord(2))), snapshotWriter.batches());
}
@Test
public void testCloseWithoutFreeze() {
- MockSnapshotWriter snapshotWriter = new MockSnapshotWriter();
+ FakeSnapshotWriter snapshotWriter = new FakeSnapshotWriter();
RaftSnapshotWriter writer = new RaftSnapshotWriter(snapshotWriter, 2);
writer.write(testRecord(0));
writer.close();
- assertFalse(snapshotWriter.frozen);
- assertTrue(snapshotWriter.closed);
- assertEquals(emptyList(), snapshotWriter.batches);
+ assertFalse(snapshotWriter.isFrozen());
+ assertTrue(snapshotWriter.isClosed());
+ assertEquals(emptyList(), snapshotWriter.batches());
}
}
diff --git a/raft/src/main/java/org/apache/kafka/snapshot/RecordsSnapshotWriter.java b/raft/src/main/java/org/apache/kafka/snapshot/RecordsSnapshotWriter.java
index eeacf608a9f..ae6202426b8 100644
--- a/raft/src/main/java/org/apache/kafka/snapshot/RecordsSnapshotWriter.java
+++ b/raft/src/main/java/org/apache/kafka/snapshot/RecordsSnapshotWriter.java
@@ -192,11 +192,12 @@ final public class RecordsSnapshotWriter<T> implements SnapshotWriter<T> {
}
@Override
- public void freeze() {
+ public long freeze() {
finalizeSnapshotWithFooter();
appendBatches(accumulator.drain());
snapshot.freeze();
accumulator.close();
+ return snapshot.sizeInBytes();
}
@Override
diff --git a/raft/src/main/java/org/apache/kafka/snapshot/SnapshotWriter.java b/raft/src/main/java/org/apache/kafka/snapshot/SnapshotWriter.java
index 537335c058a..244cc5478f4 100644
--- a/raft/src/main/java/org/apache/kafka/snapshot/SnapshotWriter.java
+++ b/raft/src/main/java/org/apache/kafka/snapshot/SnapshotWriter.java
@@ -71,8 +71,10 @@ public interface SnapshotWriter<T> extends AutoCloseable {
* Freezes the snapshot by flushing all pending writes and marking it as immutable.
*
* Also adds a {@link SnapshotFooterRecord} to the end of the snapshot
+ *
+ * @return The size of the snapshot in bytes.
*/
- void freeze();
+ long freeze();
/**
* Closes the snapshot writer.