diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml index 5473b035ea919..2ed1fac5df95d 100644 --- a/checkstyle/suppressions.xml +++ b/checkstyle/suppressions.xml @@ -268,6 +268,8 @@ files="(ReplicationControlManager|ReplicationControlManagerTest).java"/> + waitForReadyBrokers(int minBrokers) { throw new UnsupportedOperationException(); } + @Override + public CompletableFuture beginWritingSnapshot() { + throw new UnsupportedOperationException(); + } + @Override public void beginShutdown() { this.active = false; diff --git a/metadata/src/main/java/org/apache/kafka/controller/ClientQuotaControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ClientQuotaControlManager.java index 9b8e2d683b650..f6a24973b58f2 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/ClientQuotaControlManager.java +++ b/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> clientQuotaData; + final TimelineHashMap> clientQuotaData; ClientQuotaControlManager(SnapshotRegistry snapshotRegistry) { this.snapshotRegistry = snapshotRegistry; @@ -98,7 +101,7 @@ public void replay(QuotaRecord record) { Map entityMap = new HashMap<>(2); record.entity().forEach(entityData -> entityMap.put(entityData.entityType(), entityData.entityName())); ClientQuotaEntity entity = new ClientQuotaEntity(entityMap); - Map quotas = clientQuotaData.get(entity); + TimelineHashMap 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> recordEntitySupplier = () -> - validatedEntityMap.entrySet().stream().map(mapEntry -> new QuotaRecord.EntityData() + Supplier> recordEntitySupplier = () -> + validatedEntityMap.entrySet().stream().map(mapEntry -> new EntityData() .setEntityType(mapEntry.getKey()) .setEntityName(mapEntry.getValue())) .collect(Collectors.toList()); List newRecords = new ArrayList<>(newQuotaConfigs.size()); - Map currentQuotas = clientQuotaData.getOrDefault(entity, Collections.emptyMap()); + Map 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 va return new ApiError(Errors.INVALID_REQUEST, "Invalid empty client quota entity"); } - for (Map.Entry entityEntry : entity.entries().entrySet()) { + for (Entry 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 va return ApiError.NONE; } + + class ClientQuotaControlIterator implements Iterator> { + private final long epoch; + private final Iterator>> iterator; + + ClientQuotaControlIterator(long epoch) { + this.epoch = epoch; + this.iterator = clientQuotaData.entrySet(epoch).iterator(); + } + + @Override + public boolean hasNext() { + return iterator.hasNext(); + } + + @Override + public List next() { + if (!hasNext()) throw new NoSuchElementException(); + Entry> entry = iterator.next(); + ClientQuotaEntity entity = entry.getKey(); + List records = new ArrayList<>(); + for (Entry quotaEntry : entry.getValue().entrySet(epoch)) { + QuotaRecord record = new QuotaRecord(); + for (Entry 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); + } } diff --git a/metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java index 4748d195986ab..05380fc49e421 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java +++ b/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 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 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 registerBroker( public void replay(RegisterBrokerRecord record) { int brokerId = record.brokerId(); List 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 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 future, int minBrokers readyBrokersFuture = Optional.empty(); } } + + class ClusterControlIterator implements Iterator> { + private final Iterator> iterator; + + ClusterControlIterator(long epoch) { + this.iterator = brokerRegistrations.entrySet(epoch).iterator(); + } + + @Override + public boolean hasNext() { + return iterator.hasNext(); + } + + @Override + public List next() { + if (!hasNext()) throw new NoSuchElementException(); + Entry entry = iterator.next(); + int brokerId = entry.getKey(); + BrokerRegistration registration = entry.getValue(); + BrokerEndpointCollection endpoints = new BrokerEndpointCollection(); + for (Entry 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 featureEntry : registration.supportedFeatures().entrySet()) { + features.add(new BrokerFeature().setName(featureEntry.getKey()). + setMaxSupportedVersion(featureEntry.getValue().max()). + setMinSupportedVersion(featureEntry.getValue().min())); + } + List 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); + } } diff --git a/metadata/src/main/java/org/apache/kafka/controller/ConfigurationControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ConfigurationControlManager.java index dcfe92d46a54b..3e9e9e1d593c8 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/ConfigurationControlManager.java +++ b/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>> describeConfigs( void deleteTopicConfigs(String name) { configData.remove(new ConfigResource(Type.TOPIC, name)); } + + class ConfigurationControlIterator implements Iterator> { + private final long epoch; + private final Iterator>> iterator; + + ConfigurationControlIterator(long epoch) { + this.epoch = epoch; + this.iterator = configData.entrySet(epoch).iterator(); + } + + @Override + public boolean hasNext() { + return iterator.hasNext(); + } + + @Override + public List next() { + if (!hasNext()) throw new NoSuchElementException(); + List records = new ArrayList<>(); + Entry> entry = iterator.next(); + ConfigResource resource = entry.getKey(); + for (Entry 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); + } } diff --git a/metadata/src/main/java/org/apache/kafka/controller/Controller.java b/metadata/src/main/java/org/apache/kafka/controller/Controller.java index 26394632e6a94..7892c1991ee9a 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/Controller.java +++ b/metadata/src/main/java/org/apache/kafka/controller/Controller.java @@ -187,6 +187,14 @@ CompletableFuture> alterClientQuotas( Collection 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 beginWritingSnapshot(); + /** * Begin shutting down, but don't block. You must still call close to clean up all * resources. diff --git a/metadata/src/main/java/org/apache/kafka/controller/EmptySnapshotReader.java b/metadata/src/main/java/org/apache/kafka/controller/EmptySnapshotReader.java new file mode 100644 index 0000000000000..cf334c75ec0ee --- /dev/null +++ b/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 next() { + throw new NoSuchElementException(); + } +} diff --git a/metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java index 99874ac3c5ef7..fb6d7dbcce740 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java +++ b/metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java @@ -18,13 +18,16 @@ package org.apache.kafka.controller; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.Iterator; import java.util.List; import java.util.Map.Entry; import java.util.Map; +import java.util.NoSuchElementException; import java.util.Set; import java.util.TreeMap; + import org.apache.kafka.common.metadata.FeatureLevelRecord; import org.apache.kafka.common.protocol.Errors; import org.apache.kafka.common.requests.ApiError; @@ -34,12 +37,11 @@ import org.apache.kafka.metadata.VersionRange; import org.apache.kafka.timeline.SnapshotRegistry; import org.apache.kafka.timeline.TimelineHashMap; -import org.apache.kafka.timeline.TimelineHashSet; public class FeatureControlManager { /** - * The features supported by this controller's software. + * An immutable map containing the features supported by this controller's software. */ private final Map supportedFeatures; @@ -48,16 +50,10 @@ public class FeatureControlManager { */ private final TimelineHashMap finalizedVersions; - /** - * The latest feature epoch. - */ - private final TimelineHashSet epoch; - FeatureControlManager(Map supportedFeatures, SnapshotRegistry snapshotRegistry) { this.supportedFeatures = supportedFeatures; this.finalizedVersions = new TimelineHashMap<>(snapshotRegistry, 0); - this.epoch = new TimelineHashSet<>(snapshotRegistry, 0); } ControllerResult> updateFeatures( @@ -120,18 +116,39 @@ FeatureMapAndEpoch finalizedFeatures(long lastCommittedOffset) { for (Entry entry : finalizedVersions.entrySet(lastCommittedOffset)) { features.put(entry.getKey(), entry.getValue()); } - long currentEpoch = -1; - Iterator iterator = epoch.iterator(lastCommittedOffset); - if (iterator.hasNext()) { - currentEpoch = iterator.next(); - } - return new FeatureMapAndEpoch(new FeatureMap(features), currentEpoch); + return new FeatureMapAndEpoch(new FeatureMap(features), lastCommittedOffset); } - void replay(FeatureLevelRecord record, long offset) { + public void replay(FeatureLevelRecord record) { finalizedVersions.put(record.name(), new VersionRange(record.minFeatureLevel(), record.maxFeatureLevel())); - epoch.clear(); - epoch.add(offset); + } + + class FeatureControlIterator implements Iterator> { + private final Iterator> iterator; + + FeatureControlIterator(long epoch) { + this.iterator = finalizedVersions.entrySet(epoch).iterator(); + } + + @Override + public boolean hasNext() { + return iterator.hasNext(); + } + + @Override + public List next() { + if (!hasNext()) throw new NoSuchElementException(); + Entry entry = iterator.next(); + VersionRange versions = entry.getValue(); + return Collections.singletonList(new ApiMessageAndVersion(new FeatureLevelRecord(). + setName(entry.getKey()). + setMinFeatureLevel(versions.min()). + setMaxFeatureLevel(versions.max()), (short) 0)); + } + } + + FeatureControlIterator iterator(long epoch) { + return new FeatureControlIterator(epoch); } } 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 4f2b708bfe5a4..9730f0ca898af 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java +++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java @@ -17,16 +17,19 @@ package org.apache.kafka.controller; +import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.List; import java.util.Map.Entry; import java.util.Map; import java.util.Optional; +import java.util.OptionalLong; import java.util.Random; import java.util.concurrent.CompletableFuture; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; +import java.util.function.Function; import java.util.function.Supplier; import java.util.stream.Collectors; import org.apache.kafka.clients.admin.AlterConfigOp.OpType; @@ -60,9 +63,11 @@ import org.apache.kafka.common.quota.ClientQuotaAlteration; import org.apache.kafka.common.quota.ClientQuotaEntity; import org.apache.kafka.common.requests.ApiError; +import org.apache.kafka.common.utils.ExponentialBackoff; 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.controller.SnapshotGenerator.Section; import org.apache.kafka.metadata.ApiMessageAndVersion; import org.apache.kafka.metadata.BrokerHeartbeatReply; import org.apache.kafka.metadata.BrokerRegistrationReply; @@ -113,6 +118,8 @@ static public class Builder { private int defaultNumPartitions = 1; private ReplicaPlacementPolicy replicaPlacementPolicy = new SimpleReplicaPlacementPolicy(new Random()); + private Function snapshotWriterBuilder; + private SnapshotReader snapshotReader; private long sessionTimeoutNs = NANOSECONDS.convert(18, TimeUnit.SECONDS); private ControllerMetrics controllerMetrics = null; @@ -165,6 +172,16 @@ public Builder setReplicaPlacementPolicy(ReplicaPlacementPolicy replicaPlacement return this; } + public Builder setSnapshotWriterBuilder(Function snapshotWriterBuilder) { + this.snapshotWriterBuilder = snapshotWriterBuilder; + return this; + } + + public Builder setSnapshotReader(SnapshotReader snapshotReader) { + this.snapshotReader = snapshotReader; + return this; + } + public Builder setSessionTimeoutNs(long sessionTimeoutNs) { this.sessionTimeoutNs = sessionTimeoutNs; return this; @@ -175,6 +192,7 @@ public Builder setMetrics(ControllerMetrics controllerMetrics) { return this; } + @SuppressWarnings("unchecked") public QuorumController build() throws Exception { if (logManager == null) { throw new RuntimeException("You must set a metadata log manager."); @@ -189,16 +207,25 @@ public QuorumController build() throws Exception { controllerMetrics = (ControllerMetrics) Class.forName( "org.apache.kafka.controller.MockControllerMetrics").getConstructor().newInstance(); } + if (snapshotWriterBuilder == null) { + snapshotWriterBuilder = (Function) Class.forName( + "org.apache.kafka.controller.NoOpSnapshotWriterBuilder").getConstructor().newInstance(); + } + if (snapshotReader == null) { + snapshotReader = new EmptySnapshotReader(-1); + } KafkaEventQueue queue = null; try { queue = new KafkaEventQueue(time, logContext, threadNamePrefix); return new QuorumController(logContext, nodeId, queue, time, configDefs, - logManager, supportedFeatures, defaultReplicationFactor, - defaultNumPartitions, replicaPlacementPolicy, sessionTimeoutNs, - controllerMetrics); + logManager, supportedFeatures, defaultReplicationFactor, + defaultNumPartitions, replicaPlacementPolicy, snapshotWriterBuilder, + snapshotReader, sessionTimeoutNs, controllerMetrics); } catch (Exception e) { Utils.closeQuietly(queue, "event queue"); throw e; + } finally { + Utils.closeQuietly(snapshotReader, "snapshotReader"); } } } @@ -301,6 +328,93 @@ private void appendControlEvent(String name, Runnable handler) { queue.append(event); } + private static final String GENERATE_SNAPSHOT = "generateSnapshot"; + + private static final int MAX_BATCHES_PER_GENERATE_CALL = 10; + + class SnapshotGeneratorManager implements Runnable { + private final Function writerBuilder; + private final ExponentialBackoff exponentialBackoff = + new ExponentialBackoff(10, 2, 5000, 0); + private SnapshotGenerator generator = null; + + SnapshotGeneratorManager(Function writerBuilder) { + this.writerBuilder = writerBuilder; + } + + void createSnapshotGenerator(long epoch) { + if (generator != null) { + throw new RuntimeException("Snapshot generator already exists."); + } + if (!snapshotRegistry.hasSnapshot(epoch)) { + throw new RuntimeException("Can't generate a snapshot at epoch " + epoch + + " because no such epoch exists in the snapshot registry."); + } + generator = new SnapshotGenerator(logContext, + writerBuilder.apply(epoch), + MAX_BATCHES_PER_GENERATE_CALL, + exponentialBackoff, + Arrays.asList( + new Section("features", featureControl.iterator(epoch)), + new Section("cluster", clusterControl.iterator(epoch)), + new Section("replication", replicationControl.iterator(epoch)), + new Section("configuration", configurationControl.iterator(epoch)), + new Section("clientQuotas", clientQuotaControlManager.iterator(epoch)))); + reschedule(0); + } + + void cancel() { + if (generator == null) return; + log.error("Cancelling snapshot {}", generator.epoch()); + generator.writer().close(); + generator = null; + queue.cancelDeferred(GENERATE_SNAPSHOT); + } + + void reschedule(long delayNs) { + ControlEvent event = new ControlEvent(GENERATE_SNAPSHOT, this); + queue.scheduleDeferred(event.name, + new EarliestDeadlineFunction(time.nanoseconds() + delayNs), event); + } + + @Override + public void run() { + if (generator == null) { + log.debug("No snapshot is in progress."); + return; + } + OptionalLong nextDelay; + try { + nextDelay = generator.generateBatches(); + } catch (Exception e) { + log.error("Error while generating snapshot {}", generator.epoch(), e); + generator.writer().close(); + generator = null; + return; + } + if (!nextDelay.isPresent()) { + try { + generator.writer().completeSnapshot(); + log.info("Finished generating snapshot {}.", generator.epoch()); + } catch (Exception e) { + log.error("Error while completing snapshot {}", generator.epoch(), e); + } finally { + generator.writer().close(); + generator = null; + } + return; + } + reschedule(nextDelay.getAsLong()); + } + + long snapshotEpoch() { + if (generator == null) { + return Long.MAX_VALUE; + } + return generator.epoch(); + } + } + /** * A controller event that reads the committed internal state in order to expose it * to an API. @@ -451,7 +565,7 @@ public void run() throws Exception { writeOffset = offset; resultAndOffset = ControllerResultAndOffset.of(offset, result); for (ApiMessageAndVersion message : result.records()) { - replay(message.message(), offset); + replay(message.message(), -1, offset); } snapshotRegistry.createSnapshot(offset); log.debug("Read-write operation {} will be completed when the log " + @@ -516,7 +630,7 @@ public void handleCommits(long offset, List messages) { } } for (ApiMessage message : messages) { - replay(message, offset); + replay(message, -1, offset); } } else { // If the controller is active, the records were already replayed, @@ -526,9 +640,11 @@ public void handleCommits(long offset, List messages) { // Complete any events in the purgatory that were waiting for this offset. purgatory.completeUpTo(offset); - // Delete all snapshots older than the offset. - // TODO: add an exception here for when we're writing out a log snapshot - snapshotRegistry.deleteSnapshotsUpTo(offset); + // Delete all the in-memory snapshots that we no longer need. + // If we are writing a new snapshot, then we need to keep that around; + // otherwise, we should delete up to the current committed offset. + snapshotRegistry.deleteSnapshotsUpTo( + Math.min(offset, snapshotGeneratorManager.snapshotEpoch())); } lastCommittedOffset = offset; }); @@ -545,7 +661,7 @@ public void handleNewLeader(MetaLogLeader newLeader) { newEpoch + ", but we never renounced controller epoch " + curEpoch); } - log.info("Becoming active at controller epoch {}.", newEpoch); + log.warn("Becoming active at controller epoch {}.", newEpoch); curClaimEpoch = newEpoch; controllerMetrics.setActive(true); writeOffset = lastCommittedOffset; @@ -558,9 +674,9 @@ public void handleNewLeader(MetaLogLeader newLeader) { public void handleRenounce(long oldEpoch) { appendControlEvent("handleRenounce[" + oldEpoch + "]", () -> { if (curClaimEpoch == oldEpoch) { - log.info("Renouncing the leadership at oldEpoch {} due to a metadata " + - "log event. Reverting to last committed offset {}.", curClaimEpoch, - lastCommittedOffset); + log.warn("Renouncing the leadership at oldEpoch {} due to a metadata " + + "log event. Reverting to last committed offset {}.", curClaimEpoch, + lastCommittedOffset); renounce(); } }); @@ -626,7 +742,7 @@ private void cancelMaybeFenceReplicas() { } @SuppressWarnings("unchecked") - private void replay(ApiMessage message, long offset) { + private void replay(ApiMessage message, long snapshotEpoch, long offset) { try { MetadataRecordType type = MetadataRecordType.fromId(message.apiKey()); switch (type) { @@ -658,7 +774,7 @@ private void replay(ApiMessage message, long offset) { replicationControl.replay((RemoveTopicRecord) message); break; case FEATURE_LEVEL_RECORD: - featureControl.replay((FeatureLevelRecord) message, offset); + featureControl.replay((FeatureLevelRecord) message); break; case QUOTA_RECORD: clientQuotaControlManager.replay((QuotaRecord) message); @@ -667,10 +783,18 @@ private void replay(ApiMessage message, long offset) { throw new RuntimeException("Unhandled record type " + type); } } catch (Exception e) { - log.error("Error replaying record {}", message.toString(), e); + if (snapshotEpoch < 0) { + log.error("Error replaying record {} at offset {}.", + message.toString(), offset, e); + } else { + log.error("Error replaying record {} from snapshot {} at index {}.", + message.toString(), snapshotEpoch, offset, e); + } } } + private final LogContext logContext; + private final Logger log; /** @@ -735,6 +859,11 @@ private void replay(ApiMessage message, long offset) { */ private final ReplicationControlManager replicationControl; + /** + * Manages generating controller snapshots. + */ + private final SnapshotGeneratorManager snapshotGeneratorManager; + /** * The interface that we use to mutate the Raft log. */ @@ -773,15 +902,17 @@ private QuorumController(LogContext logContext, short defaultReplicationFactor, int defaultNumPartitions, ReplicaPlacementPolicy replicaPlacementPolicy, + Function snapshotWriterBuilder, + SnapshotReader snapshotReader, long sessionTimeoutNs, ControllerMetrics controllerMetrics) throws Exception { + this.logContext = logContext; this.log = logContext.logger(QuorumController.class); this.nodeId = nodeId; this.queue = queue; this.time = time; this.controllerMetrics = controllerMetrics; this.snapshotRegistry = new SnapshotRegistry(logContext); - snapshotRegistry.createSnapshot(-1); this.purgatory = new ControllerPurgatory(); this.configurationControl = new ConfigurationControlManager(logContext, snapshotRegistry, configDefs); @@ -789,14 +920,24 @@ private QuorumController(LogContext logContext, this.clusterControl = new ClusterControlManager(logContext, time, snapshotRegistry, sessionTimeoutNs, replicaPlacementPolicy); this.featureControl = new FeatureControlManager(supportedFeatures, snapshotRegistry); + this.snapshotGeneratorManager = new SnapshotGeneratorManager(snapshotWriterBuilder); this.replicationControl = new ReplicationControlManager(snapshotRegistry, logContext, defaultReplicationFactor, defaultNumPartitions, configurationControl, clusterControl); this.logManager = logManager; this.metaLogListener = new QuorumMetaLogListener(); this.curClaimEpoch = -1L; - this.lastCommittedOffset = -1L; + this.lastCommittedOffset = snapshotReader.epoch(); this.writeOffset = -1L; + + while (snapshotReader.hasNext()) { + List batch = snapshotReader.next(); + long index = 0; + for (ApiMessage message : batch) { + replay(message, snapshotReader.epoch(), index++); + } + } + snapshotRegistry.createSnapshot(lastCommittedOffset); this.logManager.register(metaLogListener); } @@ -954,6 +1095,18 @@ public CompletableFuture> alterClientQuotas( }); } + @Override + public CompletableFuture beginWritingSnapshot() { + CompletableFuture future = new CompletableFuture<>(); + appendControlEvent("beginWritingSnapshot", () -> { + if (snapshotGeneratorManager.generator == null) { + snapshotGeneratorManager.createSnapshotGenerator(lastCommittedOffset); + } + future.complete(snapshotGeneratorManager.generator.epoch()); + }); + return future; + } + @Override public CompletableFuture waitForReadyBrokers(int minBrokers) { final CompletableFuture future = new CompletableFuture<>(); diff --git a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java index 97dee1c590887..10e849091931f 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java +++ b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java @@ -71,6 +71,7 @@ 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.SET; @@ -1005,4 +1006,47 @@ ControllerResult maybeFenceStaleBrokers() { } return ControllerResult.of(records, null); } + + class ReplicationControlIterator implements Iterator> { + private final long epoch; + private final Iterator iterator; + + ReplicationControlIterator(long epoch) { + this.epoch = epoch; + this.iterator = topics.values(epoch).iterator(); + } + + @Override + public boolean hasNext() { + return iterator.hasNext(); + } + + @Override + public List next() { + if (!hasNext()) throw new NoSuchElementException(); + TopicControlInfo topic = iterator.next(); + List records = new ArrayList<>(); + records.add(new ApiMessageAndVersion(new TopicRecord(). + setName(topic.name). + setTopicId(topic.id), (short) 0)); + for (Entry entry : topic.parts.entrySet(epoch)) { + PartitionControlInfo partition = entry.getValue(); + records.add(new ApiMessageAndVersion(new PartitionRecord(). + setPartitionId(entry.getKey()). + setTopicId(topic.id). + setReplicas(Replicas.toList(partition.replicas)). + setIsr(Replicas.toList(partition.isr)). + setRemovingReplicas(Replicas.toList(partition.removingReplicas)). + setAddingReplicas(Replicas.toList(partition.addingReplicas)). + setLeader(partition.leader). + setLeaderEpoch(partition.leaderEpoch). + setPartitionEpoch(partition.partitionEpoch), (short) 0)); + } + return records; + } + } + + ReplicationControlIterator iterator(long epoch) { + return new ReplicationControlIterator(epoch); + } } diff --git a/metadata/src/main/java/org/apache/kafka/controller/SnapshotGenerator.java b/metadata/src/main/java/org/apache/kafka/controller/SnapshotGenerator.java new file mode 100644 index 0000000000000..91295c2a2f769 --- /dev/null +++ b/metadata/src/main/java/org/apache/kafka/controller/SnapshotGenerator.java @@ -0,0 +1,146 @@ +/* + * 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.Collections; +import java.util.Iterator; +import java.util.List; +import java.util.OptionalLong; + +import org.apache.kafka.common.utils.ExponentialBackoff; +import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.metadata.ApiMessageAndVersion; +import org.slf4j.Logger; + + +final class SnapshotGenerator { + static class Section { + private final String name; + private final Iterator> iterator; + + Section(String name, Iterator> iterator) { + this.name = name; + this.iterator = iterator; + } + + String name() { + return name; + } + + Iterator> iterator() { + return iterator; + } + } + + private final Logger log; + private final SnapshotWriter writer; + private final int maxBatchesPerGenerateCall; + private final ExponentialBackoff exponentialBackoff; + private final List
sections; + private final Iterator
sectionIterator; + private Iterator> batchIterator; + private List batch; + private Section section; + private long numRecords; + private long numWriteTries; + + SnapshotGenerator(LogContext logContext, + SnapshotWriter writer, + int maxBatchesPerGenerateCall, + ExponentialBackoff exponentialBackoff, + List
sections) { + this.log = logContext.logger(SnapshotGenerator.class); + this.writer = writer; + this.maxBatchesPerGenerateCall = maxBatchesPerGenerateCall; + this.exponentialBackoff = exponentialBackoff; + this.sections = sections; + this.sectionIterator = this.sections.iterator(); + this.batchIterator = Collections.emptyIterator(); + this.batch = null; + this.section = null; + this.numRecords = 0; + this.numWriteTries = 0; + } + + /** + * Returns the epoch of the snapshot that we are generating. + */ + long epoch() { + return writer.epoch(); + } + + SnapshotWriter writer() { + return writer; + } + + /** + * Generate the next batch of records. + * + * @return 0 if a batch was sent to the writer, + * -1 if there are no more batches to generate, + * or the number of times we tried to write and the writer + * was busy, otherwise. + */ + private long generateBatch() throws Exception { + if (batch == null) { + while (!batchIterator.hasNext()) { + if (section != null) { + log.info("Generated {} record(s) for the {} section of snapshot {}.", + numRecords, section.name(), writer.epoch()); + section = null; + numRecords = 0; + } + if (!sectionIterator.hasNext()) { + writer.completeSnapshot(); + return -1; + } + section = sectionIterator.next(); + log.info("Generating records for the {} section of snapshot {}.", + section.name(), writer.epoch()); + batchIterator = section.iterator(); + } + batch = batchIterator.next(); + } + if (writer.writeBatch(batch)) { + numRecords += batch.size(); + numWriteTries = 0; + batch = null; + return 0; + } else { + return ++numWriteTries; + } + } + + /** + * Generate the next few batches of records. + * + * @return The number of nanoseconds to delay before rescheduling the + * generateBatches event, or empty if the snapshot is done. + */ + OptionalLong generateBatches() throws Exception { + for (int numBatches = 0; numBatches < maxBatchesPerGenerateCall; numBatches++) { + long result = generateBatch(); + if (result < 0) { + return OptionalLong.empty(); + } else if (result > 0) { + return OptionalLong.of(exponentialBackoff.backoff(result - 1)); + } + } + return OptionalLong.of(0); + } +} diff --git a/metadata/src/main/java/org/apache/kafka/controller/SnapshotReader.java b/metadata/src/main/java/org/apache/kafka/controller/SnapshotReader.java new file mode 100644 index 0000000000000..1a994ed41edab --- /dev/null +++ b/metadata/src/main/java/org/apache/kafka/controller/SnapshotReader.java @@ -0,0 +1,37 @@ +/* + * 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 org.apache.kafka.common.protocol.ApiMessage; + +import java.util.Iterator; +import java.util.List; + + +interface SnapshotReader extends Iterator>, AutoCloseable { + /** + * Returns the snapshot epoch, which is the offset of this snapshot within the log. + */ + long epoch(); + + /** + * Invoked when the snapshot reader is no longer needed. This should clean + * up all reader resources. + */ + void close(); +} diff --git a/metadata/src/main/java/org/apache/kafka/controller/SnapshotWriter.java b/metadata/src/main/java/org/apache/kafka/controller/SnapshotWriter.java new file mode 100644 index 0000000000000..595f4c181b2fa --- /dev/null +++ b/metadata/src/main/java/org/apache/kafka/controller/SnapshotWriter.java @@ -0,0 +1,55 @@ +/* + * 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.io.IOException; +import java.util.List; +import org.apache.kafka.metadata.ApiMessageAndVersion; + + +interface SnapshotWriter extends AutoCloseable { + /** + * @return The epoch of the snapshot we are writing. + **/ + long epoch(); + + /** + * Write a batch of records. The writer may do more batching than is + * requested-- in other words, small batches may be coalesced into large + * ones, if that is convenient. + * + * @param batch The batch of records. + * + * @return True if the data was received; false if the writer is + * not ready for the data yet. If the writer returns false + * the snapshot generator will try again after a delay. + */ + boolean writeBatch(List batch) throws IOException; + + /** + * Invoked when the snapshot writer is no longer needed. This should clean + * up all writer resources. If this is called prior to completing the snapshot, + * any partial snapshot on disk should be deleted. + */ + void close(); + + /** + * Invoked to finish writing the snapshot to disk. + */ + void completeSnapshot() throws IOException; +} diff --git a/metadata/src/main/java/org/apache/kafka/timeline/SnapshotRegistry.java b/metadata/src/main/java/org/apache/kafka/timeline/SnapshotRegistry.java index 29a79b60faf95..90be3e5e45ed7 100644 --- a/metadata/src/main/java/org/apache/kafka/timeline/SnapshotRegistry.java +++ b/metadata/src/main/java/org/apache/kafka/timeline/SnapshotRegistry.java @@ -151,6 +151,10 @@ public List epochsList() { return result; } + public boolean hasSnapshot(long epoch) { + return snapshots.containsKey(epoch); + } + /** * Gets the snapshot for a specific epoch. */ diff --git a/metadata/src/test/java/org/apache/kafka/controller/ClientQuotaControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/ClientQuotaControlManagerTest.java index c4e8da87104f3..f24853a16a7b3 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/ClientQuotaControlManagerTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/ClientQuotaControlManagerTest.java @@ -19,16 +19,19 @@ import org.apache.kafka.common.config.internals.QuotaConfigs; 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; import org.apache.kafka.common.requests.ApiError; import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.metadata.ApiMessageAndVersion; import org.apache.kafka.timeline.SnapshotRegistry; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.HashMap; @@ -151,7 +154,7 @@ public void testAlterAndRemove() { } @Test - public void testEntityTypes() { + public void testEntityTypes() throws Exception { SnapshotRegistry snapshotRegistry = new SnapshotRegistry(new LogContext()); ClientQuotaControlManager manager = new ClientQuotaControlManager(snapshotRegistry); @@ -179,10 +182,51 @@ public void testEntityTypes() { quotasToTest.put(clientEntity("client-id-2"), quotas(QuotaConfigs.REQUEST_PERCENTAGE_OVERRIDE_CONFIG, 60.60)); - List alters = new ArrayList<>(); quotasToTest.forEach((entity, quota) -> entityQuotaToAlterations(entity, quota, alters::add)); alterQuotas(alters, manager); + + ControllerTestUtils.assertBatchIteratorContains(Arrays.asList( + Arrays.asList(new ApiMessageAndVersion(new QuotaRecord().setEntity(Arrays.asList( + new EntityData().setEntityType("user").setEntityName("user-1"), + new EntityData().setEntityType("client-id").setEntityName("client-id-1"))). + setKey("request_percentage").setValue(50.5).setRemove(false), (short) 0)), + Arrays.asList(new ApiMessageAndVersion(new QuotaRecord().setEntity(Arrays.asList( + new EntityData().setEntityType("user").setEntityName("user-2"), + new EntityData().setEntityType("client-id").setEntityName("client-id-1"))). + setKey("request_percentage").setValue(51.51).setRemove(false), (short) 0)), + Arrays.asList(new ApiMessageAndVersion(new QuotaRecord().setEntity(Arrays.asList( + new EntityData().setEntityType("user").setEntityName("user-3"), + new EntityData().setEntityType("client-id").setEntityName("client-id-2"))). + setKey("request_percentage").setValue(52.52).setRemove(false), (short) 0)), + Arrays.asList(new ApiMessageAndVersion(new QuotaRecord().setEntity(Arrays.asList( + new EntityData().setEntityType("user").setEntityName(null), + new EntityData().setEntityType("client-id").setEntityName("client-id-1"))). + setKey("request_percentage").setValue(53.53).setRemove(false), (short) 0)), + Arrays.asList(new ApiMessageAndVersion(new QuotaRecord().setEntity(Arrays.asList( + new EntityData().setEntityType("user").setEntityName("user-1"), + new EntityData().setEntityType("client-id").setEntityName(null))). + setKey("request_percentage").setValue(54.54).setRemove(false), (short) 0)), + Arrays.asList(new ApiMessageAndVersion(new QuotaRecord().setEntity(Arrays.asList( + new EntityData().setEntityType("user").setEntityName("user-3"), + new EntityData().setEntityType("client-id").setEntityName(null))). + setKey("request_percentage").setValue(55.55).setRemove(false), (short) 0)), + Arrays.asList(new ApiMessageAndVersion(new QuotaRecord().setEntity(Arrays.asList( + new EntityData().setEntityType("user").setEntityName("user-1"))). + setKey("request_percentage").setValue(56.56).setRemove(false), (short) 0)), + Arrays.asList(new ApiMessageAndVersion(new QuotaRecord().setEntity(Arrays.asList( + new EntityData().setEntityType("user").setEntityName("user-2"))). + setKey("request_percentage").setValue(57.57).setRemove(false), (short) 0)), + Arrays.asList(new ApiMessageAndVersion(new QuotaRecord().setEntity(Arrays.asList( + new EntityData().setEntityType("user").setEntityName("user-3"))). + setKey("request_percentage").setValue(58.58).setRemove(false), (short) 0)), + Arrays.asList(new ApiMessageAndVersion(new QuotaRecord().setEntity(Arrays.asList( + new EntityData().setEntityType("user").setEntityName(null))). + setKey("request_percentage").setValue(59.59).setRemove(false), (short) 0)), + Arrays.asList(new ApiMessageAndVersion(new QuotaRecord().setEntity(Arrays.asList( + new EntityData().setEntityType("client-id").setEntityName("client-id-2"))). + setKey("request_percentage").setValue(60.60).setRemove(false), (short) 0))), + manager.iterator(Long.MAX_VALUE)); } static void entityQuotaToAlterations(ClientQuotaEntity entity, Map quota, diff --git a/metadata/src/test/java/org/apache/kafka/controller/ClusterControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/ClusterControlManagerTest.java index c410a686344d2..009e8811e1514 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/ClusterControlManagerTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/ClusterControlManagerTest.java @@ -17,6 +17,7 @@ package org.apache.kafka.controller; +import java.util.Arrays; import java.util.Collections; import java.util.HashSet; import java.util.List; @@ -26,11 +27,14 @@ import org.apache.kafka.common.Uuid; import org.apache.kafka.common.errors.StaleBrokerEpochException; 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.UnfenceBrokerRecord; import org.apache.kafka.common.metadata.UnregisterBrokerRecord; import org.apache.kafka.common.security.auth.SecurityProtocol; import org.apache.kafka.common.utils.LogContext; import org.apache.kafka.common.utils.MockTime; +import org.apache.kafka.metadata.ApiMessageAndVersion; import org.apache.kafka.metadata.BrokerRegistration; import org.apache.kafka.timeline.SnapshotRegistry; import org.junit.jupiter.api.Test; @@ -58,7 +62,7 @@ public void testReplay() { assertFalse(clusterControl.unfenced(0)); RegisterBrokerRecord brokerRecord = new RegisterBrokerRecord().setBrokerEpoch(100).setBrokerId(1); - brokerRecord.endPoints().add(new RegisterBrokerRecord.BrokerEndpoint(). + brokerRecord.endPoints().add(new BrokerEndpoint(). setSecurityProtocol(SecurityProtocol.PLAINTEXT.id). setPort((short) 9092). setName("PLAINTEXT"). @@ -86,7 +90,7 @@ public void testUnregister() throws Exception { setBrokerEpoch(100). setIncarnationId(Uuid.fromString("fPZv1VBsRFmnlRvmGcOW9w")). setRack("arack"); - brokerRecord.endPoints().add(new RegisterBrokerRecord.BrokerEndpoint(). + brokerRecord.endPoints().add(new BrokerEndpoint(). setSecurityProtocol(SecurityProtocol.PLAINTEXT.id). setPort((short) 9092). setName("PLAINTEXT"). @@ -122,7 +126,7 @@ public void testPlaceReplicas(int numUsableBrokers) throws Exception { for (int i = 0; i < numUsableBrokers; i++) { RegisterBrokerRecord brokerRecord = new RegisterBrokerRecord().setBrokerEpoch(100).setBrokerId(i); - brokerRecord.endPoints().add(new RegisterBrokerRecord.BrokerEndpoint(). + brokerRecord.endPoints().add(new BrokerEndpoint(). setSecurityProtocol(SecurityProtocol.PLAINTEXT.id). setPort((short) 9092). setName("PLAINTEXT"). @@ -147,4 +151,57 @@ public void testPlaceReplicas(int numUsableBrokers) throws Exception { } } } + + @Test + public void testIterator() throws Exception { + MockTime time = new MockTime(0, 0, 0); + SnapshotRegistry snapshotRegistry = new SnapshotRegistry(new LogContext()); + ClusterControlManager clusterControl = new ClusterControlManager( + new LogContext(), time, snapshotRegistry, 1000, + new SimpleReplicaPlacementPolicy(new Random())); + clusterControl.activate(); + assertFalse(clusterControl.unfenced(0)); + for (int i = 0; i < 3; i++) { + RegisterBrokerRecord brokerRecord = new RegisterBrokerRecord(). + setBrokerEpoch(100).setBrokerId(i).setRack(null); + brokerRecord.endPoints().add(new BrokerEndpoint(). + setSecurityProtocol(SecurityProtocol.PLAINTEXT.id). + setPort((short) 9092 + i). + setName("PLAINTEXT"). + setHost("example.com")); + clusterControl.replay(brokerRecord); + } + for (int i = 0; i < 2; i++) { + UnfenceBrokerRecord unfenceBrokerRecord = + new UnfenceBrokerRecord().setId(i).setEpoch(100); + clusterControl.replay(unfenceBrokerRecord); + } + ControllerTestUtils.assertBatchIteratorContains(Arrays.asList( + Arrays.asList(new ApiMessageAndVersion(new RegisterBrokerRecord(). + setBrokerEpoch(100).setBrokerId(0).setRack(null). + setEndPoints(new BrokerEndpointCollection(Collections.singleton( + new BrokerEndpoint().setSecurityProtocol(SecurityProtocol.PLAINTEXT.id). + setPort((short) 9092). + setName("PLAINTEXT"). + setHost("example.com")).iterator())), (short) 0), + new ApiMessageAndVersion(new UnfenceBrokerRecord(). + setId(0).setEpoch(100), (short) 0)), + Arrays.asList(new ApiMessageAndVersion(new RegisterBrokerRecord(). + setBrokerEpoch(100).setBrokerId(1).setRack(null). + setEndPoints(new BrokerEndpointCollection(Collections.singleton( + new BrokerEndpoint().setSecurityProtocol(SecurityProtocol.PLAINTEXT.id). + setPort((short) 9093). + setName("PLAINTEXT"). + setHost("example.com")).iterator())), (short) 0), + new ApiMessageAndVersion(new UnfenceBrokerRecord(). + setId(1).setEpoch(100), (short) 0)), + Arrays.asList(new ApiMessageAndVersion(new RegisterBrokerRecord(). + setBrokerEpoch(100).setBrokerId(2).setRack(null). + setEndPoints(new BrokerEndpointCollection(Collections.singleton( + new BrokerEndpoint().setSecurityProtocol(SecurityProtocol.PLAINTEXT.id). + setPort((short) 9094). + setName("PLAINTEXT"). + setHost("example.com")).iterator())), (short) 0))), + clusterControl.iterator(Long.MAX_VALUE)); + } } diff --git a/metadata/src/test/java/org/apache/kafka/controller/ConfigurationControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/ConfigurationControlManagerTest.java index 561a25b2d0ee2..608b428f68a85 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/ConfigurationControlManagerTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/ConfigurationControlManagerTest.java @@ -81,7 +81,7 @@ static Entry entry(A a, B b) { } @Test - public void testReplay() { + public void testReplay() throws Exception { SnapshotRegistry snapshotRegistry = new SnapshotRegistry(new LogContext()); ConfigurationControlManager manager = new ConfigurationControlManager(new LogContext(), snapshotRegistry, CONFIGS); @@ -103,6 +103,14 @@ public void testReplay() { setName("def").setValue("blah")); assertEquals(toMap(entry("abc", "x,y,z"), entry("def", "blah")), manager.getConfigs(MYTOPIC)); + ControllerTestUtils.assertBatchIteratorContains(Arrays.asList( + Arrays.asList(new ApiMessageAndVersion(new ConfigRecord(). + setResourceType(TOPIC.id()).setResourceName("mytopic"). + setName("abc").setValue("x,y,z"), (short) 0), + new ApiMessageAndVersion(new ConfigRecord(). + setResourceType(TOPIC.id()).setResourceName("mytopic"). + setName("def").setValue("blah"), (short) 0))), + manager.iterator(Long.MAX_VALUE)); } @Test diff --git a/metadata/src/test/java/org/apache/kafka/controller/ControllerTestUtils.java b/metadata/src/test/java/org/apache/kafka/controller/ControllerTestUtils.java index 746c7efb55836..10def7ecb352a 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/ControllerTestUtils.java +++ b/metadata/src/test/java/org/apache/kafka/controller/ControllerTestUtils.java @@ -18,14 +18,21 @@ package org.apache.kafka.controller; import org.apache.kafka.common.protocol.ApiMessage; +import org.apache.kafka.common.protocol.Message; +import org.apache.kafka.common.utils.ImplicitLinkedHashCollection; import org.apache.kafka.metadata.ApiMessageAndVersion; +import java.lang.reflect.Field; import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.Comparator; import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Set; +import static org.junit.jupiter.api.Assertions.assertEquals; + public class ControllerTestUtils { public static void replayAll(Object target, @@ -48,4 +55,48 @@ public static Set iteratorToSet(Iterator iterator) { } return set; } + + public static void assertBatchIteratorContains(List> batches, + Iterator> iterator) throws Exception { + List> actual = new ArrayList<>(); + while (iterator.hasNext()) { + actual.add(new ArrayList<>(iterator.next())); + } + deepSortRecords(actual); + List> expected = new ArrayList<>(); + for (List batch : batches) { + expected.add(new ArrayList<>(batch)); + } + deepSortRecords(expected); + assertEquals(expected, actual); + } + + @SuppressWarnings("unchecked") + public static void deepSortRecords(Object o) throws Exception { + if (o == null) { + return; + } else if (o instanceof List) { + List list = (List) o; + for (Object entry : list) { + if (entry != null) { + if (Number.class.isAssignableFrom(entry.getClass())) { + return; + } + deepSortRecords(entry); + } + } + list.sort(Comparator.comparing(Object::toString)); + } else if (o instanceof ImplicitLinkedHashCollection) { + ImplicitLinkedHashCollection coll = (ImplicitLinkedHashCollection) o; + for (Object entry : coll) { + deepSortRecords(entry); + } + coll.sort(Comparator.comparing(Object::toString)); + } else if (o instanceof Message || o instanceof ApiMessageAndVersion) { + for (Field field : o.getClass().getDeclaredFields()) { + field.setAccessible(true); + deepSortRecords(field.get(o)); + } + } + } } diff --git a/metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java index 0670984e52876..0ee5ee1eb3f26 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java @@ -18,6 +18,7 @@ package org.apache.kafka.controller; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -88,7 +89,7 @@ public void testReplay() { snapshotRegistry.createSnapshot(-1); FeatureControlManager manager = new FeatureControlManager( rangeMap("foo", 1, 2), snapshotRegistry); - manager.replay(record, 123); + manager.replay(record); snapshotRegistry.createSnapshot(123); assertEquals(new FeatureMapAndEpoch(new FeatureMap(rangeMap("foo", 1, 2)), 123), manager.finalizedFeatures(123)); @@ -121,7 +122,7 @@ public void testUpdateFeaturesErrorCases() { ControllerResult> result = manager.updateFeatures( rangeMap("foo", 1, 3), Collections.emptySet(), Collections.emptyMap()); assertEquals(Collections.singletonMap("foo", ApiError.NONE), result.response()); - manager.replay((FeatureLevelRecord) result.records().get(0).message(), 3); + manager.replay((FeatureLevelRecord) result.records().get(0).message()); snapshotRegistry.createSnapshot(3); assertEquals(ControllerResult.atomicOf(Collections.emptyList(), Collections. @@ -151,4 +152,25 @@ public void testUpdateFeaturesErrorCases() { ) ); } + + @Test + public void testFeatureControlIterator() throws Exception { + SnapshotRegistry snapshotRegistry = new SnapshotRegistry(new LogContext()); + FeatureControlManager manager = new FeatureControlManager( + rangeMap("foo", 1, 5, "bar", 1, 2), snapshotRegistry); + ControllerResult> result = manager. + updateFeatures(rangeMap("foo", 1, 5, "bar", 1, 1), + Collections.emptySet(), Collections.emptyMap()); + ControllerTestUtils.replayAll(manager, result.records()); + ControllerTestUtils.assertBatchIteratorContains(Arrays.asList( + Arrays.asList(new ApiMessageAndVersion(new FeatureLevelRecord(). + setName("foo"). + setMinFeatureLevel((short) 1). + setMaxFeatureLevel((short) 5), (short) 0)), + Arrays.asList(new ApiMessageAndVersion(new FeatureLevelRecord(). + setName("bar"). + setMinFeatureLevel((short) 1). + setMaxFeatureLevel((short) 1), (short) 0))), + manager.iterator(Long.MAX_VALUE)); + } } diff --git a/metadata/src/test/java/org/apache/kafka/controller/MockSnapshotReader.java b/metadata/src/test/java/org/apache/kafka/controller/MockSnapshotReader.java new file mode 100644 index 0000000000000..31ef31b927165 --- /dev/null +++ b/metadata/src/test/java/org/apache/kafka/controller/MockSnapshotReader.java @@ -0,0 +1,55 @@ +/* + * 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 org.apache.kafka.common.protocol.ApiMessage; + +import java.util.Iterator; +import java.util.List; + + +public class MockSnapshotReader implements SnapshotReader { + private final long epoch; + private final Iterator> iterator; + + public MockSnapshotReader(long epoch, + Iterator> iterator) { + this.epoch = epoch; + this.iterator = iterator; + } + + @Override + public long epoch() { + return epoch; + } + + @Override + public void close() { + // nothing to do + } + + @Override + public boolean hasNext() { + return iterator.hasNext(); + } + + @Override + public List next() { + return iterator.next(); + } +} diff --git a/metadata/src/test/java/org/apache/kafka/controller/MockSnapshotWriter.java b/metadata/src/test/java/org/apache/kafka/controller/MockSnapshotWriter.java new file mode 100644 index 0000000000000..76b4934482390 --- /dev/null +++ b/metadata/src/test/java/org/apache/kafka/controller/MockSnapshotWriter.java @@ -0,0 +1,94 @@ +/* + * 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 org.apache.kafka.common.protocol.ApiMessage; +import org.apache.kafka.metadata.ApiMessageAndVersion; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + + +class MockSnapshotWriter implements SnapshotWriter { + private final long epoch; + private boolean ready = true; + private boolean completed = false; + private boolean closed = false; + private final List> batches = new ArrayList<>(); + + public MockSnapshotWriter(long epoch) { + this.epoch = epoch; + } + + @Override + public long epoch() { + return epoch; + } + + @Override + public synchronized boolean writeBatch(List batch) throws IOException { + if (completed) throw new RuntimeException("writer has been completed"); + if (closed) throw new RuntimeException("writer is closed"); + if (!ready) return false; + batches.add(batch); + return true; + } + + synchronized void setReady(boolean ready) { + this.ready = ready; + } + + @Override + public synchronized void close() { + this.closed = true; + } + + @Override + public synchronized void completeSnapshot() throws IOException { + if (closed) throw new RuntimeException("writer is closed"); + this.notifyAll(); + this.completed = true; + } + + synchronized void waitForCompletion() throws InterruptedException { + while (!completed) { + this.wait(); + } + } + + synchronized boolean completed() { + return completed; + } + + synchronized List> batches() { + return batches; + } + + public MockSnapshotReader toReader() { + List> readerBatches = new ArrayList<>(); + for (List batch : batches) { + List readerBatch = new ArrayList<>(); + for (ApiMessageAndVersion messageAndVersion : batch) { + readerBatch.add(messageAndVersion.message()); + } + readerBatches.add(readerBatch); + } + return new MockSnapshotReader(epoch, readerBatches.iterator()); + } +} diff --git a/metadata/src/test/java/org/apache/kafka/controller/NoOpSnapshotWriter.java b/metadata/src/test/java/org/apache/kafka/controller/NoOpSnapshotWriter.java new file mode 100644 index 0000000000000..dd8da43f1aa91 --- /dev/null +++ b/metadata/src/test/java/org/apache/kafka/controller/NoOpSnapshotWriter.java @@ -0,0 +1,49 @@ +/* + * 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.io.IOException; +import java.util.List; +import org.apache.kafka.metadata.ApiMessageAndVersion; + + +public final class NoOpSnapshotWriter implements SnapshotWriter { + private final long epoch; + + public NoOpSnapshotWriter(long epoch) { + this.epoch = epoch; + } + + @Override + public long epoch() { + return epoch; + } + + @Override + public boolean writeBatch(List batch) throws IOException { + return true; + } + + @Override + public void close() { + } + + @Override + public void completeSnapshot() throws IOException { + } +} diff --git a/metadata/src/test/java/org/apache/kafka/controller/NoOpSnapshotWriterBuilder.java b/metadata/src/test/java/org/apache/kafka/controller/NoOpSnapshotWriterBuilder.java new file mode 100644 index 0000000000000..62d1f8da37673 --- /dev/null +++ b/metadata/src/test/java/org/apache/kafka/controller/NoOpSnapshotWriterBuilder.java @@ -0,0 +1,28 @@ +/* + * 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.function.Function; + + +public final class NoOpSnapshotWriterBuilder implements Function { + @Override + public SnapshotWriter apply(Long epoch) { + return new NoOpSnapshotWriter(epoch); + } +} diff --git a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java index e82081147ee94..d0588a0d27038 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java @@ -17,22 +17,38 @@ package org.apache.kafka.controller; +import java.util.Arrays; import java.util.Collections; +import java.util.HashMap; import java.util.Iterator; +import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.LinkedBlockingDeque; +import java.util.function.Function; + import org.apache.kafka.common.Uuid; import org.apache.kafka.common.config.ConfigResource; import org.apache.kafka.common.message.BrokerHeartbeatRequestData; import org.apache.kafka.common.message.BrokerRegistrationRequestData.Listener; import org.apache.kafka.common.message.BrokerRegistrationRequestData.ListenerCollection; import org.apache.kafka.common.message.BrokerRegistrationRequestData; +import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableReplicaAssignment; +import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableReplicaAssignmentCollection; import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic; import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicCollection; import org.apache.kafka.common.message.CreateTopicsRequestData; +import org.apache.kafka.common.message.CreateTopicsResponseData; +import org.apache.kafka.common.metadata.PartitionRecord; +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.TopicRecord; +import org.apache.kafka.common.metadata.UnfenceBrokerRecord; import org.apache.kafka.common.protocol.Errors; import org.apache.kafka.common.requests.ApiError; import org.apache.kafka.controller.BrokersToIsrs.TopicIdPartition; +import org.apache.kafka.metadata.ApiMessageAndVersion; import org.apache.kafka.metadata.BrokerHeartbeatReply; import org.apache.kafka.metadata.BrokerRegistrationReply; import org.apache.kafka.metalog.LocalLogManagerTestEnv; @@ -177,4 +193,134 @@ public void testUnregisterBroker() throws Throwable { } } } + + static class MockSnapshotWriterBuilder implements Function { + final LinkedBlockingDeque writers = new LinkedBlockingDeque<>(); + + @Override + public SnapshotWriter apply(Long epoch) { + MockSnapshotWriter writer = new MockSnapshotWriter(epoch); + writers.add(writer); + return writer; + } + } + + @Test + public void testSnapshotSaveAndLoad() throws Throwable { + MockSnapshotWriterBuilder snapshotWriterBuilder = new MockSnapshotWriterBuilder(); + final int numBrokers = 4; + MockSnapshotWriter writer = null; + Map brokerEpochs = new HashMap<>(); + Uuid fooId; + try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(3)) { + try (QuorumControllerTestEnv controlEnv = + new QuorumControllerTestEnv(logEnv, b -> b.setConfigDefs(CONFIGS). + setSnapshotWriterBuilder(snapshotWriterBuilder))) { + QuorumController active = controlEnv.activeController(); + for (int i = 0; i < numBrokers; i++) { + BrokerRegistrationReply reply = active.registerBroker( + new BrokerRegistrationRequestData(). + setBrokerId(i). + setRack(null). + setClusterId("06B-K3N1TBCNYFgruEVP0Q"). + setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwB" + i)). + setListeners(new ListenerCollection(Arrays.asList(new Listener(). + setName("PLAINTEXT").setHost("localhost"). + setPort(9092 + i)).iterator()))).get(); + brokerEpochs.put(i, reply.epoch()); + } + for (int i = 0; i < numBrokers - 1; i++) { + assertEquals(new BrokerHeartbeatReply(true, false, false, false), + active.processBrokerHeartbeat(new BrokerHeartbeatRequestData(). + setWantFence(false).setBrokerEpoch(brokerEpochs.get(i)). + setBrokerId(i).setCurrentMetadataOffset(100000L)).get()); + } + CreateTopicsResponseData fooData = active.createTopics( + new CreateTopicsRequestData().setTopics( + new CreatableTopicCollection(Collections.singleton( + new CreatableTopic().setName("foo").setNumPartitions(-1). + setReplicationFactor((short) -1). + setAssignments(new CreatableReplicaAssignmentCollection( + Arrays.asList(new CreatableReplicaAssignment(). + setPartitionIndex(0). + setBrokerIds(Arrays.asList(0, 1, 2)), + new CreatableReplicaAssignment(). + setPartitionIndex(1). + setBrokerIds(Arrays.asList(1, 2, 0))). + iterator()))).iterator()))).get(); + fooId = fooData.topics().find("foo").topicId(); + long snapshotEpoch = active.beginWritingSnapshot().get(); + writer = snapshotWriterBuilder.writers.takeFirst(); + assertEquals(snapshotEpoch, writer.epoch()); + writer.waitForCompletion(); + checkSnapshotContents(fooId, brokerEpochs, writer.batches().iterator()); + } + + final MockSnapshotReader reader = writer.toReader(); + try (QuorumControllerTestEnv controlEnv = + new QuorumControllerTestEnv(logEnv, b -> b.setConfigDefs(CONFIGS). + setSnapshotReader(reader). + setSnapshotWriterBuilder(snapshotWriterBuilder))) { + QuorumController active = controlEnv.activeController(); + long snapshotEpoch = active.beginWritingSnapshot().get(); + writer = snapshotWriterBuilder.writers.takeFirst(); + assertEquals(snapshotEpoch, writer.epoch()); + writer.waitForCompletion(); + checkSnapshotContents(fooId, brokerEpochs, writer.batches().iterator()); + } + } + } + + private void checkSnapshotContents(Uuid fooId, + Map brokerEpochs, + Iterator> iterator) throws Exception { + ControllerTestUtils.assertBatchIteratorContains(Arrays.asList( + Arrays.asList(new ApiMessageAndVersion(new TopicRecord(). + setName("foo").setTopicId(fooId), (short) 0), + new ApiMessageAndVersion(new PartitionRecord().setPartitionId(0). + setTopicId(fooId).setReplicas(Arrays.asList(0, 1, 2)). + setIsr(Arrays.asList(0, 1, 2)).setRemovingReplicas(null). + setAddingReplicas(null).setLeader(0).setLeaderEpoch(0). + setPartitionEpoch(0), (short) 0), + new ApiMessageAndVersion(new PartitionRecord().setPartitionId(1). + setTopicId(fooId).setReplicas(Arrays.asList(1, 2, 0)). + setIsr(Arrays.asList(1, 2, 0)).setRemovingReplicas(null). + setAddingReplicas(null).setLeader(1).setLeaderEpoch(0). + setPartitionEpoch(0), (short) 0)), + Arrays.asList(new ApiMessageAndVersion(new RegisterBrokerRecord(). + setBrokerId(0).setBrokerEpoch(brokerEpochs.get(0)). + setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwB0")). + setEndPoints(new BrokerEndpointCollection(Arrays.asList( + new BrokerEndpoint().setName("PLAINTEXT").setHost("localhost"). + setPort(9092).setSecurityProtocol((short) 0)).iterator())). + setRack(null), (short) 0), + new ApiMessageAndVersion(new UnfenceBrokerRecord(). + setId(0).setEpoch(brokerEpochs.get(0)), (short) 0)), + Arrays.asList(new ApiMessageAndVersion(new RegisterBrokerRecord(). + setBrokerId(1).setBrokerEpoch(brokerEpochs.get(1)). + setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwB1")). + setEndPoints(new BrokerEndpointCollection(Arrays.asList( + new BrokerEndpoint().setName("PLAINTEXT").setHost("localhost"). + setPort(9093).setSecurityProtocol((short) 0)).iterator())). + setRack(null), (short) 0), + new ApiMessageAndVersion(new UnfenceBrokerRecord(). + setId(1).setEpoch(brokerEpochs.get(1)), (short) 0)), + Arrays.asList(new ApiMessageAndVersion(new RegisterBrokerRecord(). + setBrokerId(2).setBrokerEpoch(brokerEpochs.get(2)). + setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwB2")). + setEndPoints(new BrokerEndpointCollection(Arrays.asList( + new BrokerEndpoint().setName("PLAINTEXT").setHost("localhost"). + setPort(9094).setSecurityProtocol((short) 0)).iterator())). + setRack(null), (short) 0), + new ApiMessageAndVersion(new UnfenceBrokerRecord(). + setId(2).setEpoch(brokerEpochs.get(2)), (short) 0)), + Arrays.asList(new ApiMessageAndVersion(new RegisterBrokerRecord(). + setBrokerId(3).setBrokerEpoch(brokerEpochs.get(3)). + setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwB3")). + setEndPoints(new BrokerEndpointCollection(Arrays.asList( + new BrokerEndpoint().setName("PLAINTEXT").setHost("localhost"). + setPort(9095).setSecurityProtocol((short) 0)).iterator())). + setRack(null), (short) 0))), + iterator); + } } diff --git a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java index fa7fc0094e748..6739944a02988 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java @@ -30,6 +30,7 @@ import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicCollection; import org.apache.kafka.common.message.CreateTopicsResponseData; import org.apache.kafka.common.message.CreateTopicsResponseData.CreatableTopicResult; +import org.apache.kafka.common.metadata.PartitionRecord; import org.apache.kafka.common.metadata.RegisterBrokerRecord; import org.apache.kafka.common.metadata.TopicRecord; import org.apache.kafka.common.protocol.Errors; @@ -162,6 +163,16 @@ public void testCreateTopics() throws Exception { setErrorCode(Errors.TOPIC_ALREADY_EXISTS.code()). setErrorMessage(Errors.TOPIC_ALREADY_EXISTS.exception().getMessage())); assertEquals(expectedResponse3, result3.response()); + Uuid fooId = result2.response().topics().find("foo").topicId(); + ControllerTestUtils.assertBatchIteratorContains(Arrays.asList( + Arrays.asList(new ApiMessageAndVersion(new PartitionRecord(). + setPartitionId(0).setTopicId(fooId). + setReplicas(Arrays.asList(2, 0, 1)).setIsr(Arrays.asList(2, 0, 1)). + setRemovingReplicas(null).setAddingReplicas(null).setLeader(2). + setLeaderEpoch(0).setPartitionEpoch(0), (short) 0), + new ApiMessageAndVersion(new TopicRecord(). + setTopicId(fooId).setName("foo"), (short) 0))), + ctx.replicationControl.iterator(Long.MAX_VALUE)); } @Test diff --git a/metadata/src/test/java/org/apache/kafka/controller/SnapshotGeneratorTest.java b/metadata/src/test/java/org/apache/kafka/controller/SnapshotGeneratorTest.java new file mode 100644 index 0000000000000..e8346211f1614 --- /dev/null +++ b/metadata/src/test/java/org/apache/kafka/controller/SnapshotGeneratorTest.java @@ -0,0 +1,110 @@ +/* + * 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 org.apache.kafka.common.Uuid; +import org.apache.kafka.common.config.ConfigResource; +import org.apache.kafka.common.metadata.ConfigRecord; +import org.apache.kafka.common.metadata.TopicRecord; +import org.apache.kafka.common.utils.ExponentialBackoff; +import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.controller.SnapshotGenerator.Section; +import org.apache.kafka.metadata.ApiMessageAndVersion; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import java.util.Arrays; +import java.util.List; +import java.util.OptionalLong; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + + +@Timeout(40) +public class SnapshotGeneratorTest { + private static final List> BATCHES; + + static { + BATCHES = Arrays.asList( + Arrays.asList(new ApiMessageAndVersion(new TopicRecord(). + setName("foo").setTopicId(Uuid.randomUuid()), (short) 0)), + Arrays.asList(new ApiMessageAndVersion(new TopicRecord(). + setName("bar").setTopicId(Uuid.randomUuid()), (short) 0)), + Arrays.asList(new ApiMessageAndVersion(new TopicRecord(). + setName("baz").setTopicId(Uuid.randomUuid()), (short) 0)), + Arrays.asList(new ApiMessageAndVersion(new ConfigRecord(). + setResourceName("foo").setResourceType(ConfigResource.Type.TOPIC.id()). + setName("retention.ms").setValue("10000000"), (short) 0), + new ApiMessageAndVersion(new ConfigRecord(). + setResourceName("foo").setResourceType(ConfigResource.Type.TOPIC.id()). + setName("max.message.bytes").setValue("100000000"), (short) 0)), + Arrays.asList(new ApiMessageAndVersion(new ConfigRecord(). + setResourceName("bar").setResourceType(ConfigResource.Type.TOPIC.id()). + setName("retention.ms").setValue("5000000"), (short) 0))); + } + + @Test + public void testGenerateBatches() throws Exception { + MockSnapshotWriter writer = new MockSnapshotWriter(123); + ExponentialBackoff exponentialBackoff = + new ExponentialBackoff(100, 2, 400, 0.0); + List
sections = Arrays.asList(new Section("replication", + Arrays.asList(BATCHES.get(0), BATCHES.get(1), BATCHES.get(2)).iterator()), + new Section("configuration", + Arrays.asList(BATCHES.get(3), BATCHES.get(4)).iterator())); + SnapshotGenerator generator = new SnapshotGenerator(new LogContext(), + writer, 2, exponentialBackoff, sections); + assertFalse(writer.completed()); + assertEquals(123L, generator.epoch()); + assertEquals(writer, generator.writer()); + assertEquals(OptionalLong.of(0L), generator.generateBatches()); + assertEquals(OptionalLong.of(0L), generator.generateBatches()); + assertFalse(writer.completed()); + assertEquals(OptionalLong.empty(), generator.generateBatches()); + assertTrue(writer.completed()); + } + + @Test + public void testGenerateBatchesWithBackoff() throws Exception { + MockSnapshotWriter writer = new MockSnapshotWriter(123); + ExponentialBackoff exponentialBackoff = + new ExponentialBackoff(100, 2, 400, 0.0); + List
sections = Arrays.asList(new Section("replication", + Arrays.asList(BATCHES.get(0), BATCHES.get(1), BATCHES.get(2)).iterator()), + new Section("configuration", + Arrays.asList(BATCHES.get(3), BATCHES.get(4)).iterator())); + SnapshotGenerator generator = new SnapshotGenerator(new LogContext(), + writer, 2, exponentialBackoff, sections); + assertEquals(123L, generator.epoch()); + assertEquals(writer, generator.writer()); + assertFalse(writer.completed()); + assertEquals(OptionalLong.of(0L), generator.generateBatches()); + writer.setReady(false); + assertEquals(OptionalLong.of(100L), generator.generateBatches()); + assertEquals(OptionalLong.of(200L), generator.generateBatches()); + assertEquals(OptionalLong.of(400L), generator.generateBatches()); + assertEquals(OptionalLong.of(400L), generator.generateBatches()); + writer.setReady(true); + assertFalse(writer.completed()); + assertEquals(OptionalLong.of(0L), generator.generateBatches()); + assertEquals(OptionalLong.empty(), generator.generateBatches()); + assertTrue(writer.completed()); + } +} \ No newline at end of file