diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml
index 4b07a26cba5c9..661db4272fbe1 100644
--- a/checkstyle/import-control.xml
+++ b/checkstyle/import-control.xml
@@ -269,6 +269,7 @@
+
@@ -277,8 +278,11 @@
-
-
+
+
+
+
+
diff --git a/metadata/src/main/java/org/apache/kafka/metadata/loader/BatchLoader.java b/metadata/src/main/java/org/apache/kafka/metadata/loader/BatchLoader.java
new file mode 100644
index 0000000000000..040f5ac6bd87e
--- /dev/null
+++ b/metadata/src/main/java/org/apache/kafka/metadata/loader/BatchLoader.java
@@ -0,0 +1,122 @@
+/*
+ * 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.metadata.loader;
+
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.image.MetadataDelta;
+import org.apache.kafka.raft.Batch;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.fault.FaultHandler;
+import org.slf4j.Logger;
+
+import java.util.function.Consumer;
+
+import static java.util.concurrent.TimeUnit.MICROSECONDS;
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+
+
+/**
+ * BatchLoader is an internal class used to apply a series of batches to a MetadataDelta object.
+ * Its main function is to track information about the loading process, and give good error
+ * diagnostics in the event of a failure.
+ */
+class BatchLoader implements Consumer> {
+ private final Logger log;
+ private final Time time;
+ private final FaultHandler faultHandler;
+ private final MetadataDelta delta;
+ private final String source;
+ private final boolean fromSnapshot;
+ private final long startTimeNs;
+ private long numBytes;
+ private Batch firstBatch;
+ private Batch lastBatch;
+ private int numBatches;
+ private int numRecords;
+
+ BatchLoader(
+ Logger log,
+ Time time,
+ FaultHandler faultHandler,
+ MetadataDelta delta,
+ String source,
+ boolean fromSnapshot
+ ) {
+ this.log = log;
+ this.time = time;
+ this.faultHandler = faultHandler;
+ this.delta = delta;
+ this.source = source;
+ this.fromSnapshot = fromSnapshot;
+ this.startTimeNs = time.nanoseconds();
+ this.numBytes = 0L;
+ this.firstBatch = null;
+ this.lastBatch = null;
+ this.numBatches = 0;
+ this.numRecords = 0;
+ }
+
+ @Override
+ public void accept(Batch batch) {
+ if (firstBatch == null) firstBatch = batch;
+ lastBatch = batch;
+ int recordIndex = 0;
+ long baseOffset = batch.baseOffset();
+ for (ApiMessageAndVersion record : batch.records()) {
+ if (log.isTraceEnabled()) {
+ log.trace(String.format("Loading %s batch with base offset %d, record [%d/%d]: %s",
+ source, baseOffset, recordIndex + 1, batch.records().size(), record.message()));
+ }
+ try {
+ // All snapshot records have the same offset, whereas in the log, each record has an
+ // offset which is one more than the previous one.
+ long effectiveOffset = fromSnapshot ? baseOffset : baseOffset + recordIndex;
+ delta.replay(effectiveOffset, batch.epoch(), record.message());
+ } catch (Throwable e) {
+ String message = String.format("Error loading %s batch starting at %d, " +
+ "batch %d, record [%d/%d], type %s", source, batch.baseOffset(),
+ numBatches + 1, recordIndex + 1, batch.records().size(),
+ record.message().getClass().getSimpleName());
+ faultHandler.handleFault(message, e); // Attempt to continue; do not throw.
+ }
+ recordIndex++;
+ numRecords++;
+ }
+ numBytes += batch.sizeInBytes();
+ numBatches++;
+ }
+
+ LoadInformation finish() {
+ if (firstBatch == null || lastBatch == null) {
+ throw faultHandler.handleFault("You must always load at least one batch.");
+ }
+ long endTimeNs = time.nanoseconds();
+ LoadInformation info = new LoadInformation(lastBatch.appendTimestamp(),
+ firstBatch.baseOffset(),
+ lastBatch.lastOffset(),
+ numBatches,
+ numRecords,
+ MICROSECONDS.convert(endTimeNs - startTimeNs, NANOSECONDS),
+ numBytes,
+ fromSnapshot);
+ if (log.isDebugEnabled()) {
+ log.debug("Loaded some new data from {}: {}", source, info);
+ }
+ return info;
+ }
+}
diff --git a/metadata/src/main/java/org/apache/kafka/metadata/loader/LoadInformation.java b/metadata/src/main/java/org/apache/kafka/metadata/loader/LoadInformation.java
new file mode 100644
index 0000000000000..d758bb2cdfc87
--- /dev/null
+++ b/metadata/src/main/java/org/apache/kafka/metadata/loader/LoadInformation.java
@@ -0,0 +1,159 @@
+/*
+ * 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.metadata.loader;
+
+import java.util.Objects;
+
+
+/**
+ * Contains information about a MetadataDelta that has been loaded by the MetadataLoader.
+ */
+public class LoadInformation {
+ /**
+ * The timestamp of the last batch that was part of the delta.
+ */
+ private final long lastContainedTimestamp;
+
+ /**
+ * The base offset that the delta was laoded from, or the snapshot offset if we loaded a
+ * snapshot.
+ */
+ private final long baseOffset;
+
+ /**
+ * The last offset contained in the delta, or the snapshot offset if we loaded a snapshot.
+ */
+ private final long lastContainedOffset;
+
+ /**
+ * The number of batches that were loaded in the delta.
+ */
+ private final int numBatches;
+
+ /**
+ * The number of records that were loaded in the delta.
+ */
+ private final int numRecords;
+
+ /**
+ * The time in microseconds that it took to load the delta.
+ */
+ private final long elapsedUs;
+
+ /**
+ * The total size of the records in bytes that we read while creating the delta.
+ */
+ private final long numBytes;
+
+ /**
+ * True only if we loaded the delta from a snapshot.
+ */
+ private final boolean fromSnapshot;
+
+ public LoadInformation(
+ long lastContainedTimestamp,
+ long baseOffset,
+ long lastContainedOffset,
+ int numBatches,
+ int numRecords,
+ long elapsedUs,
+ long numBytes,
+ boolean fromSnapshot
+ ) {
+ this.lastContainedTimestamp = lastContainedTimestamp;
+ this.baseOffset = baseOffset;
+ this.lastContainedOffset = lastContainedOffset;
+ this.numBatches = numBatches;
+ this.numRecords = numRecords;
+ this.elapsedUs = elapsedUs;
+ this.numBytes = numBytes;
+ this.fromSnapshot = fromSnapshot;
+ }
+
+ public long lastContainedTimestamp() {
+ return lastContainedTimestamp;
+ }
+
+ public long baseOffset() {
+ return baseOffset;
+ }
+
+ public long lastContainedOffset() {
+ return lastContainedOffset;
+ }
+
+ public int numBatches() {
+ return numBatches;
+ }
+
+ public int numRecords() {
+ return numRecords;
+ }
+
+ public long elapsedUs() {
+ return elapsedUs;
+ }
+
+ public long numBytes() {
+ return numBytes;
+ }
+
+ public boolean fromSnapshot() {
+ return fromSnapshot;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(lastContainedTimestamp,
+ baseOffset,
+ lastContainedOffset,
+ numBatches,
+ numRecords,
+ elapsedUs,
+ numBytes,
+ fromSnapshot);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (o == null || !o.getClass().equals(this.getClass())) return false;
+ LoadInformation other = (LoadInformation) o;
+ return lastContainedTimestamp == other.lastContainedTimestamp &&
+ baseOffset == other.baseOffset &&
+ lastContainedOffset == other.lastContainedOffset &&
+ numBatches == other.numBatches &&
+ numRecords == other.numRecords &&
+ elapsedUs == other.elapsedUs &&
+ numBytes == other.numBytes &&
+ fromSnapshot == other.fromSnapshot;
+ }
+
+ @Override
+ public String toString() {
+ return "LoadInformation(" +
+ "lastContainedTimestamp=" + lastContainedTimestamp +
+ ", baseOffset=" + baseOffset +
+ ", lastContainedOffset=" + lastContainedOffset +
+ ", numBatches=" + numBatches +
+ ", numRecords=" + numRecords +
+ ", elapsedUs=" + elapsedUs +
+ ", numBytes=" + numBytes +
+ ", fromSnapshot=" + fromSnapshot +
+ ")";
+ }
+}
diff --git a/metadata/src/main/java/org/apache/kafka/metadata/loader/MetadataLoader.java b/metadata/src/main/java/org/apache/kafka/metadata/loader/MetadataLoader.java
new file mode 100644
index 0000000000000..8e5900efd16ce
--- /dev/null
+++ b/metadata/src/main/java/org/apache/kafka/metadata/loader/MetadataLoader.java
@@ -0,0 +1,310 @@
+/*
+ * 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.metadata.loader;
+
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.image.MetadataDelta;
+import org.apache.kafka.image.MetadataImage;
+import org.apache.kafka.metadata.publisher.MetadataPublisher;
+import org.apache.kafka.queue.KafkaEventQueue;
+import org.apache.kafka.raft.Batch;
+import org.apache.kafka.raft.BatchReader;
+import org.apache.kafka.raft.LeaderAndEpoch;
+import org.apache.kafka.raft.RaftClient;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.fault.FaultHandler;
+import org.apache.kafka.server.fault.FaultHandlerException;
+import org.apache.kafka.snapshot.SnapshotReader;
+import org.slf4j.Logger;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+
+/**
+ * The MetadataLoader follows changes provided by a RaftClient, and packages them into metadata
+ * deltas and images that can be consumed by publishers.
+ *
+ * The Loader maintains its own thread, which is used to make all callbacks into publishers. If a
+ * publisher A is installed before B, A will receive all callbacks before B. This is also true if
+ * A and B are installed as part of a list [A, B].
+ *
+ * Publishers should not modify any data structures passed to them.
+ *
+ * It is possible to change the list of publishers dynamically over time. Whenever a new publisher is
+ * added, it receives a catch-up delta which contains the full state. Any publisher installed when the
+ * loader is closed will itself be closed.
+ */
+public class MetadataLoader implements RaftClient.Listener, AutoCloseable {
+ public static class Builder {
+ private int nodeId = -1;
+ private Time time = Time.SYSTEM;
+ private LogContext logContext = null;
+ private String threadNamePrefix = "";
+ private FaultHandler faultHandler = null;
+
+ public Builder setNodeId(int nodeId) {
+ this.nodeId = nodeId;
+ return this;
+ }
+
+ public Builder setTime(Time time) {
+ this.time = time;
+ return this;
+ }
+
+ public Builder setThreadNamePrefix(String threadNamePrefix) {
+ this.threadNamePrefix = threadNamePrefix;
+ return this;
+ }
+
+ public Builder setFaultHandler(FaultHandler faultHandler) {
+ this.faultHandler = faultHandler;
+ return this;
+ }
+
+ public MetadataLoader build() {
+ if (faultHandler == null) throw new RuntimeException("You must set a fault handler.");
+ if (logContext == null) {
+ logContext = new LogContext("[MetadataLoader id=" + nodeId + "] ");
+ }
+ return new MetadataLoader(nodeId,
+ time,
+ logContext,
+ threadNamePrefix,
+ faultHandler);
+ }
+ }
+
+ /**
+ * The log4j logger for this loader.
+ */
+ private final Logger log;
+
+ /**
+ * The node id.
+ */
+ private final int nodeId;
+
+ /**
+ * The clock used by this loader.
+ */
+ private final Time time;
+
+ /**
+ * The fault handler to use if metadata loading fails.
+ */
+ private final FaultHandler faultHandler;
+
+ /**
+ * The publishers who should receive cluster metadata updates.
+ */
+ private final List publishers;
+
+ /**
+ * The current metadata image. Accessed only from the event queue thread.
+ */
+ private MetadataImage image;
+
+ /**
+ * The event queue which runs this loader.
+ */
+ private final KafkaEventQueue eventQueue;
+
+ private MetadataLoader(
+ int nodeId,
+ Time time,
+ LogContext logContext,
+ String threadNamePrefix,
+ FaultHandler faultHandler
+ ) {
+ this.log = logContext.logger(MetadataLoader.class);
+ this.nodeId = nodeId;
+ this.time = time;
+ this.faultHandler = faultHandler;
+ this.publishers = new ArrayList<>();
+ this.image = MetadataImage.EMPTY;
+ this.eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix);
+ }
+
+ @Override
+ public void handleCommit(BatchReader reader) {
+ eventQueue.append(() -> {
+ long lastImageOffset = image.highestOffsetAndEpoch().offset;
+ try {
+ MetadataDelta delta = new MetadataDelta(image);
+ BatchLoader loader = new BatchLoader(log, time, faultHandler, delta, "log", false);
+ reader.forEachRemaining(loader);
+ LoadInformation info = loader.finish();
+ apply("log commits", delta, info);
+ publish("log commits", publishers, delta, info);
+ } catch (FaultHandlerException e) {
+ // Ignore these because they have already been handled.
+ } catch (Throwable e) {
+ // This is a general catch-all block where we don't expect to end up;
+ // failure-prone operations should have individual try/catch blocks around them.
+ faultHandler.handleFault("Unhandled fault in MetadataLoader#handleCommit " +
+ "after last image offset " + lastImageOffset, e);
+ }
+ });
+ }
+
+ void apply(String what, MetadataDelta delta, LoadInformation info) {
+ try {
+ MetadataImage newImage = delta.apply();
+ image = newImage;
+ } catch (Throwable e) {
+ throw faultHandler.handleFault("Error generating new metadata image from the " + what +
+ " ending at offset " + info.lastContainedOffset(), e);
+ }
+ }
+
+ void publish(
+ String what,
+ List targetPublishers,
+ MetadataDelta delta,
+ LoadInformation info
+ ) {
+ for (MetadataPublisher publisher : targetPublishers) {
+ try {
+ publisher.publish(delta, image, info);
+ } catch (Throwable e) {
+ faultHandler.handleFault("Error publishing the " + what + " ending at " +
+ info.lastContainedOffset() + " with publisher " + publisher.name(), e);
+ }
+ }
+ }
+
+ @Override
+ public void handleSnapshot(SnapshotReader reader) {
+ eventQueue.append(() -> {
+ String snapName = String.format("snapshot %020d-%010d",
+ reader.lastContainedLogOffset(), reader.lastContainedLogEpoch());
+ try {
+ MetadataDelta delta = new MetadataDelta(image);
+ BatchLoader loader = new BatchLoader(log, time, faultHandler, delta, snapName, true);
+ reader.forEachRemaining(loader);
+ LoadInformation info = loader.finish();
+ apply("snapshot", delta, info);
+ publish("snapshot", publishers, delta, info);
+ } catch (FaultHandlerException e) {
+ // Ignore these because they have already been handled.
+ } catch (Throwable e) {
+ // This is a general catch-all block where we don't expect to end up;
+ // failure-prone operations should have individual try/catch blocks around them.
+ faultHandler.handleFault("Unhandled fault in MetadataLoader#handleSnapshot " +
+ "for snapshot " + snapName, e);
+ }
+ });
+ }
+
+ @Override
+ public void handleLeaderChange(LeaderAndEpoch leader) {
+ // Nothing to do.
+ }
+
+ /**
+ * Install a list of publishers. When a publisher is installed, we will publish a MetadataDelta
+ * to it which contains the entire current image.
+ *
+ * @param newPublishers The publishers to install.
+ *
+ * @return A future which yields null when the publishers have been added, or
+ * an exception if the installation failed.
+ */
+ public CompletableFuture installPublishers(List newPublishers) {
+ if (!newPublishers.isEmpty()) return CompletableFuture.completedFuture(null);
+ CompletableFuture future = new CompletableFuture<>();
+ eventQueue.beginShutdown("installPublishers", () -> {
+ try {
+ // Publishers can't be re-installed if they're already present.
+ for (MetadataPublisher publisher : newPublishers) {
+ if (publishers.contains(publisher)) {
+ throw faultHandler.handleFault("Attempted to install publisher " + publisher.name() +
+ ", which is already installed.");
+ }
+ }
+ String source = "installPublishers image at offset " + image.highestOffsetAndEpoch().offset;
+ MetadataDelta delta = new MetadataDelta(image);
+ BatchLoader loader = new BatchLoader(log, time, faultHandler, delta, source, true);
+ long now = time.milliseconds();
+ // Use the records in the current metadata image to build up a delta that contains
+ // everything. The new publishers will need it because everything is new to them.
+ image.write(recordList -> {
+ Batch batch = Batch.data(image.highestOffsetAndEpoch().offset,
+ image.highestOffsetAndEpoch().epoch,
+ now,
+ 0, // We don't populate the size field here.
+ recordList);
+ loader.accept(batch);
+ });
+ LoadInformation info = loader.finish();
+ // Publish just to the new publishers.
+ publish(source, newPublishers, delta, info);
+ publishers.addAll(newPublishers);
+ future.complete(null);
+ } catch (Throwable e) {
+ future.completeExceptionally(faultHandler.handleFault("Unhandled fault in " +
+ "MetadataLoader#installPublishers", e));
+ }
+ });
+ return future;
+ }
+
+ /**
+ * Remove a publisher.
+ *
+ * @param publisher The publisher to remove.
+ *
+ * @return A future which yields null when the publisher has been removed,
+ * or an exception if the removal failed.
+ */
+ public CompletableFuture removePublisher(MetadataPublisher publisher) {
+ CompletableFuture future = new CompletableFuture<>();
+ eventQueue.beginShutdown("removePublisher", () -> {
+ if (!publishers.remove(publisher)) {
+ throw faultHandler.handleFault("Attempted to remove publisher " + publisher.name() +
+ ", which is not installed.");
+ }
+ future.complete(null);
+ });
+ return future;
+ }
+
+ @Override
+ public void beginShutdown() {
+ eventQueue.beginShutdown("beginShutdown", () -> {
+ for (MetadataPublisher publisher : publishers) {
+ try {
+ publisher.close();
+ } catch (Throwable e) {
+ faultHandler.handleFault("Got unexpected exception while closing " +
+ "publisher " + publisher.name(), e);
+ }
+ }
+ });
+ }
+
+ @Override
+ public void close() throws Exception {
+ beginShutdown();
+ eventQueue.close();
+ }
+}
diff --git a/metadata/src/main/java/org/apache/kafka/metadata/publisher/BatchSplitter.java b/metadata/src/main/java/org/apache/kafka/metadata/publisher/BatchSplitter.java
new file mode 100644
index 0000000000000..dc1a25d503ccd
--- /dev/null
+++ b/metadata/src/main/java/org/apache/kafka/metadata/publisher/BatchSplitter.java
@@ -0,0 +1,50 @@
+/*
+ * 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.metadata.publisher;
+
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+
+import java.util.List;
+import java.util.function.Consumer;
+
+
+/**
+ * Takes as input a potentially long list of records, and feeds the associated consumer a series
+ * of smaller lists of records.
+ */
+public class BatchSplitter implements Consumer> {
+ private final int maxRecordsInBatch;
+ private final Consumer> consumer;
+
+ public BatchSplitter(
+ int maxRecordsInBatch,
+ Consumer> consumer
+ ) {
+ this.maxRecordsInBatch = maxRecordsInBatch;
+ this.consumer = consumer;
+ }
+
+ @Override
+ public void accept(List messages) {
+ int i = 0;
+ while (i < messages.size()) {
+ consumer.accept(messages.subList(i, Math.min(i + maxRecordsInBatch, messages.size())));
+ i += maxRecordsInBatch;
+ }
+ }
+}
diff --git a/metadata/src/main/java/org/apache/kafka/metadata/publisher/MetadataPublisher.java b/metadata/src/main/java/org/apache/kafka/metadata/publisher/MetadataPublisher.java
new file mode 100644
index 0000000000000..1fd6568fd5fa7
--- /dev/null
+++ b/metadata/src/main/java/org/apache/kafka/metadata/publisher/MetadataPublisher.java
@@ -0,0 +1,53 @@
+/*
+ * 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.metadata.publisher;
+
+import org.apache.kafka.image.MetadataDelta;
+import org.apache.kafka.image.MetadataImage;
+import org.apache.kafka.metadata.loader.LoadInformation;
+
+
+/**
+ * Publishes metadata deltas which we have loaded from the log and snapshots.
+ *
+ * Publishers receive a stream of callbacks from the metadata loader which keeps them notified
+ * of the latest cluster metadata. This interface abstracts away some of the complications of
+ * following the cluster metadata. For example, if the loader needs to read a snapshot, it will
+ * present the contents of the snapshot in the form of a delta from the previous state.
+ */
+public interface MetadataPublisher extends AutoCloseable {
+ /**
+ * Returns the name of this publisher.
+ *
+ * @return The publisher name.
+ */
+ String name();
+
+ /**
+ * Publish a change to the cluster metadata.
+ *
+ * @param delta The delta between the previous state and the new one.
+ * @param newImage The complete new state.
+ * @param info Some information associated with this publish event.
+ */
+ void publish(
+ MetadataDelta delta,
+ MetadataImage newImage,
+ LoadInformation info
+ );
+}
diff --git a/metadata/src/main/java/org/apache/kafka/metadata/publisher/MetadataSnapshotter.java b/metadata/src/main/java/org/apache/kafka/metadata/publisher/MetadataSnapshotter.java
new file mode 100644
index 0000000000000..3f6ecfc5fcbe6
--- /dev/null
+++ b/metadata/src/main/java/org/apache/kafka/metadata/publisher/MetadataSnapshotter.java
@@ -0,0 +1,205 @@
+/*
+ * 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.metadata.publisher;
+
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.metadata.loader.MetadataLoader;
+import org.apache.kafka.queue.EventQueue;
+import org.apache.kafka.queue.KafkaEventQueue;
+import org.apache.kafka.raft.RaftClient;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.snapshot.SnapshotWriter;
+import org.slf4j.Logger;
+
+import java.util.Optional;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import static java.util.concurrent.TimeUnit.MICROSECONDS;
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+
+
+public class MetadataSnapshotter implements Consumer, AutoCloseable {
+ /**
+ * The maximum number of records we will put in each batch.
+ *
+ * From the perspective of the Raft layer, the limit on batch size is specified in terms of
+ * bytes, not number of records. @See {@link KafkaRaftClient#MAX_BATCH_SIZE_BYTES} for details.
+ * However, it's more convenient to limit the batch size here in terms of number of records.
+ * So we chose a low number that will not cause problems.
+ */
+ private static final int MAX_RECORDS_PER_BATCH = 1024;
+
+ public static class Builder {
+ private Time time = Time.SYSTEM;
+ private LogContext logContext = null;
+ private String threadNamePrefix = "";
+ private Function> writerCreator = null;
+ private Supplier suppressionReasonSupplier = () -> "";
+
+ public Builder setTime(Time time) {
+ this.time = time;
+ return this;
+ }
+
+ public Builder setLogContext(LogContext logContext) {
+ this.logContext = logContext;
+ return this;
+ }
+
+ public Builder setThreadNamePrefix(String threadNamePrefix) {
+ this.threadNamePrefix = threadNamePrefix;
+ return this;
+ }
+
+ public Builder setWriterCreator(Function> writerCreator) {
+ this.writerCreator = writerCreator;
+ return this;
+ }
+
+ public Builder setWriterCreator(RaftClient raftClient) {
+ this.writerCreator = request -> {
+ Optional> writer = raftClient.createSnapshot(
+ request.offset(), request.epoch(), request.lastContainedLogTimeMs());
+ if (!writer.isPresent()) {
+ throw new RuntimeException("the committed offset is greater than the " +
+ "high water mark or less than the log start offset.");
+ }
+ return writer.get();
+ };
+ return this;
+ }
+
+ public Builder setSuppressionReasonSupplier(Supplier suppressionReasonSupplier) {
+ this.suppressionReasonSupplier = suppressionReasonSupplier;
+ return this;
+ }
+
+ public MetadataSnapshotter build() {
+ if (logContext == null) {
+ logContext = new LogContext("[Metadatanapshotter ]");
+ }
+ if (writerCreator == null) {
+ throw new RuntimeException("You must set a writerCreator.");
+ }
+ return new MetadataSnapshotter(time,
+ logContext,
+ threadNamePrefix,
+ writerCreator,
+ suppressionReasonSupplier);
+ }
+ }
+
+ /**
+ * The log4j logger for this snapshotter.
+ */
+ private final Logger log;
+
+ /**
+ * The clock supplier for this snapshotter.
+ */
+ private final Time time;
+
+ /**
+ * Creates SnapshotWriter objects.
+ */
+ private final Function> writerCreator;
+
+ /**
+ * Returns an empty string if a snapshot can be created; otherwise, returns the reason why it
+ * cannot be created.
+ */
+ private final Supplier suppressionReasonSupplier;
+
+ /**
+ * The event queue which runs this snapshotter.
+ */
+ private final EventQueue eventQueue;
+
+ /**
+ * The offset of the snapshot we are currently writing, or -1 if we are not writing one.
+ * Modified only by the event queue thread.
+ */
+ private volatile long currentSnapshotOffset;
+
+ private MetadataSnapshotter(
+ Time time,
+ LogContext logContext,
+ String threadNamePrefix,
+ Function> writerCreator,
+ Supplier suppressionReasonSupplier
+ ) {
+ this.log = logContext.logger(MetadataLoader.class);
+ this.time = time;
+ this.writerCreator = writerCreator;
+ this.suppressionReasonSupplier = suppressionReasonSupplier;
+ this.eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix);
+ this.currentSnapshotOffset = -1L;
+ }
+
+ @Override
+ public void accept(SnapshotCreationRequest request) {
+ long offset = currentSnapshotOffset;
+ if (offset < 0L) {
+ if (!request.force()) {
+ log.warn("Ignoring request to create new snapshot at offset " +
+ request.offset() + " because we are already writing a " +
+ "snapshot at offset " + offset + ".");
+ }
+ }
+ eventQueue.append(() -> {
+ String suppressionReason = suppressionReasonSupplier.get();
+ if (!suppressionReason.isEmpty()) {
+ log.error("Unable to create snapshot at offset {}: snapshots have " +
+ "been supressed because {}.", request.offset(), suppressionReason);
+ return;
+ }
+ String forcedString = request.force() ? "forced " : "";
+ log.debug("Starting {}snapshot at offset {}.", forcedString, request.offset());
+ currentSnapshotOffset = request.offset();
+ long startNs = time.nanoseconds();
+ try (SnapshotWriter writer = writerCreator.apply(request)) {
+ BatchSplitter batchSplitter = new BatchSplitter(MAX_RECORDS_PER_BATCH,
+ recordList -> writer.append(recordList));
+ request.image().write(batchSplitter);
+ writer.freeze();
+ } catch (Throwable e) {
+ log.error("Failed to create snapshot at offset {}", request.offset(), e);
+ return;
+ } finally {
+ currentSnapshotOffset = -1L;
+ }
+ long endNs = time.nanoseconds();
+ long elapsedUs = MICROSECONDS.convert(endNs - startNs, NANOSECONDS);
+ log.info("Wrote {}snapshot at offset {} in {} microseconds.",
+ forcedString, request.offset(), elapsedUs);
+ });
+ }
+
+ public void beginShutdown() {
+ eventQueue.beginShutdown("MetadataSnapshotter#beginShutdown");
+ }
+
+ @Override
+ public void close() throws InterruptedException {
+ beginShutdown();
+ eventQueue.close();
+ }
+}
diff --git a/metadata/src/main/java/org/apache/kafka/metadata/publisher/SnapshotCreationRequest.java b/metadata/src/main/java/org/apache/kafka/metadata/publisher/SnapshotCreationRequest.java
new file mode 100644
index 0000000000000..ab0359b3631e0
--- /dev/null
+++ b/metadata/src/main/java/org/apache/kafka/metadata/publisher/SnapshotCreationRequest.java
@@ -0,0 +1,84 @@
+/*
+ * 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.metadata.publisher;
+
+import org.apache.kafka.image.MetadataImage;
+
+import java.util.Objects;
+
+
+public class SnapshotCreationRequest {
+ private final long lastContainedLogTimeMs;
+ private final MetadataImage image;
+ private final boolean force;
+
+ public SnapshotCreationRequest(
+ long lastContainedLogTimeMs,
+ MetadataImage image,
+ boolean force
+ ) {
+ this.lastContainedLogTimeMs = lastContainedLogTimeMs;
+ this.image = image;
+ this.force = force;
+ }
+
+ public long lastContainedLogTimeMs() {
+ return lastContainedLogTimeMs;
+ }
+
+ public MetadataImage image() {
+ return image;
+ }
+
+ public boolean force() {
+ return force;
+ }
+
+ public long offset() {
+ return image.highestOffsetAndEpoch().offset;
+ }
+
+ public int epoch() {
+ return image.highestOffsetAndEpoch().epoch;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(lastContainedLogTimeMs,
+ image,
+ force);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (o == null || !(o.getClass().equals(this.getClass()))) return false;
+ SnapshotCreationRequest other = (SnapshotCreationRequest) o;
+ return lastContainedLogTimeMs == other.lastContainedLogTimeMs &&
+ image.equals(other.image) &&
+ force == other.force;
+ }
+
+ @Override
+ public String toString() {
+ return "SnapshotCreationRequest(" +
+ "lastContainedLogTimeMs=" + lastContainedLogTimeMs +
+ ", image=" + image +
+ ", force=" + force +
+ ")";
+ }
+}
diff --git a/metadata/src/main/java/org/apache/kafka/metadata/publisher/SnapshotPublisher.java b/metadata/src/main/java/org/apache/kafka/metadata/publisher/SnapshotPublisher.java
new file mode 100644
index 0000000000000..698cf8211df6f
--- /dev/null
+++ b/metadata/src/main/java/org/apache/kafka/metadata/publisher/SnapshotPublisher.java
@@ -0,0 +1,144 @@
+/*
+ * 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.metadata.publisher;
+
+import org.apache.kafka.image.MetadataDelta;
+import org.apache.kafka.image.MetadataImage;
+import org.apache.kafka.metadata.loader.LoadInformation;
+
+import java.util.function.Consumer;
+
+
+/**
+ * Publishes metadata deltas which we have loaded from the log and snapshots.
+ *
+ * Publishers receive a stream of callbacks from the metadata loader which keeps them notified
+ * of the latest cluster metadata. This interface abstracts away some of the complications of
+ * following the cluster metadata. For example, if the loader needs to read a snapshot, it will
+ * present the contents of the snapshot in the form of a delta from the previous state.
+ */
+public class SnapshotPublisher implements MetadataPublisher {
+ public static class Builder {
+ private Consumer requestHandler = null;
+ private long maxBytesBetweenSnapshots = Long.MAX_VALUE;
+
+ public Builder setRequestHandler(Consumer requestHandler) {
+ this.requestHandler = requestHandler;
+ return this;
+ }
+
+ public Builder setMaxBytesBetweenSnapshots(long maxBytesBetweenSnapshots) {
+ this.maxBytesBetweenSnapshots = maxBytesBetweenSnapshots;
+ return this;
+ }
+
+ public SnapshotPublisher build() {
+ if (requestHandler == null) throw new RuntimeException("You must set the request handler.");
+ return new SnapshotPublisher(requestHandler,
+ maxBytesBetweenSnapshots);
+ }
+ }
+
+ @Override
+ public String name() {
+ return "SnapshotPublisher";
+ }
+
+ /**
+ * The object which creates handles snapshot creation requests.
+ */
+ private final Consumer requestHandler;
+
+ /**
+ * The maximum number of bytes of log messages we should allow before triggering a snapshot.
+ */
+ private final long maxBytesBetweenSnapshots;
+
+ /**
+ * The value that MetadataImage had when publish was last called.
+ */
+ private MetadataImage prevImage;
+
+ /**
+ * The value that lastContainedTimestamp had when publish was last called.
+ */
+ private long prevLastContainedTimestamp;
+
+ /**
+ * The number of bytes of log messages that we have seen since the last snapshot load or creation.
+ */
+ private long bytesSinceLastSnapshot;
+
+ SnapshotPublisher(
+ Consumer requestHandler,
+ long maxBytesBetweenSnapshots
+ ) {
+ this.requestHandler = requestHandler;
+ this.maxBytesBetweenSnapshots = maxBytesBetweenSnapshots;
+ this.prevImage = MetadataImage.EMPTY;
+ this.prevLastContainedTimestamp = 0L;
+ clearStats();
+ }
+
+ @Override
+ public void publish(
+ MetadataDelta delta,
+ MetadataImage newImage,
+ LoadInformation info
+ ) {
+ if (info.fromSnapshot()) {
+ // If we loaded a snapshot, we never want to create a snapshot in response, since
+ // that would merely duplicate what we just loaded. We also want to reset the clock
+ // on when we should create the next snapshot.
+ clearStats();
+ } else if (delta.featuresDelta() != null && delta.featuresDelta().metadataVersionChange().isPresent()) {
+ // If we are reading the log and the MetadataVersion changed, we must snapshot immediately.
+ // Except in the trivial case where the previous image was literally empty.
+ if (!prevImage.isEmpty()) {
+ requestHandler.accept(new SnapshotCreationRequest(prevLastContainedTimestamp, prevImage, true));
+ }
+ clearStats();
+ } else {
+ // Check if the statistics indicate that we should snapshot.
+ updateStats(info);
+ if (shouldSnapshotBasedOnStats()) {
+ requestHandler.accept(new SnapshotCreationRequest(info.lastContainedTimestamp(), newImage, false));
+ clearStats();
+ }
+ }
+ this.prevLastContainedTimestamp = info.lastContainedTimestamp();
+ }
+
+ void updateStats(LoadInformation info) {
+ bytesSinceLastSnapshot += info.numBytes();
+ }
+
+ boolean shouldSnapshotBasedOnStats() {
+ if (bytesSinceLastSnapshot >= maxBytesBetweenSnapshots) return true;
+ return false;
+ }
+
+ void clearStats() {
+ bytesSinceLastSnapshot = 0L;
+ }
+
+ @Override
+ public void close() throws Exception {
+ // nothing to do
+ }
+}