Add new metadata loader and publisher [WIP]

checkstyle/import-control.xml

@@ -269,6 +269,7 @@
     <allow pkg="org.apache.kafka.raft" />
     <allow pkg="org.apache.kafka.server.authorizer" />
     <allow pkg="org.apache.kafka.server.common" />
+    <allow pkg="org.apache.kafka.server.fault" />
     <allow pkg="org.apache.kafka.test" />
     <subpackage name="authorizer">
       <allow pkg="org.apache.kafka.common.acl" />
@@ -277,8 +278,11 @@
       <allow pkg="org.apache.kafka.controller" />
       <allow pkg="org.apache.kafka.metadata" />
     </subpackage>
-    <subpackage name="fault">
-      <allow pkg="org.apache.kafka.server.fault" />
+    <subpackage name="loader">
+      <allow pkg="org.apache.kafka.snapshot" />
+    </subpackage>
+    <subpackage name="publisher">
+      <allow pkg="org.apache.kafka.snapshot" />
     </subpackage>
   </subpackage>
 

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<Batch<ApiMessageAndVersion>> {
+    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<ApiMessageAndVersion> firstBatch;
+    private Batch<ApiMessageAndVersion> 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<ApiMessageAndVersion> 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;
+    }
+}

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 +
+                ")";
+    }
+}