KAFKA-12467: Implement QuorumController snapshot generation (apache#10366)

Implement controller-side snapshot generation.Implement QuorumController snapshot
generation.  Note that this PR does not handle KRaft integration, just the internal 
snapshot record generation and consumption logic.

Reading a snapshot is relatively straightforward.  When the  QuorumController
starts up, it loads the most recent snapshot.  This is just a series of records
that we replay, plus a log offset ("snapshot epoch") that we advance to.

Writing a snapshot is more complex.  There are several components:
the SnapshotWriter which persists the snapshot, the SnapshotGenerator
which manages writing each batch of records, and the SnapshotGeneratorManager
which interfaces the preceding two classes with the event queue.

Controller snapshots are done incrementally.  In order to avoid blocking the
controller thread for a long time, we pull a few record batches at a time from
our record batch iterators.  These iterators are implemented by controller
manager classes such as ReplicationControlManager, ClusterControlManager, etc.

Finally, this PR adds ControllerTestUtils#deepSortRecords and
ControllerTestUtils#assertBatchIteratorContains, which make it easier to write
unit tests.  Since records are often constructed from unsorted data structures,
it is often useful to sort them before comparing them.

Reviewers: David Arthur <[email protected]>

Author: cmccabe

checkstyle/suppressions.xml

@@ -268,6 +268,8 @@
               files="(ReplicationControlManager|ReplicationControlManagerTest).java"/>
     <suppress checks="ClassFanOutComplexity"
               files="(QuorumController|ReplicationControlManager).java"/>
+    <suppress checks="ParameterNumber"
+              files="(QuorumController).java"/>
     <suppress checks="CyclomaticComplexity"
               files="(ReplicationControlManager).java"/>
     <suppress checks="NPathComplexity"

core/src/test/java/kafka/test/MockController.java

@@ -201,6 +201,11 @@ public CompletableFuture<Void> waitForReadyBrokers(int minBrokers) {
         throw new UnsupportedOperationException();
     }
 
+    @Override
+    public CompletableFuture<Long> beginWritingSnapshot() {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     public void beginShutdown() {
         this.active = false;

metadata/src/main/java/org/apache/kafka/controller/ClientQuotaControlManager.java

@@ -21,6 +21,7 @@
 import org.apache.kafka.common.config.internals.QuotaConfigs;
 import org.apache.kafka.common.errors.InvalidRequestException;
 import org.apache.kafka.common.metadata.QuotaRecord;
+import org.apache.kafka.common.metadata.QuotaRecord.EntityData;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.quota.ClientQuotaAlteration;
 import org.apache.kafka.common.quota.ClientQuotaEntity;
@@ -35,18 +36,20 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
+import java.util.NoSuchElementException;
 import java.util.Objects;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
 
 public class ClientQuotaControlManager {
-
     private final SnapshotRegistry snapshotRegistry;
 
-    final TimelineHashMap<ClientQuotaEntity, Map<String, Double>> clientQuotaData;
+    final TimelineHashMap<ClientQuotaEntity, TimelineHashMap<String, Double>> clientQuotaData;
 
     ClientQuotaControlManager(SnapshotRegistry snapshotRegistry) {
         this.snapshotRegistry = snapshotRegistry;
@@ -98,7 +101,7 @@ public void replay(QuotaRecord record) {
         Map<String, String> entityMap = new HashMap<>(2);
         record.entity().forEach(entityData -> entityMap.put(entityData.entityType(), entityData.entityName()));
         ClientQuotaEntity entity = new ClientQuotaEntity(entityMap);
-        Map<String, Double> quotas = clientQuotaData.get(entity);
+        TimelineHashMap<String, Double> quotas = clientQuotaData.get(entity);
         if (quotas == null) {
             quotas = new TimelineHashMap<>(snapshotRegistry, 0);
             clientQuotaData.put(entity, quotas);
@@ -136,14 +139,15 @@ private void alterClientQuotaEntity(
         }
 
         // Don't share objects between different records
-        Supplier<List<QuotaRecord.EntityData>> recordEntitySupplier = () ->
-                validatedEntityMap.entrySet().stream().map(mapEntry -> new QuotaRecord.EntityData()
+        Supplier<List<EntityData>> recordEntitySupplier = () ->
+                validatedEntityMap.entrySet().stream().map(mapEntry -> new EntityData()
                         .setEntityType(mapEntry.getKey())
                         .setEntityName(mapEntry.getValue()))
                         .collect(Collectors.toList());
 
         List<ApiMessageAndVersion> newRecords = new ArrayList<>(newQuotaConfigs.size());
-        Map<String, Double> currentQuotas = clientQuotaData.getOrDefault(entity, Collections.emptyMap());
+        Map<String, Double> currentQuotas = clientQuotaData.containsKey(entity) ?
+            clientQuotaData.get(entity) : Collections.emptyMap();
         newQuotaConfigs.forEach((key, newValue) -> {
             if (newValue == null) {
                 if (currentQuotas.containsKey(key)) {
@@ -249,7 +253,7 @@ private ApiError validateEntity(ClientQuotaEntity entity, Map<String, String> va
             return new ApiError(Errors.INVALID_REQUEST, "Invalid empty client quota entity");
         }
 
-        for (Map.Entry<String, String> entityEntry : entity.entries().entrySet()) {
+        for (Entry<String, String> entityEntry : entity.entries().entrySet()) {
             String entityType = entityEntry.getKey();
             String entityName = entityEntry.getValue();
             if (validatedEntityMap.containsKey(entityType)) {
@@ -272,4 +276,44 @@ private ApiError validateEntity(ClientQuotaEntity entity, Map<String, String> va
 
         return ApiError.NONE;
     }
+
+    class ClientQuotaControlIterator implements Iterator<List<ApiMessageAndVersion>> {
+        private final long epoch;
+        private final Iterator<Entry<ClientQuotaEntity, TimelineHashMap<String, Double>>> iterator;
+
+        ClientQuotaControlIterator(long epoch) {
+            this.epoch = epoch;
+            this.iterator = clientQuotaData.entrySet(epoch).iterator();
+        }
+
+        @Override
+        public boolean hasNext() {
+            return iterator.hasNext();
+        }
+
+        @Override
+        public List<ApiMessageAndVersion> next() {
+            if (!hasNext()) throw new NoSuchElementException();
+            Entry<ClientQuotaEntity, TimelineHashMap<String, Double>> entry = iterator.next();
+            ClientQuotaEntity entity = entry.getKey();
+            List<ApiMessageAndVersion> records = new ArrayList<>();
+            for (Entry<String, Double> quotaEntry : entry.getValue().entrySet(epoch)) {
+                QuotaRecord record = new QuotaRecord();
+                for (Entry<String, String> entityEntry : entity.entries().entrySet()) {
+                    record.entity().add(new EntityData().
+                        setEntityType(entityEntry.getKey()).
+                        setEntityName(entityEntry.getValue()));
+                }
+                record.setKey(quotaEntry.getKey());
+                record.setValue(quotaEntry.getValue());
+                record.setRemove(false);
+                records.add(new ApiMessageAndVersion(record, (short) 0));
+            }
+            return records;
+        }
+    }
+
+    ClientQuotaControlIterator iterator(long epoch) {
+        return new ClientQuotaControlIterator(epoch);
+    }
 }

metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java

@@ -24,6 +24,10 @@
 import org.apache.kafka.common.message.BrokerRegistrationRequestData;
 import org.apache.kafka.common.metadata.FenceBrokerRecord;
 import org.apache.kafka.common.metadata.RegisterBrokerRecord;
+import org.apache.kafka.common.metadata.RegisterBrokerRecord.BrokerEndpoint;
+import org.apache.kafka.common.metadata.RegisterBrokerRecord.BrokerEndpointCollection;
+import org.apache.kafka.common.metadata.RegisterBrokerRecord.BrokerFeature;
+import org.apache.kafka.common.metadata.RegisterBrokerRecord.BrokerFeatureCollection;
 import org.apache.kafka.common.metadata.UnfenceBrokerRecord;
 import org.apache.kafka.common.metadata.UnregisterBrokerRecord;
 import org.apache.kafka.common.security.auth.SecurityProtocol;
@@ -40,8 +44,11 @@
 
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
+import java.util.NoSuchElementException;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 
@@ -184,7 +191,7 @@ public ControllerResult<BrokerRegistrationReply> registerBroker(
             setBrokerEpoch(brokerEpoch).
             setRack(request.rack());
         for (BrokerRegistrationRequestData.Listener listener : request.listeners()) {
-            record.endPoints().add(new RegisterBrokerRecord.BrokerEndpoint().
+            record.endPoints().add(new BrokerEndpoint().
                 setHost(listener.host()).
                 setName(listener.name()).
                 setPort(listener.port()).
@@ -199,7 +206,7 @@ public ControllerResult<BrokerRegistrationReply> registerBroker(
                         "the broker has an unsupported version of " + feature.name());
                 }
             }
-            record.features().add(new RegisterBrokerRecord.BrokerFeature().
+            record.features().add(new BrokerFeature().
                 setName(feature.name()).
                 setMinSupportedVersion(feature.minSupportedVersion()).
                 setMaxSupportedVersion(feature.maxSupportedVersion()));
@@ -219,13 +226,13 @@ public ControllerResult<BrokerRegistrationReply> registerBroker(
     public void replay(RegisterBrokerRecord record) {
         int brokerId = record.brokerId();
         List<Endpoint> listeners = new ArrayList<>();
-        for (RegisterBrokerRecord.BrokerEndpoint endpoint : record.endPoints()) {
+        for (BrokerEndpoint endpoint : record.endPoints()) {
             listeners.add(new Endpoint(endpoint.name(),
                 SecurityProtocol.forId(endpoint.securityProtocol()),
                 endpoint.host(), endpoint.port()));
         }
         Map<String, VersionRange> features = new HashMap<>();
-        for (RegisterBrokerRecord.BrokerFeature feature : record.features()) {
+        for (BrokerFeature feature : record.features()) {
             features.put(feature.name(), new VersionRange(
                 feature.minSupportedVersion(), feature.maxSupportedVersion()));
         }
@@ -343,4 +350,56 @@ public void addReadyBrokersFuture(CompletableFuture<Void> future, int minBrokers
             readyBrokersFuture = Optional.empty();
         }
     }
+
+    class ClusterControlIterator implements Iterator<List<ApiMessageAndVersion>> {
+        private final Iterator<Entry<Integer, BrokerRegistration>> iterator;
+
+        ClusterControlIterator(long epoch) {
+            this.iterator = brokerRegistrations.entrySet(epoch).iterator();
+        }
+
+        @Override
+        public boolean hasNext() {
+            return iterator.hasNext();
+        }
+
+        @Override
+        public List<ApiMessageAndVersion> next() {
+            if (!hasNext()) throw new NoSuchElementException();
+            Entry<Integer, BrokerRegistration> entry = iterator.next();
+            int brokerId = entry.getKey();
+            BrokerRegistration registration = entry.getValue();
+            BrokerEndpointCollection endpoints = new BrokerEndpointCollection();
+            for (Entry<String, Endpoint> endpointEntry : registration.listeners().entrySet()) {
+                endpoints.add(new BrokerEndpoint().setName(endpointEntry.getKey()).
+                    setHost(endpointEntry.getValue().host()).
+                    setPort(endpointEntry.getValue().port()).
+                    setSecurityProtocol(endpointEntry.getValue().securityProtocol().id));
+            }
+            BrokerFeatureCollection features = new BrokerFeatureCollection();
+            for (Entry<String, VersionRange> featureEntry : registration.supportedFeatures().entrySet()) {
+                features.add(new BrokerFeature().setName(featureEntry.getKey()).
+                    setMaxSupportedVersion(featureEntry.getValue().max()).
+                    setMinSupportedVersion(featureEntry.getValue().min()));
+            }
+            List<ApiMessageAndVersion> batch = new ArrayList<>();
+            batch.add(new ApiMessageAndVersion(new RegisterBrokerRecord().
+                setBrokerId(brokerId).
+                setIncarnationId(registration.incarnationId()).
+                setBrokerEpoch(registration.epoch()).
+                setEndPoints(endpoints).
+                setFeatures(features).
+                setRack(registration.rack().orElse(null)), (short) 0));
+            if (!registration.fenced()) {
+                batch.add(new ApiMessageAndVersion(new UnfenceBrokerRecord().
+                    setId(brokerId).
+                    setEpoch(registration.epoch()), (short) 0));
+            }
+            return batch;
+        }
+    }
+
+    ClusterControlIterator iterator(long epoch) {
+        return new ClusterControlIterator(epoch);
+    }
 }

metadata/src/main/java/org/apache/kafka/controller/ConfigurationControlManager.java

@@ -40,10 +40,12 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.NoSuchElementException;
 import java.util.Objects;
 
 import static org.apache.kafka.clients.admin.AlterConfigOp.OpType.APPEND;
 
+
 public class ConfigurationControlManager {
     private final Logger log;
     private final SnapshotRegistry snapshotRegistry;
@@ -316,6 +318,9 @@ public void replay(ConfigRecord record) {
         } else {
             configs.put(record.name(), record.value());
         }
+        if (configs.isEmpty()) {
+            configData.remove(configResource);
+        }
         log.info("{}: set configuration {} to {}", configResource, record.name(), record.value());
     }
 
@@ -368,4 +373,39 @@ public Map<ConfigResource, ResultOrError<Map<String, String>>> describeConfigs(
     void deleteTopicConfigs(String name) {
         configData.remove(new ConfigResource(Type.TOPIC, name));
     }
+
+    class ConfigurationControlIterator implements Iterator<List<ApiMessageAndVersion>> {
+        private final long epoch;
+        private final Iterator<Entry<ConfigResource, TimelineHashMap<String, String>>> iterator;
+
+        ConfigurationControlIterator(long epoch) {
+            this.epoch = epoch;
+            this.iterator = configData.entrySet(epoch).iterator();
+        }
+
+        @Override
+        public boolean hasNext() {
+            return iterator.hasNext();
+        }
+
+        @Override
+        public List<ApiMessageAndVersion> next() {
+            if (!hasNext()) throw new NoSuchElementException();
+            List<ApiMessageAndVersion> records = new ArrayList<>();
+            Entry<ConfigResource, TimelineHashMap<String, String>> entry = iterator.next();
+            ConfigResource resource = entry.getKey();
+            for (Entry<String, String> configEntry : entry.getValue().entrySet(epoch)) {
+                records.add(new ApiMessageAndVersion(new ConfigRecord().
+                    setResourceName(resource.name()).
+                    setResourceType(resource.type().id()).
+                    setName(configEntry.getKey()).
+                    setValue(configEntry.getValue()), (short) 0));
+            }
+            return records;
+        }
+    }
+
+    ConfigurationControlIterator iterator(long epoch) {
+        return new ConfigurationControlIterator(epoch);
+    }
 }

metadata/src/main/java/org/apache/kafka/controller/Controller.java

@@ -187,6 +187,14 @@ CompletableFuture<Map<ClientQuotaEntity, ApiError>> alterClientQuotas(
         Collection<ClientQuotaAlteration> quotaAlterations, boolean validateOnly
     );
 
+    /**
+     * Begin writing a controller snapshot.  If there was already an ongoing snapshot, it
+     * simply returns information about that snapshot rather than starting a new one.
+     *
+     * @return              A future yielding the epoch of the snapshot.
+     */
+    CompletableFuture<Long> beginWritingSnapshot();
+
     /**
      * Begin shutting down, but don't block.  You must still call close to clean up all
      * resources.

metadata/src/main/java/org/apache/kafka/controller/EmptySnapshotReader.java

@@ -0,0 +1,52 @@
+/*
+ * 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.controller;
+
+import java.util.List;
+import java.util.NoSuchElementException;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+
+
+public class EmptySnapshotReader implements SnapshotReader {
+    private final long epoch;
+
+    public EmptySnapshotReader(long epoch) {
+        this.epoch = epoch;
+    }
+
+    @Override
+    public long epoch() {
+        return epoch;
+    }
+
+    @Override
+    public void close() {
+        // Nothing to do
+    }
+
+    @Override
+    public boolean hasNext() {
+        return false;
+    }
+
+    @Override
+    public List<ApiMessage> next() {
+        throw new NoSuchElementException();
+    }
+}