From 6f77455f7aabcbb8b0f6eca30471dec399a2af76 Mon Sep 17 00:00:00 2001 From: David Arthur Date: Tue, 9 Feb 2021 13:53:36 -0500 Subject: [PATCH 1/5] Pull in files from kip-500 branch --- .../metadata/BrokerMetadataListener.scala | 260 ++++++++++++ .../metadata/LocalConfigRepository.scala | 93 +++++ .../metadata/LocalConfigRepositoryTest.scala | 56 +++ .../apache/kafka/metalog/LocalLogManager.java | 378 ++++++++++++++++++ .../apache/kafka/metalog/MetaLogLeader.java | 58 +++ .../apache/kafka/metalog/MetaLogListener.java | 55 +++ .../apache/kafka/metalog/MetaLogManager.java | 79 ++++ .../kafka/metalog/LocalLogManagerTest.java | 153 +++++++ .../kafka/metalog/LocalLogManagerTestEnv.java | 143 +++++++ .../metalog/MockMetaLogManagerListener.java | 77 ++++ 10 files changed, 1352 insertions(+) create mode 100644 core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala create mode 100644 core/src/main/scala/kafka/server/metadata/LocalConfigRepository.scala create mode 100644 core/src/test/scala/kafka/server/metadata/LocalConfigRepositoryTest.scala create mode 100644 metadata/src/main/java/org/apache/kafka/metalog/LocalLogManager.java create mode 100644 metadata/src/main/java/org/apache/kafka/metalog/MetaLogLeader.java create mode 100644 metadata/src/main/java/org/apache/kafka/metalog/MetaLogListener.java create mode 100644 metadata/src/main/java/org/apache/kafka/metalog/MetaLogManager.java create mode 100644 metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTest.java create mode 100644 metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTestEnv.java create mode 100644 metadata/src/test/java/org/apache/kafka/metalog/MockMetaLogManagerListener.java diff --git a/core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala b/core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala new file mode 100644 index 0000000000000..625ec2ba7f14f --- /dev/null +++ b/core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala @@ -0,0 +1,260 @@ +/** + * 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 kafka.server.metadata + +import java.util +import java.util.concurrent.TimeUnit +import kafka.coordinator.group.GroupCoordinator +import kafka.coordinator.transaction.TransactionCoordinator +import kafka.log.LogManager +import kafka.metrics.KafkaMetricsGroup +import kafka.server.{RaftReplicaManager, RequestHandlerHelper} +import org.apache.kafka.common.config.ConfigResource +import org.apache.kafka.common.metadata.MetadataRecordType._ +import org.apache.kafka.common.metadata._ +import org.apache.kafka.common.protocol.ApiMessage +import org.apache.kafka.common.utils.{LogContext, Time} +import org.apache.kafka.metalog.{MetaLogLeader, MetaLogListener} +import org.apache.kafka.queue.{EventQueue, KafkaEventQueue} + +import scala.jdk.CollectionConverters._ + +object BrokerMetadataListener{ + val MetadataBatchProcessingTimeUs = "MetadataBatchProcessingTimeUs" + val MetadataBatchSizes = "MetadataBatchSizes" +} + +class BrokerMetadataListener(val brokerId: Int, + val time: Time, + val metadataCache: RaftMetadataCache, + val configRepository: LocalConfigRepository, + val groupCoordinator: GroupCoordinator, + val replicaManager: RaftReplicaManager, + val txnCoordinator: TransactionCoordinator, + val logManager: LogManager, + val threadNamePrefix: Option[String] + ) extends MetaLogListener with KafkaMetricsGroup { + val logContext = new LogContext(s"[BrokerMetadataListener id=${brokerId}] ") + val log = logContext.logger(classOf[BrokerMetadataListener]) + logIdent = logContext.logPrefix() + + /** + * A histogram tracking the time in microseconds it took to process batches of events. + */ + private val batchProcessingTimeHist = newHistogram(BrokerMetadataListener.MetadataBatchProcessingTimeUs) + + /** + * A histogram tracking the sizes of batches that we have processed. + */ + private val metadataBatchSizeHist = newHistogram(BrokerMetadataListener.MetadataBatchSizes) + + /** + * The highest metadata offset that we've seen. Written only from the event queue thread. + */ + @volatile private var _highestMetadataOffset = -1L + + val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse("")) + + def highestMetadataOffset(): Long = _highestMetadataOffset + + /** + * Handle new metadata records. + */ + override def handleCommits(lastOffset: Long, records: util.List[ApiMessage]): Unit = { + eventQueue.append(new HandleCommitsEvent(lastOffset, records)) + } + + class HandleCommitsEvent(lastOffset: Long, + records: util.List[ApiMessage]) + extends EventQueue.FailureLoggingEvent(log) { + override def run(): Unit = { + if (isDebugEnabled) { + debug(s"Metadata batch ${lastOffset}: handling ${records.size()} record(s).") + } + val imageBuilder = + MetadataImageBuilder(brokerId, log, metadataCache.currentImage()) + val startNs = time.nanoseconds() + var index = 0 + metadataBatchSizeHist.update(records.size()) + records.iterator().asScala.foreach { case record => + try { + if (isTraceEnabled) { + trace("Metadata batch %d: processing [%d/%d]: %s.".format(lastOffset, index + 1, + records.size(), record.toString())) + } + handleMessage(imageBuilder, record, lastOffset) + } catch { + case e: Exception => error(s"Unable to handle record ${index} in batch " + + s"ending at offset ${lastOffset}", e) + } + index = index + 1 + } + if (imageBuilder.hasChanges) { + val newImage = imageBuilder.build() + if (isTraceEnabled) { + trace(s"Metadata batch ${lastOffset}: creating new metadata image ${newImage}") + } else if (isDebugEnabled) { + debug(s"Metadata batch ${lastOffset}: creating new metadata image") + } + metadataCache.image(newImage) + } else if (isDebugEnabled) { + debug(s"Metadata batch ${lastOffset}: no new metadata image required.") + } + if (imageBuilder.hasPartitionChanges) { + if (isDebugEnabled) { + debug(s"Metadata batch ${lastOffset}: applying partition changes") + } + replicaManager.handleMetadataRecords(imageBuilder, lastOffset, + RequestHandlerHelper.onLeadershipChange(groupCoordinator, txnCoordinator, _, _)) + } else if (isDebugEnabled) { + debug(s"Metadata batch ${lastOffset}: no partition changes found.") + } + _highestMetadataOffset = lastOffset + val endNs = time.nanoseconds() + val deltaUs = TimeUnit.MICROSECONDS.convert(endNs - startNs, TimeUnit.NANOSECONDS) + debug(s"Metadata batch ${lastOffset}: advanced highest metadata offset in ${deltaUs} " + + "microseconds.") + batchProcessingTimeHist.update(deltaUs) + } + } + + private def handleMessage(imageBuilder: MetadataImageBuilder, + record: ApiMessage, + lastOffset: Long): Unit = { + val recordType = try { + fromId(record.apiKey()) + } catch { + case e: Exception => throw new RuntimeException("Unknown metadata record type " + + s"${record.apiKey()} in batch ending at offset ${lastOffset}.") + } + recordType match { + case REGISTER_BROKER_RECORD => handleRegisterBrokerRecord(imageBuilder, + record.asInstanceOf[RegisterBrokerRecord]) + case UNREGISTER_BROKER_RECORD => handleUnregisterBrokerRecord(imageBuilder, + record.asInstanceOf[UnregisterBrokerRecord]) + case TOPIC_RECORD => handleTopicRecord(imageBuilder, + record.asInstanceOf[TopicRecord]) + case PARTITION_RECORD => handlePartitionRecord(imageBuilder, + record.asInstanceOf[PartitionRecord]) + case CONFIG_RECORD => handleConfigRecord(record.asInstanceOf[ConfigRecord]) + case ISR_CHANGE_RECORD => handleIsrChangeRecord(imageBuilder, + record.asInstanceOf[IsrChangeRecord]) + case FENCE_BROKER_RECORD => handleFenceBrokerRecord(imageBuilder, + record.asInstanceOf[FenceBrokerRecord]) + case UNFENCE_BROKER_RECORD => handleUnfenceBrokerRecord(imageBuilder, + record.asInstanceOf[UnfenceBrokerRecord]) + case REMOVE_TOPIC_RECORD => handleRemoveTopicRecord(imageBuilder, + record.asInstanceOf[RemoveTopicRecord]) + // TODO: handle FEATURE_LEVEL_RECORD + case _ => throw new RuntimeException(s"Unsupported record type ${recordType}") + } + } + + def handleRegisterBrokerRecord(imageBuilder: MetadataImageBuilder, + record: RegisterBrokerRecord): Unit = { + val broker = MetadataBroker(record) + imageBuilder.brokersBuilder().add(broker) + } + + def handleUnregisterBrokerRecord(imageBuilder: MetadataImageBuilder, + record: UnregisterBrokerRecord): Unit = { + imageBuilder.brokersBuilder().remove(record.brokerId()) + } + + def handleTopicRecord(imageBuilder: MetadataImageBuilder, + record: TopicRecord): Unit = { + imageBuilder.partitionsBuilder().addUuidMapping(record.name(), record.topicId()) + } + + def handlePartitionRecord(imageBuilder: MetadataImageBuilder, + record: PartitionRecord): Unit = { + imageBuilder.topicIdToName(record.topicId()) match { + case None => throw new RuntimeException(s"Unable to locate topic with ID ${record.topicId}") + case Some(name) => + val partition = MetadataPartition(name, record) + imageBuilder.partitionsBuilder().set(partition) + } + } + + def handleConfigRecord(record: ConfigRecord): Unit = { + val t = ConfigResource.Type.forId(record.resourceType()) + if (t == ConfigResource.Type.UNKNOWN) { + throw new RuntimeException("Unable to understand config resource type " + + s"${Integer.valueOf(record.resourceType())}") + } + val resource = new ConfigResource(t, record.resourceName()) + configRepository.setConfig(resource, record.name(), record.value()) + } + + def handleIsrChangeRecord(imageBuilder: MetadataImageBuilder, + record: IsrChangeRecord): Unit = { + imageBuilder.partitionsBuilder().handleIsrChange(record) + } + + def handleFenceBrokerRecord(imageBuilder: MetadataImageBuilder, + record: FenceBrokerRecord): Unit = { + // TODO: add broker epoch to metadata cache, and check it here. + imageBuilder.brokersBuilder().changeFencing(record.id(), true) + } + + def handleUnfenceBrokerRecord(imageBuilder: MetadataImageBuilder, + record: UnfenceBrokerRecord): Unit = { + // TODO: add broker epoch to metadata cache, and check it here. + imageBuilder.brokersBuilder().changeFencing(record.id(), false) + } + + def handleRemoveTopicRecord(imageBuilder: MetadataImageBuilder, + record: RemoveTopicRecord): Unit = { + val removedPartitions = imageBuilder.partitionsBuilder(). + removeTopicById(record.topicId()) + groupCoordinator.handleDeletedPartitions(removedPartitions.map(_.toTopicPartition).toSeq) + } + + class HandleNewLeaderEvent(leader: MetaLogLeader) + extends EventQueue.FailureLoggingEvent(log) { + override def run(): Unit = { + val imageBuilder = + MetadataImageBuilder(brokerId, log, metadataCache.currentImage()) + if (leader.nodeId() < 0) { + imageBuilder.controllerId(None) + } else { + imageBuilder.controllerId(Some(leader.nodeId())) + } + metadataCache.image(imageBuilder.build()) + } + } + + override def handleNewLeader(leader: MetaLogLeader): Unit = { + eventQueue.append(new HandleNewLeaderEvent(leader)) + } + + class ShutdownEvent() extends EventQueue.FailureLoggingEvent(log) { + override def run(): Unit = { + removeMetric(BrokerMetadataListener.MetadataBatchProcessingTimeUs) + removeMetric(BrokerMetadataListener.MetadataBatchSizes) + } + } + + override def beginShutdown(): Unit = { + eventQueue.beginShutdown("beginShutdown", new ShutdownEvent()) + } + + def close(): Unit = { + beginShutdown() + eventQueue.close() + } +} diff --git a/core/src/main/scala/kafka/server/metadata/LocalConfigRepository.scala b/core/src/main/scala/kafka/server/metadata/LocalConfigRepository.scala new file mode 100644 index 0000000000000..a46993ba777cd --- /dev/null +++ b/core/src/main/scala/kafka/server/metadata/LocalConfigRepository.scala @@ -0,0 +1,93 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package kafka.server.metadata + +import java.util +import java.util.Properties +import java.util.concurrent.ConcurrentHashMap +import java.util.function.BiFunction + +import org.apache.kafka.common.config.ConfigResource +import org.apache.kafka.common.config.ConfigResource.Type + +import scala.jdk.CollectionConverters._ + +/** + * A ConfigRepository that stores configurations locally. + */ +class LocalConfigRepository extends ConfigRepository { + val configMap = new ConcurrentHashMap[ConfigResource, util.HashMap[String, String]] + + def setTopicConfig(topic: String, key: String, value: String): Unit = { + setConfig(new ConfigResource(Type.TOPIC, topic), key, value) + } + + def setBrokerConfig(id: Int, key: String, value: String): Unit = { + setConfig(new ConfigResource(Type.BROKER, id.toString()), key, value) + } + + def setConfig(configResource: ConfigResource, key: String, value: String): Unit = { + configMap.compute(configResource, new BiFunction[ConfigResource, util.HashMap[String, String], util.HashMap[String, String]] { + override def apply(resource: ConfigResource, + curConfigs: util.HashMap[String, String]): util.HashMap[String, String] = { + if (value == null) { + if (curConfigs == null) { + null + } else { + val newConfigs = new util.HashMap[String, String](curConfigs) + newConfigs.remove(key) + if (newConfigs.isEmpty) { + null + } else { + newConfigs + } + } + } else { + if (curConfigs == null) { + val newConfigs = new util.HashMap[String, String](1) + newConfigs.put(key, value) + newConfigs + } else { + val newConfigs = new util.HashMap[String, String](curConfigs.size() + 1) + newConfigs.putAll(curConfigs) + newConfigs.put(key, value) + newConfigs + } + } + } + }) + } + + def config(configResource: ConfigResource): Properties = { + val properties = new Properties() + Option(configMap.get(configResource)).foreach { + _.entrySet().iterator().asScala.foreach { case e => + properties.put(e.getKey, e.getValue) + } + } + properties + } + + override def topicConfig(topic: String): Properties = { + config(new ConfigResource(Type.TOPIC, topic)) + } + + override def brokerConfig(id: Int): Properties = { + config(new ConfigResource(Type.BROKER, id.toString())) + } +} diff --git a/core/src/test/scala/kafka/server/metadata/LocalConfigRepositoryTest.scala b/core/src/test/scala/kafka/server/metadata/LocalConfigRepositoryTest.scala new file mode 100644 index 0000000000000..dd651ebbf4175 --- /dev/null +++ b/core/src/test/scala/kafka/server/metadata/LocalConfigRepositoryTest.scala @@ -0,0 +1,56 @@ +/** + * 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 kafka.server.metadata + +import org.junit.jupiter.api.Assertions._ +import org.junit.jupiter.api.{Test, Timeout} + +import java.util.Properties +import java.util.concurrent.TimeUnit + + +@Timeout(value = 120000, unit = TimeUnit.MILLISECONDS) +class LocalConfigRepositoryTest { + @Test + def testEmptyRepository(): Unit = { + val repository = new LocalConfigRepository() + assertEquals(new Properties(), repository.brokerConfig(0)) + assertEquals(new Properties(), repository.topicConfig("foo")) + } + + @Test + def testSetConfig(): Unit = { + val repository = new LocalConfigRepository() + repository.setBrokerConfig(0, "foo", null) + assertEquals(new Properties(), repository.brokerConfig(0)) + + repository.setBrokerConfig(1, "foo", "bar") + val brokerProperties = new Properties() + brokerProperties.put("foo", "bar") + assertEquals(brokerProperties, repository.brokerConfig(1)) + + val brokerProperties2 = new Properties() + brokerProperties2.put("foo", "bar") + brokerProperties2.put("foo2", "baz") + repository.setBrokerConfig(1, "foo2", "baz") + assertEquals(brokerProperties2, repository.brokerConfig(1)) + + repository.setBrokerConfig(1, "foo2", null) + assertEquals(brokerProperties, repository.brokerConfig(1)) + } +} diff --git a/metadata/src/main/java/org/apache/kafka/metalog/LocalLogManager.java b/metadata/src/main/java/org/apache/kafka/metalog/LocalLogManager.java new file mode 100644 index 0000000000000..ef85314e0ef2f --- /dev/null +++ b/metadata/src/main/java/org/apache/kafka/metalog/LocalLogManager.java @@ -0,0 +1,378 @@ +/* + * 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.metalog; + +import org.apache.kafka.common.protocol.ApiMessage; +import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.common.utils.Time; +import org.apache.kafka.metadata.ApiMessageAndVersion; +import org.apache.kafka.queue.EventQueue; +import org.apache.kafka.queue.KafkaEventQueue; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.AbstractMap.SimpleImmutableEntry; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map.Entry; +import java.util.Objects; +import java.util.TreeMap; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ThreadLocalRandom; +import java.util.stream.Collectors; + +/** + * The LocalLogManager is a test implementation that relies on the contents of memory. + */ +public final class LocalLogManager implements MetaLogManager, AutoCloseable { + interface LocalBatch { + int size(); + } + + static class LeaderChangeBatch implements LocalBatch { + private final MetaLogLeader newLeader; + + LeaderChangeBatch(MetaLogLeader newLeader) { + this.newLeader = newLeader; + } + + @Override + public int size() { + return 1; + } + + @Override + public boolean equals(Object o) { + if (!(o instanceof LeaderChangeBatch)) return false; + LeaderChangeBatch other = (LeaderChangeBatch) o; + if (!other.newLeader.equals(newLeader)) return false; + return true; + } + + @Override + public int hashCode() { + return Objects.hash(newLeader); + } + + @Override + public String toString() { + return "LeaderChangeBatch(newLeader=" + newLeader + ")"; + } + } + + static class LocalRecordBatch implements LocalBatch { + private final List records; + + LocalRecordBatch(List records) { + this.records = records; + } + + @Override + public int size() { + return records.size(); + } + + @Override + public boolean equals(Object o) { + if (!(o instanceof LocalRecordBatch)) return false; + LocalRecordBatch other = (LocalRecordBatch) o; + if (!other.records.equals(records)) return false; + return true; + } + + @Override + public int hashCode() { + return Objects.hash(records); + } + + @Override + public String toString() { + return "LocalRecordBatch(records=" + records + ")"; + } + } + + public static class SharedLogData { + private final Logger log = LoggerFactory.getLogger(SharedLogData.class); + private final HashMap logManagers = new HashMap<>(); + private final TreeMap batches = new TreeMap<>(); + private MetaLogLeader leader = new MetaLogLeader(-1, -1); + private long prevOffset = -1; + + synchronized void registerLogManager(LocalLogManager logManager) { + if (logManagers.put(logManager.nodeId(), logManager) != null) { + throw new RuntimeException("Can't have multiple LocalLogManagers " + + "with id " + logManager.nodeId()); + } + electLeaderIfNeeded(); + } + + synchronized void unregisterLogManager(LocalLogManager logManager) { + if (!logManagers.remove(logManager.nodeId(), logManager)) { + throw new RuntimeException("Log manager " + logManager.nodeId() + + " was not found."); + } + } + + synchronized long tryAppend(int nodeId, long epoch, LocalBatch batch) { + if (epoch != leader.epoch()) { + log.trace("tryAppend(nodeId={}, epoch={}): the provided epoch does not " + + "match the current leader epoch of {}.", nodeId, epoch, leader.epoch()); + return Long.MAX_VALUE; + } + if (nodeId != leader.nodeId()) { + log.trace("tryAppend(nodeId={}, epoch={}): the given node id does not " + + "match the current leader id of {}.", nodeId, leader.nodeId()); + return Long.MAX_VALUE; + } + log.trace("tryAppend(nodeId={}): appending {}.", nodeId, batch); + long offset = append(batch); + electLeaderIfNeeded(); + return offset; + } + + synchronized long append(LocalBatch batch) { + prevOffset += batch.size(); + log.debug("append(batch={}, prevOffset={})", batch, prevOffset); + batches.put(prevOffset, batch); + if (batch instanceof LeaderChangeBatch) { + LeaderChangeBatch leaderChangeBatch = (LeaderChangeBatch) batch; + leader = leaderChangeBatch.newLeader; + } + for (LocalLogManager logManager : logManagers.values()) { + logManager.scheduleLogCheck(); + } + return prevOffset; + } + + synchronized void electLeaderIfNeeded() { + if (leader.nodeId() != -1 || logManagers.isEmpty()) { + return; + } + int nextLeaderIndex = ThreadLocalRandom.current().nextInt(logManagers.size()); + Iterator iter = logManagers.keySet().iterator(); + Integer nextLeaderNode = null; + for (int i = 0; i <= nextLeaderIndex; i++) { + nextLeaderNode = iter.next(); + } + MetaLogLeader newLeader = new MetaLogLeader(nextLeaderNode, leader.epoch() + 1); + log.info("Elected new leader: {}.", newLeader); + append(new LeaderChangeBatch(newLeader)); + } + + synchronized Entry nextBatch(long offset) { + Entry entry = batches.higherEntry(offset); + if (entry == null) { + return null; + } + return new SimpleImmutableEntry<>(entry.getKey(), entry.getValue()); + } + } + + private static class MetaLogListenerData { + private long offset = -1; + private final MetaLogListener listener; + + MetaLogListenerData(MetaLogListener listener) { + this.listener = listener; + } + } + + private final Logger log; + + private final int nodeId; + + private final SharedLogData shared; + + private final EventQueue eventQueue; + + private boolean initialized = false; + + private boolean shutdown = false; + + private long maxReadOffset = Long.MAX_VALUE; + + private final List listeners = new ArrayList<>(); + + private volatile MetaLogLeader leader = new MetaLogLeader(-1, -1); + + public LocalLogManager(LogContext logContext, + int nodeId, + SharedLogData shared, + String threadNamePrefix) { + this.log = logContext.logger(LocalLogManager.class); + this.nodeId = nodeId; + this.shared = shared; + this.eventQueue = new KafkaEventQueue(Time.SYSTEM, logContext, threadNamePrefix); + shared.registerLogManager(this); + } + + private void scheduleLogCheck() { + eventQueue.append(() -> { + try { + log.debug("Node {}: running log check.", nodeId); + int numEntriesFound = 0; + for (MetaLogListenerData listenerData : listeners) { + while (true) { + Entry entry = shared.nextBatch(listenerData.offset); + if (entry == null) { + log.trace("Node {}: reached the end of the log after finding " + + "{} entries.", nodeId, numEntriesFound); + break; + } + long entryOffset = entry.getKey(); + if (entryOffset > maxReadOffset) { + log.trace("Node {}: after {} entries, not reading the next " + + "entry because its offset is {}, and maxReadOffset is {}.", + nodeId, numEntriesFound, entryOffset, maxReadOffset); + break; + } + if (entry.getValue() instanceof LeaderChangeBatch) { + LeaderChangeBatch batch = (LeaderChangeBatch) entry.getValue(); + log.trace("Node {}: handling LeaderChange to {}.", + nodeId, batch.newLeader); + listenerData.listener.handleNewLeader(batch.newLeader); + if (batch.newLeader.epoch() > leader.epoch()) { + leader = batch.newLeader; + } + } else if (entry.getValue() instanceof LocalRecordBatch) { + LocalRecordBatch batch = (LocalRecordBatch) entry.getValue(); + log.trace("Node {}: handling LocalRecordBatch with offset {}.", + nodeId, entryOffset); + listenerData.listener.handleCommits(entryOffset, batch.records); + } + numEntriesFound++; + listenerData.offset = entryOffset; + } + } + log.trace("Completed log check for node " + nodeId); + } catch (Exception e) { + log.error("Exception while handling log check", e); + } + }); + } + + public void beginShutdown() { + eventQueue.beginShutdown("beginShutdown", () -> { + try { + if (initialized && !shutdown) { + log.debug("Node {}: beginning shutdown.", nodeId); + renounce(leader.epoch()); + for (MetaLogListenerData listenerData : listeners) { + listenerData.listener.beginShutdown(); + } + shared.unregisterLogManager(this); + } + } catch (Exception e) { + log.error("Unexpected exception while sending beginShutdown callbacks", e); + } + shutdown = true; + }); + } + + @Override + public void close() throws InterruptedException { + log.debug("Node {}: closing.", nodeId); + beginShutdown(); + eventQueue.close(); + } + + @Override + public void initialize() throws Exception { + eventQueue.append(() -> { + log.debug("initialized local log manager for node " + nodeId); + initialized = true; + }); + } + + @Override + public void register(MetaLogListener listener) throws Exception { + CompletableFuture future = new CompletableFuture<>(); + eventQueue.append(() -> { + if (shutdown) { + log.info("Node {}: can't register because local log manager has " + + "already been shut down.", nodeId); + future.complete(null); + } else if (initialized) { + log.info("Node {}: registered MetaLogListener.", nodeId); + listeners.add(new MetaLogListenerData(listener)); + shared.electLeaderIfNeeded(); + scheduleLogCheck(); + future.complete(null); + } else { + log.info("Node {}: can't register because local log manager has not " + + "been initialized.", nodeId); + future.completeExceptionally(new RuntimeException( + "LocalLogManager was not initialized.")); + } + }); + future.get(); + } + + @Override + public long scheduleWrite(long epoch, List batch) { + return shared.tryAppend(nodeId, leader.epoch(), new LocalRecordBatch( + batch.stream().map(r -> r.message()).collect(Collectors.toList()))); + } + + @Override + public void renounce(long epoch) { + MetaLogLeader curLeader = leader; + MetaLogLeader nextLeader = new MetaLogLeader(-1, curLeader.epoch() + 1); + shared.tryAppend(nodeId, curLeader.epoch(), new LeaderChangeBatch(nextLeader)); + } + + @Override + public MetaLogLeader leader() { + return leader; + } + + @Override + public int nodeId() { + return nodeId; + } + + public List listeners() { + final CompletableFuture> future = new CompletableFuture<>(); + eventQueue.append(() -> { + future.complete(listeners.stream().map(l -> l.listener).collect(Collectors.toList())); + }); + try { + return future.get(); + } catch (ExecutionException | InterruptedException e) { + throw new RuntimeException(e); + } + } + + public void setMaxReadOffset(long maxReadOffset) { + CompletableFuture future = new CompletableFuture<>(); + eventQueue.append(() -> { + log.trace("Node {}: set maxReadOffset to {}.", nodeId, maxReadOffset); + this.maxReadOffset = maxReadOffset; + scheduleLogCheck(); + future.complete(null); + }); + try { + future.get(); + } catch (ExecutionException | InterruptedException e) { + throw new RuntimeException(e); + } + } +} diff --git a/metadata/src/main/java/org/apache/kafka/metalog/MetaLogLeader.java b/metadata/src/main/java/org/apache/kafka/metalog/MetaLogLeader.java new file mode 100644 index 0000000000000..2bf4f7c718bd5 --- /dev/null +++ b/metadata/src/main/java/org/apache/kafka/metalog/MetaLogLeader.java @@ -0,0 +1,58 @@ +/* + * 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.metalog; + +import java.util.Objects; + +/** + * The current leader of the MetaLog. + */ +public class MetaLogLeader { + private final int nodeId; + private final long epoch; + + public MetaLogLeader(int nodeId, long epoch) { + this.nodeId = nodeId; + this.epoch = epoch; + } + + public int nodeId() { + return nodeId; + } + + public long epoch() { + return epoch; + } + + @Override + public boolean equals(Object o) { + if (!(o instanceof MetaLogLeader)) return false; + MetaLogLeader other = (MetaLogLeader) o; + return other.nodeId == nodeId && other.epoch == epoch; + } + + @Override + public int hashCode() { + return Objects.hash(nodeId, epoch); + } + + @Override + public String toString() { + return "MetaLogLeader(nodeId=" + nodeId + ", epoch=" + epoch + ")"; + } +} diff --git a/metadata/src/main/java/org/apache/kafka/metalog/MetaLogListener.java b/metadata/src/main/java/org/apache/kafka/metalog/MetaLogListener.java new file mode 100644 index 0000000000000..93744202dc90f --- /dev/null +++ b/metadata/src/main/java/org/apache/kafka/metalog/MetaLogListener.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.metalog; + +import org.apache.kafka.common.protocol.ApiMessage; + +import java.util.List; + +/** + * Listeners receive notifications from the MetaLogManager. + */ +public interface MetaLogListener { + /** + * Called when the MetaLogManager commits some messages. + * + * @param lastOffset The last offset found in all the given messages. + * @param messages The messages. + */ + void handleCommits(long lastOffset, List messages); + + /** + * Called when a new leader is elected. + * + * @param leader The new leader id and epoch. + */ + default void handleNewLeader(MetaLogLeader leader) {} + + /** + * Called when the MetaLogManager has renounced the leadership. + * + * @param epoch The controller epoch that has ended. + */ + default void handleRenounce(long epoch) {} + + /** + * Called when the MetaLogManager has finished shutting down, and wants to tell its + * listener that it is safe to shut down as well. + */ + default void beginShutdown() {} +} diff --git a/metadata/src/main/java/org/apache/kafka/metalog/MetaLogManager.java b/metadata/src/main/java/org/apache/kafka/metalog/MetaLogManager.java new file mode 100644 index 0000000000000..94b35ea72bf64 --- /dev/null +++ b/metadata/src/main/java/org/apache/kafka/metalog/MetaLogManager.java @@ -0,0 +1,79 @@ +/* + * 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.metalog; + +import org.apache.kafka.metadata.ApiMessageAndVersion; + +import java.util.List; + +/** + * The MetaLogManager handles storing metadata and electing leaders. + */ +public interface MetaLogManager { + + /** + * Start this meta log manager. + * The manager must be ready to accept incoming calls after this function returns. + * It is an error to initialize a MetaLogManager more than once. + */ + void initialize() throws Exception; + + /** + * Register the listener. The manager must be initialized already. + * The listener must be ready to accept incoming calls immediately. + * + * @param listener The listener to register. + */ + void register(MetaLogListener listener) throws Exception; + + /** + * Schedule a write to the log. + * + * The write will be scheduled to happen at some time in the future. There is no + * error return or exception thrown if the write fails. Instead, the listener may + * regard the write as successful if and only if the MetaLogManager reaches the given + * index before renouncing its leadership. The listener should determine this by + * monitoring the committed indexes. + * + * @param epoch The controller epoch. + * @param batch The batch of messages to write. + * + * @return The index of the message. + */ + long scheduleWrite(long epoch, List batch); + + /** + * Renounce the leadership. + * + * @param epoch The epoch. If this does not match the current epoch, this + * call will be ignored. + */ + void renounce(long epoch); + + /** + * Returns the current leader. The active node may change immediately after this + * function is called, of course. + */ + MetaLogLeader leader(); + + /** + * Returns the node id. + */ + int nodeId(); + +} diff --git a/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTest.java b/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTest.java new file mode 100644 index 0000000000000..9d4eb8b594e29 --- /dev/null +++ b/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTest.java @@ -0,0 +1,153 @@ +/* + * 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.metalog; + +import org.apache.kafka.common.metadata.RegisterBrokerRecord; +import org.apache.kafka.metadata.ApiMessageAndVersion; +import org.apache.kafka.test.TestUtils; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Arrays; +import java.util.List; +import java.util.stream.Collectors; + +import static org.apache.kafka.metalog.MockMetaLogManagerListener.COMMIT; +import static org.apache.kafka.metalog.MockMetaLogManagerListener.LAST_COMMITTED_OFFSET; +import static org.apache.kafka.metalog.MockMetaLogManagerListener.SHUTDOWN; +import static org.junit.jupiter.api.Assertions.assertEquals; + +@Timeout(value = 40) +public class LocalLogManagerTest { + private static final Logger log = LoggerFactory.getLogger(LocalLogManagerTest.class); + + /** + * Test creating a LocalLogManager and closing it. + */ + @Test + public void testCreateAndClose() throws Exception { + try (LocalLogManagerTestEnv env = + LocalLogManagerTestEnv.createWithMockListeners(1)) { + env.close(); + assertEquals(null, env.firstError.get()); + } + } + + /** + * Test that the local log maanger will claim leadership. + */ + @Test + public void testClaimsLeadership() throws Exception { + try (LocalLogManagerTestEnv env = + LocalLogManagerTestEnv.createWithMockListeners(1)) { + assertEquals(new MetaLogLeader(0, 0), env.waitForLeader()); + env.close(); + assertEquals(null, env.firstError.get()); + } + } + + /** + * Test that we can pass leadership back and forth between log managers. + */ + @Test + public void testPassLeadership() throws Exception { + try (LocalLogManagerTestEnv env = + LocalLogManagerTestEnv.createWithMockListeners(3)) { + MetaLogLeader first = env.waitForLeader(); + MetaLogLeader cur = first; + do { + env.logManagers().get(cur.nodeId()).renounce(cur.epoch()); + MetaLogLeader next = env.waitForLeader(); + while (next.epoch() == cur.epoch()) { + Thread.sleep(1); + next = env.waitForLeader(); + } + long expectedNextEpoch = cur.epoch() + 2; + assertEquals(expectedNextEpoch, next.epoch(), "Expected next epoch to be " + expectedNextEpoch + + ", but found " + next); + cur = next; + } while (cur.nodeId() == first.nodeId()); + env.close(); + assertEquals(null, env.firstError.get()); + } + } + + private static void waitForLastCommittedOffset(long targetOffset, + LocalLogManager logManager) throws InterruptedException { + TestUtils.retryOnExceptionWithTimeout(3, 20000, () -> { + MockMetaLogManagerListener listener = + (MockMetaLogManagerListener) logManager.listeners().get(0); + long highestOffset = -1; + for (String event : listener.serializedEvents()) { + if (event.startsWith(LAST_COMMITTED_OFFSET)) { + long offset = Long.valueOf( + event.substring(LAST_COMMITTED_OFFSET.length() + 1)); + if (offset < highestOffset) { + throw new RuntimeException("Invalid offset: " + offset + + " is less than the previous offset of " + highestOffset); + } + highestOffset = offset; + } + } + if (highestOffset < targetOffset) { + throw new RuntimeException("Offset for log manager " + + logManager.nodeId() + " only reached " + highestOffset); + } + }); + } + + /** + * Test that all the log managers see all the commits. + */ + @Test + public void testCommits() throws Exception { + try (LocalLogManagerTestEnv env = + LocalLogManagerTestEnv.createWithMockListeners(3)) { + MetaLogLeader leaderInfo = env.waitForLeader(); + LocalLogManager activeLogManager = env.logManagers().get(leaderInfo.nodeId()); + long epoch = activeLogManager.leader().epoch(); + List messages = Arrays.asList( + new ApiMessageAndVersion(new RegisterBrokerRecord().setBrokerId(0), (short) 0), + new ApiMessageAndVersion(new RegisterBrokerRecord().setBrokerId(1), (short) 0), + new ApiMessageAndVersion(new RegisterBrokerRecord().setBrokerId(2), (short) 0)); + assertEquals(3, activeLogManager.scheduleWrite(epoch, messages)); + for (LocalLogManager logManager : env.logManagers()) { + waitForLastCommittedOffset(3, logManager); + } + List listeners = env.logManagers().stream(). + map(m -> (MockMetaLogManagerListener) m.listeners().get(0)). + collect(Collectors.toList()); + env.close(); + for (MockMetaLogManagerListener listener : listeners) { + List events = listener.serializedEvents(); + assertEquals(SHUTDOWN, events.get(events.size() - 1)); + int foundIndex = 0; + for (String event : events) { + if (event.startsWith(COMMIT)) { + assertEquals(messages.get(foundIndex).message().toString(), + event.substring(COMMIT.length() + 1)); + foundIndex++; + } + } + assertEquals(messages.size(), foundIndex); + } + } + } +} diff --git a/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTestEnv.java b/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTestEnv.java new file mode 100644 index 0000000000000..52aeea052bdde --- /dev/null +++ b/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTestEnv.java @@ -0,0 +1,143 @@ +/* + * 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.metalog; + +import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.common.utils.Utils; +import org.apache.kafka.metalog.LocalLogManager.SharedLogData; +import org.apache.kafka.test.TestUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicReference; + +public class LocalLogManagerTestEnv implements AutoCloseable { + private static final Logger log = + LoggerFactory.getLogger(LocalLogManagerTestEnv.class); + + /** + * The first error we encountered during this test, or the empty string if we have + * not encountered any. + */ + final AtomicReference firstError = new AtomicReference<>(null); + + /** + * The test directory, which we will delete once the test is over. + */ + private final File dir; + + /** + * The shared data for our LocalLogManager instances. + */ + private final SharedLogData shared; + + /** + * A list of log managers. + */ + private final List logManagers; + + public static LocalLogManagerTestEnv createWithMockListeners(int numManagers) throws Exception { + LocalLogManagerTestEnv testEnv = new LocalLogManagerTestEnv(numManagers); + try { + for (LocalLogManager logManager : testEnv.logManagers) { + logManager.register(new MockMetaLogManagerListener()); + } + } catch (Exception e) { + testEnv.close(); + throw e; + } + return testEnv; + } + + public LocalLogManagerTestEnv(int numManagers) throws Exception { + dir = TestUtils.tempDirectory(); + shared = new SharedLogData(); + List newLogManagers = new ArrayList<>(numManagers); + try { + for (int nodeId = 0; nodeId < numManagers; nodeId++) { + newLogManagers.add(new LocalLogManager( + new LogContext(String.format("[LocalLogManager %d] ", nodeId)), + nodeId, + shared, + String.format("LocalLogManager-%d_", nodeId))); + } + for (LocalLogManager logManager : newLogManagers) { + logManager.initialize(); + } + } catch (Throwable t) { + for (LocalLogManager logManager : newLogManagers) { + logManager.close(); + } + throw t; + } + this.logManagers = newLogManagers; + } + + AtomicReference firstError() { + return firstError; + } + + File dir() { + return dir; + } + + MetaLogLeader waitForLeader() throws InterruptedException { + AtomicReference value = new AtomicReference<>(null); + TestUtils.retryOnExceptionWithTimeout(3, 20000, () -> { + MetaLogLeader result = null; + for (LocalLogManager logManager : logManagers) { + MetaLogLeader leader = logManager.leader(); + if (leader.nodeId() == logManager.nodeId()) { + if (result != null) { + throw new RuntimeException("node " + leader.nodeId() + + " thinks it's the leader, but so does " + result.nodeId()); + } + result = leader; + } + } + if (result == null) { + throw new RuntimeException("No leader found."); + } + value.set(result); + }); + return value.get(); + } + + public List logManagers() { + return logManagers; + } + + @Override + public void close() throws InterruptedException { + try { + for (LocalLogManager logManager : logManagers) { + logManager.beginShutdown(); + } + for (LocalLogManager logManager : logManagers) { + logManager.close(); + } + Utils.delete(dir); + } catch (IOException e) { + log.error("Error deleting {}", dir.getAbsolutePath(), e); + } + } +} diff --git a/metadata/src/test/java/org/apache/kafka/metalog/MockMetaLogManagerListener.java b/metadata/src/test/java/org/apache/kafka/metalog/MockMetaLogManagerListener.java new file mode 100644 index 0000000000000..fe61ec070285b --- /dev/null +++ b/metadata/src/test/java/org/apache/kafka/metalog/MockMetaLogManagerListener.java @@ -0,0 +1,77 @@ +/* + * 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.metalog; + +import org.apache.kafka.common.protocol.ApiMessage; + +import java.util.ArrayList; +import java.util.List; + +public class MockMetaLogManagerListener implements MetaLogListener { + public static final String COMMIT = "COMMIT"; + public static final String LAST_COMMITTED_OFFSET = "LAST_COMMITTED_OFFSET"; + public static final String NEW_LEADER = "NEW_LEADER"; + public static final String RENOUNCE = "RENOUNCE"; + public static final String SHUTDOWN = "SHUTDOWN"; + + private final List serializedEvents = new ArrayList<>(); + + @Override + public synchronized void handleCommits(long lastCommittedOffset, List messages) { + for (ApiMessage message : messages) { + StringBuilder bld = new StringBuilder(); + bld.append(COMMIT).append(" ").append(message.toString()); + serializedEvents.add(bld.toString()); + } + StringBuilder bld = new StringBuilder(); + bld.append(LAST_COMMITTED_OFFSET).append(" ").append(lastCommittedOffset); + serializedEvents.add(bld.toString()); + } + + @Override + public void handleNewLeader(MetaLogLeader leader) { + StringBuilder bld = new StringBuilder(); + bld.append(NEW_LEADER).append(" "). + append(leader.nodeId()).append(" ").append(leader.epoch()); + synchronized (this) { + serializedEvents.add(bld.toString()); + } + } + + @Override + public void handleRenounce(long epoch) { + StringBuilder bld = new StringBuilder(); + bld.append(RENOUNCE).append(" ").append(epoch); + synchronized (this) { + serializedEvents.add(bld.toString()); + } + } + + @Override + public void beginShutdown() { + StringBuilder bld = new StringBuilder(); + bld.append(SHUTDOWN); + synchronized (this) { + serializedEvents.add(bld.toString()); + } + } + + public synchronized List serializedEvents() { + return new ArrayList<>(serializedEvents); + } +} From 6429e08ebc1bde63f0fac2e824d1e55833192426 Mon Sep 17 00:00:00 2001 From: David Arthur Date: Wed, 10 Feb 2021 18:14:52 -0500 Subject: [PATCH 2/5] Add back the quota processor --- .../server/metadata/BrokerMetadataListener.scala | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala b/core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala index 625ec2ba7f14f..c74d8c3d9a240 100644 --- a/core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala +++ b/core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala @@ -46,7 +46,8 @@ class BrokerMetadataListener(val brokerId: Int, val replicaManager: RaftReplicaManager, val txnCoordinator: TransactionCoordinator, val logManager: LogManager, - val threadNamePrefix: Option[String] + val threadNamePrefix: Option[String], + val clientQuotaManager: ClientQuotaMetadataManager ) extends MetaLogListener with KafkaMetricsGroup { val logContext = new LogContext(s"[BrokerMetadataListener id=${brokerId}] ") val log = logContext.logger(classOf[BrokerMetadataListener]) @@ -159,6 +160,8 @@ class BrokerMetadataListener(val brokerId: Int, record.asInstanceOf[UnfenceBrokerRecord]) case REMOVE_TOPIC_RECORD => handleRemoveTopicRecord(imageBuilder, record.asInstanceOf[RemoveTopicRecord]) + case QUOTA_RECORD => handleQuotaRecord(imageBuilder, + record.asInstanceOf[QuotaRecord]) // TODO: handle FEATURE_LEVEL_RECORD case _ => throw new RuntimeException(s"Unsupported record type ${recordType}") } @@ -224,6 +227,12 @@ class BrokerMetadataListener(val brokerId: Int, groupCoordinator.handleDeletedPartitions(removedPartitions.map(_.toTopicPartition).toSeq) } + def handleQuotaRecord(imageBuilder: MetadataImageBuilder, + record: QuotaRecord): Unit = { + // TODO add quotas to MetadataImageBuilder + clientQuotaManager.handleQuotaRecord(record) + } + class HandleNewLeaderEvent(leader: MetaLogLeader) extends EventQueue.FailureLoggingEvent(log) { override def run(): Unit = { From 5bbbe46638309215b814c5a0fd35b83171e5289f Mon Sep 17 00:00:00 2001 From: David Arthur Date: Thu, 11 Feb 2021 14:18:47 -0500 Subject: [PATCH 3/5] Remove redundant metalog folder --- .../apache/kafka/metalog/LocalLogManager.java | 378 ------------------ .../apache/kafka/metalog/MetaLogManager.java | 6 +- .../kafka/metalog/metalog/MetaLogLeader.java | 58 --- .../metalog/metalog/MetaLogListener.java | 55 --- .../kafka/metalog/metalog/MetaLogManager.java | 79 ---- .../metalog/metalog/LocalLogManagerTest.java | 153 ------- .../metalog/LocalLogManagerTestEnv.java | 143 ------- .../metalog/MockMetaLogManagerListener.java | 77 ---- 8 files changed, 3 insertions(+), 946 deletions(-) delete mode 100644 metadata/src/main/java/org/apache/kafka/metalog/LocalLogManager.java delete mode 100644 metadata/src/main/java/org/apache/kafka/metalog/metalog/MetaLogLeader.java delete mode 100644 metadata/src/main/java/org/apache/kafka/metalog/metalog/MetaLogListener.java delete mode 100644 metadata/src/main/java/org/apache/kafka/metalog/metalog/MetaLogManager.java delete mode 100644 metadata/src/test/java/org/apache/kafka/metalog/metalog/LocalLogManagerTest.java delete mode 100644 metadata/src/test/java/org/apache/kafka/metalog/metalog/LocalLogManagerTestEnv.java delete mode 100644 metadata/src/test/java/org/apache/kafka/metalog/metalog/MockMetaLogManagerListener.java diff --git a/metadata/src/main/java/org/apache/kafka/metalog/LocalLogManager.java b/metadata/src/main/java/org/apache/kafka/metalog/LocalLogManager.java deleted file mode 100644 index ef85314e0ef2f..0000000000000 --- a/metadata/src/main/java/org/apache/kafka/metalog/LocalLogManager.java +++ /dev/null @@ -1,378 +0,0 @@ -/* - * 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.metalog; - -import org.apache.kafka.common.protocol.ApiMessage; -import org.apache.kafka.common.utils.LogContext; -import org.apache.kafka.common.utils.Time; -import org.apache.kafka.metadata.ApiMessageAndVersion; -import org.apache.kafka.queue.EventQueue; -import org.apache.kafka.queue.KafkaEventQueue; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.AbstractMap.SimpleImmutableEntry; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.Iterator; -import java.util.List; -import java.util.Map.Entry; -import java.util.Objects; -import java.util.TreeMap; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.ThreadLocalRandom; -import java.util.stream.Collectors; - -/** - * The LocalLogManager is a test implementation that relies on the contents of memory. - */ -public final class LocalLogManager implements MetaLogManager, AutoCloseable { - interface LocalBatch { - int size(); - } - - static class LeaderChangeBatch implements LocalBatch { - private final MetaLogLeader newLeader; - - LeaderChangeBatch(MetaLogLeader newLeader) { - this.newLeader = newLeader; - } - - @Override - public int size() { - return 1; - } - - @Override - public boolean equals(Object o) { - if (!(o instanceof LeaderChangeBatch)) return false; - LeaderChangeBatch other = (LeaderChangeBatch) o; - if (!other.newLeader.equals(newLeader)) return false; - return true; - } - - @Override - public int hashCode() { - return Objects.hash(newLeader); - } - - @Override - public String toString() { - return "LeaderChangeBatch(newLeader=" + newLeader + ")"; - } - } - - static class LocalRecordBatch implements LocalBatch { - private final List records; - - LocalRecordBatch(List records) { - this.records = records; - } - - @Override - public int size() { - return records.size(); - } - - @Override - public boolean equals(Object o) { - if (!(o instanceof LocalRecordBatch)) return false; - LocalRecordBatch other = (LocalRecordBatch) o; - if (!other.records.equals(records)) return false; - return true; - } - - @Override - public int hashCode() { - return Objects.hash(records); - } - - @Override - public String toString() { - return "LocalRecordBatch(records=" + records + ")"; - } - } - - public static class SharedLogData { - private final Logger log = LoggerFactory.getLogger(SharedLogData.class); - private final HashMap logManagers = new HashMap<>(); - private final TreeMap batches = new TreeMap<>(); - private MetaLogLeader leader = new MetaLogLeader(-1, -1); - private long prevOffset = -1; - - synchronized void registerLogManager(LocalLogManager logManager) { - if (logManagers.put(logManager.nodeId(), logManager) != null) { - throw new RuntimeException("Can't have multiple LocalLogManagers " + - "with id " + logManager.nodeId()); - } - electLeaderIfNeeded(); - } - - synchronized void unregisterLogManager(LocalLogManager logManager) { - if (!logManagers.remove(logManager.nodeId(), logManager)) { - throw new RuntimeException("Log manager " + logManager.nodeId() + - " was not found."); - } - } - - synchronized long tryAppend(int nodeId, long epoch, LocalBatch batch) { - if (epoch != leader.epoch()) { - log.trace("tryAppend(nodeId={}, epoch={}): the provided epoch does not " + - "match the current leader epoch of {}.", nodeId, epoch, leader.epoch()); - return Long.MAX_VALUE; - } - if (nodeId != leader.nodeId()) { - log.trace("tryAppend(nodeId={}, epoch={}): the given node id does not " + - "match the current leader id of {}.", nodeId, leader.nodeId()); - return Long.MAX_VALUE; - } - log.trace("tryAppend(nodeId={}): appending {}.", nodeId, batch); - long offset = append(batch); - electLeaderIfNeeded(); - return offset; - } - - synchronized long append(LocalBatch batch) { - prevOffset += batch.size(); - log.debug("append(batch={}, prevOffset={})", batch, prevOffset); - batches.put(prevOffset, batch); - if (batch instanceof LeaderChangeBatch) { - LeaderChangeBatch leaderChangeBatch = (LeaderChangeBatch) batch; - leader = leaderChangeBatch.newLeader; - } - for (LocalLogManager logManager : logManagers.values()) { - logManager.scheduleLogCheck(); - } - return prevOffset; - } - - synchronized void electLeaderIfNeeded() { - if (leader.nodeId() != -1 || logManagers.isEmpty()) { - return; - } - int nextLeaderIndex = ThreadLocalRandom.current().nextInt(logManagers.size()); - Iterator iter = logManagers.keySet().iterator(); - Integer nextLeaderNode = null; - for (int i = 0; i <= nextLeaderIndex; i++) { - nextLeaderNode = iter.next(); - } - MetaLogLeader newLeader = new MetaLogLeader(nextLeaderNode, leader.epoch() + 1); - log.info("Elected new leader: {}.", newLeader); - append(new LeaderChangeBatch(newLeader)); - } - - synchronized Entry nextBatch(long offset) { - Entry entry = batches.higherEntry(offset); - if (entry == null) { - return null; - } - return new SimpleImmutableEntry<>(entry.getKey(), entry.getValue()); - } - } - - private static class MetaLogListenerData { - private long offset = -1; - private final MetaLogListener listener; - - MetaLogListenerData(MetaLogListener listener) { - this.listener = listener; - } - } - - private final Logger log; - - private final int nodeId; - - private final SharedLogData shared; - - private final EventQueue eventQueue; - - private boolean initialized = false; - - private boolean shutdown = false; - - private long maxReadOffset = Long.MAX_VALUE; - - private final List listeners = new ArrayList<>(); - - private volatile MetaLogLeader leader = new MetaLogLeader(-1, -1); - - public LocalLogManager(LogContext logContext, - int nodeId, - SharedLogData shared, - String threadNamePrefix) { - this.log = logContext.logger(LocalLogManager.class); - this.nodeId = nodeId; - this.shared = shared; - this.eventQueue = new KafkaEventQueue(Time.SYSTEM, logContext, threadNamePrefix); - shared.registerLogManager(this); - } - - private void scheduleLogCheck() { - eventQueue.append(() -> { - try { - log.debug("Node {}: running log check.", nodeId); - int numEntriesFound = 0; - for (MetaLogListenerData listenerData : listeners) { - while (true) { - Entry entry = shared.nextBatch(listenerData.offset); - if (entry == null) { - log.trace("Node {}: reached the end of the log after finding " + - "{} entries.", nodeId, numEntriesFound); - break; - } - long entryOffset = entry.getKey(); - if (entryOffset > maxReadOffset) { - log.trace("Node {}: after {} entries, not reading the next " + - "entry because its offset is {}, and maxReadOffset is {}.", - nodeId, numEntriesFound, entryOffset, maxReadOffset); - break; - } - if (entry.getValue() instanceof LeaderChangeBatch) { - LeaderChangeBatch batch = (LeaderChangeBatch) entry.getValue(); - log.trace("Node {}: handling LeaderChange to {}.", - nodeId, batch.newLeader); - listenerData.listener.handleNewLeader(batch.newLeader); - if (batch.newLeader.epoch() > leader.epoch()) { - leader = batch.newLeader; - } - } else if (entry.getValue() instanceof LocalRecordBatch) { - LocalRecordBatch batch = (LocalRecordBatch) entry.getValue(); - log.trace("Node {}: handling LocalRecordBatch with offset {}.", - nodeId, entryOffset); - listenerData.listener.handleCommits(entryOffset, batch.records); - } - numEntriesFound++; - listenerData.offset = entryOffset; - } - } - log.trace("Completed log check for node " + nodeId); - } catch (Exception e) { - log.error("Exception while handling log check", e); - } - }); - } - - public void beginShutdown() { - eventQueue.beginShutdown("beginShutdown", () -> { - try { - if (initialized && !shutdown) { - log.debug("Node {}: beginning shutdown.", nodeId); - renounce(leader.epoch()); - for (MetaLogListenerData listenerData : listeners) { - listenerData.listener.beginShutdown(); - } - shared.unregisterLogManager(this); - } - } catch (Exception e) { - log.error("Unexpected exception while sending beginShutdown callbacks", e); - } - shutdown = true; - }); - } - - @Override - public void close() throws InterruptedException { - log.debug("Node {}: closing.", nodeId); - beginShutdown(); - eventQueue.close(); - } - - @Override - public void initialize() throws Exception { - eventQueue.append(() -> { - log.debug("initialized local log manager for node " + nodeId); - initialized = true; - }); - } - - @Override - public void register(MetaLogListener listener) throws Exception { - CompletableFuture future = new CompletableFuture<>(); - eventQueue.append(() -> { - if (shutdown) { - log.info("Node {}: can't register because local log manager has " + - "already been shut down.", nodeId); - future.complete(null); - } else if (initialized) { - log.info("Node {}: registered MetaLogListener.", nodeId); - listeners.add(new MetaLogListenerData(listener)); - shared.electLeaderIfNeeded(); - scheduleLogCheck(); - future.complete(null); - } else { - log.info("Node {}: can't register because local log manager has not " + - "been initialized.", nodeId); - future.completeExceptionally(new RuntimeException( - "LocalLogManager was not initialized.")); - } - }); - future.get(); - } - - @Override - public long scheduleWrite(long epoch, List batch) { - return shared.tryAppend(nodeId, leader.epoch(), new LocalRecordBatch( - batch.stream().map(r -> r.message()).collect(Collectors.toList()))); - } - - @Override - public void renounce(long epoch) { - MetaLogLeader curLeader = leader; - MetaLogLeader nextLeader = new MetaLogLeader(-1, curLeader.epoch() + 1); - shared.tryAppend(nodeId, curLeader.epoch(), new LeaderChangeBatch(nextLeader)); - } - - @Override - public MetaLogLeader leader() { - return leader; - } - - @Override - public int nodeId() { - return nodeId; - } - - public List listeners() { - final CompletableFuture> future = new CompletableFuture<>(); - eventQueue.append(() -> { - future.complete(listeners.stream().map(l -> l.listener).collect(Collectors.toList())); - }); - try { - return future.get(); - } catch (ExecutionException | InterruptedException e) { - throw new RuntimeException(e); - } - } - - public void setMaxReadOffset(long maxReadOffset) { - CompletableFuture future = new CompletableFuture<>(); - eventQueue.append(() -> { - log.trace("Node {}: set maxReadOffset to {}.", nodeId, maxReadOffset); - this.maxReadOffset = maxReadOffset; - scheduleLogCheck(); - future.complete(null); - }); - try { - future.get(); - } catch (ExecutionException | InterruptedException e) { - throw new RuntimeException(e); - } - } -} diff --git a/metadata/src/main/java/org/apache/kafka/metalog/MetaLogManager.java b/metadata/src/main/java/org/apache/kafka/metalog/MetaLogManager.java index 94b35ea72bf64..67a6ca5385f75 100644 --- a/metadata/src/main/java/org/apache/kafka/metalog/MetaLogManager.java +++ b/metadata/src/main/java/org/apache/kafka/metalog/MetaLogManager.java @@ -47,13 +47,13 @@ public interface MetaLogManager { * The write will be scheduled to happen at some time in the future. There is no * error return or exception thrown if the write fails. Instead, the listener may * regard the write as successful if and only if the MetaLogManager reaches the given - * index before renouncing its leadership. The listener should determine this by - * monitoring the committed indexes. + * offset before renouncing its leadership. The listener should determine this by + * monitoring the committed offsets. * * @param epoch The controller epoch. * @param batch The batch of messages to write. * - * @return The index of the message. + * @return The offset of the message. */ long scheduleWrite(long epoch, List batch); diff --git a/metadata/src/main/java/org/apache/kafka/metalog/metalog/MetaLogLeader.java b/metadata/src/main/java/org/apache/kafka/metalog/metalog/MetaLogLeader.java deleted file mode 100644 index 2bf4f7c718bd5..0000000000000 --- a/metadata/src/main/java/org/apache/kafka/metalog/metalog/MetaLogLeader.java +++ /dev/null @@ -1,58 +0,0 @@ -/* - * 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.metalog; - -import java.util.Objects; - -/** - * The current leader of the MetaLog. - */ -public class MetaLogLeader { - private final int nodeId; - private final long epoch; - - public MetaLogLeader(int nodeId, long epoch) { - this.nodeId = nodeId; - this.epoch = epoch; - } - - public int nodeId() { - return nodeId; - } - - public long epoch() { - return epoch; - } - - @Override - public boolean equals(Object o) { - if (!(o instanceof MetaLogLeader)) return false; - MetaLogLeader other = (MetaLogLeader) o; - return other.nodeId == nodeId && other.epoch == epoch; - } - - @Override - public int hashCode() { - return Objects.hash(nodeId, epoch); - } - - @Override - public String toString() { - return "MetaLogLeader(nodeId=" + nodeId + ", epoch=" + epoch + ")"; - } -} diff --git a/metadata/src/main/java/org/apache/kafka/metalog/metalog/MetaLogListener.java b/metadata/src/main/java/org/apache/kafka/metalog/metalog/MetaLogListener.java deleted file mode 100644 index 93744202dc90f..0000000000000 --- a/metadata/src/main/java/org/apache/kafka/metalog/metalog/MetaLogListener.java +++ /dev/null @@ -1,55 +0,0 @@ -/* - * 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.metalog; - -import org.apache.kafka.common.protocol.ApiMessage; - -import java.util.List; - -/** - * Listeners receive notifications from the MetaLogManager. - */ -public interface MetaLogListener { - /** - * Called when the MetaLogManager commits some messages. - * - * @param lastOffset The last offset found in all the given messages. - * @param messages The messages. - */ - void handleCommits(long lastOffset, List messages); - - /** - * Called when a new leader is elected. - * - * @param leader The new leader id and epoch. - */ - default void handleNewLeader(MetaLogLeader leader) {} - - /** - * Called when the MetaLogManager has renounced the leadership. - * - * @param epoch The controller epoch that has ended. - */ - default void handleRenounce(long epoch) {} - - /** - * Called when the MetaLogManager has finished shutting down, and wants to tell its - * listener that it is safe to shut down as well. - */ - default void beginShutdown() {} -} diff --git a/metadata/src/main/java/org/apache/kafka/metalog/metalog/MetaLogManager.java b/metadata/src/main/java/org/apache/kafka/metalog/metalog/MetaLogManager.java deleted file mode 100644 index 67a6ca5385f75..0000000000000 --- a/metadata/src/main/java/org/apache/kafka/metalog/metalog/MetaLogManager.java +++ /dev/null @@ -1,79 +0,0 @@ -/* - * 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.metalog; - -import org.apache.kafka.metadata.ApiMessageAndVersion; - -import java.util.List; - -/** - * The MetaLogManager handles storing metadata and electing leaders. - */ -public interface MetaLogManager { - - /** - * Start this meta log manager. - * The manager must be ready to accept incoming calls after this function returns. - * It is an error to initialize a MetaLogManager more than once. - */ - void initialize() throws Exception; - - /** - * Register the listener. The manager must be initialized already. - * The listener must be ready to accept incoming calls immediately. - * - * @param listener The listener to register. - */ - void register(MetaLogListener listener) throws Exception; - - /** - * Schedule a write to the log. - * - * The write will be scheduled to happen at some time in the future. There is no - * error return or exception thrown if the write fails. Instead, the listener may - * regard the write as successful if and only if the MetaLogManager reaches the given - * offset before renouncing its leadership. The listener should determine this by - * monitoring the committed offsets. - * - * @param epoch The controller epoch. - * @param batch The batch of messages to write. - * - * @return The offset of the message. - */ - long scheduleWrite(long epoch, List batch); - - /** - * Renounce the leadership. - * - * @param epoch The epoch. If this does not match the current epoch, this - * call will be ignored. - */ - void renounce(long epoch); - - /** - * Returns the current leader. The active node may change immediately after this - * function is called, of course. - */ - MetaLogLeader leader(); - - /** - * Returns the node id. - */ - int nodeId(); - -} diff --git a/metadata/src/test/java/org/apache/kafka/metalog/metalog/LocalLogManagerTest.java b/metadata/src/test/java/org/apache/kafka/metalog/metalog/LocalLogManagerTest.java deleted file mode 100644 index 9d4eb8b594e29..0000000000000 --- a/metadata/src/test/java/org/apache/kafka/metalog/metalog/LocalLogManagerTest.java +++ /dev/null @@ -1,153 +0,0 @@ -/* - * 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.metalog; - -import org.apache.kafka.common.metadata.RegisterBrokerRecord; -import org.apache.kafka.metadata.ApiMessageAndVersion; -import org.apache.kafka.test.TestUtils; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.Timeout; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.Arrays; -import java.util.List; -import java.util.stream.Collectors; - -import static org.apache.kafka.metalog.MockMetaLogManagerListener.COMMIT; -import static org.apache.kafka.metalog.MockMetaLogManagerListener.LAST_COMMITTED_OFFSET; -import static org.apache.kafka.metalog.MockMetaLogManagerListener.SHUTDOWN; -import static org.junit.jupiter.api.Assertions.assertEquals; - -@Timeout(value = 40) -public class LocalLogManagerTest { - private static final Logger log = LoggerFactory.getLogger(LocalLogManagerTest.class); - - /** - * Test creating a LocalLogManager and closing it. - */ - @Test - public void testCreateAndClose() throws Exception { - try (LocalLogManagerTestEnv env = - LocalLogManagerTestEnv.createWithMockListeners(1)) { - env.close(); - assertEquals(null, env.firstError.get()); - } - } - - /** - * Test that the local log maanger will claim leadership. - */ - @Test - public void testClaimsLeadership() throws Exception { - try (LocalLogManagerTestEnv env = - LocalLogManagerTestEnv.createWithMockListeners(1)) { - assertEquals(new MetaLogLeader(0, 0), env.waitForLeader()); - env.close(); - assertEquals(null, env.firstError.get()); - } - } - - /** - * Test that we can pass leadership back and forth between log managers. - */ - @Test - public void testPassLeadership() throws Exception { - try (LocalLogManagerTestEnv env = - LocalLogManagerTestEnv.createWithMockListeners(3)) { - MetaLogLeader first = env.waitForLeader(); - MetaLogLeader cur = first; - do { - env.logManagers().get(cur.nodeId()).renounce(cur.epoch()); - MetaLogLeader next = env.waitForLeader(); - while (next.epoch() == cur.epoch()) { - Thread.sleep(1); - next = env.waitForLeader(); - } - long expectedNextEpoch = cur.epoch() + 2; - assertEquals(expectedNextEpoch, next.epoch(), "Expected next epoch to be " + expectedNextEpoch + - ", but found " + next); - cur = next; - } while (cur.nodeId() == first.nodeId()); - env.close(); - assertEquals(null, env.firstError.get()); - } - } - - private static void waitForLastCommittedOffset(long targetOffset, - LocalLogManager logManager) throws InterruptedException { - TestUtils.retryOnExceptionWithTimeout(3, 20000, () -> { - MockMetaLogManagerListener listener = - (MockMetaLogManagerListener) logManager.listeners().get(0); - long highestOffset = -1; - for (String event : listener.serializedEvents()) { - if (event.startsWith(LAST_COMMITTED_OFFSET)) { - long offset = Long.valueOf( - event.substring(LAST_COMMITTED_OFFSET.length() + 1)); - if (offset < highestOffset) { - throw new RuntimeException("Invalid offset: " + offset + - " is less than the previous offset of " + highestOffset); - } - highestOffset = offset; - } - } - if (highestOffset < targetOffset) { - throw new RuntimeException("Offset for log manager " + - logManager.nodeId() + " only reached " + highestOffset); - } - }); - } - - /** - * Test that all the log managers see all the commits. - */ - @Test - public void testCommits() throws Exception { - try (LocalLogManagerTestEnv env = - LocalLogManagerTestEnv.createWithMockListeners(3)) { - MetaLogLeader leaderInfo = env.waitForLeader(); - LocalLogManager activeLogManager = env.logManagers().get(leaderInfo.nodeId()); - long epoch = activeLogManager.leader().epoch(); - List messages = Arrays.asList( - new ApiMessageAndVersion(new RegisterBrokerRecord().setBrokerId(0), (short) 0), - new ApiMessageAndVersion(new RegisterBrokerRecord().setBrokerId(1), (short) 0), - new ApiMessageAndVersion(new RegisterBrokerRecord().setBrokerId(2), (short) 0)); - assertEquals(3, activeLogManager.scheduleWrite(epoch, messages)); - for (LocalLogManager logManager : env.logManagers()) { - waitForLastCommittedOffset(3, logManager); - } - List listeners = env.logManagers().stream(). - map(m -> (MockMetaLogManagerListener) m.listeners().get(0)). - collect(Collectors.toList()); - env.close(); - for (MockMetaLogManagerListener listener : listeners) { - List events = listener.serializedEvents(); - assertEquals(SHUTDOWN, events.get(events.size() - 1)); - int foundIndex = 0; - for (String event : events) { - if (event.startsWith(COMMIT)) { - assertEquals(messages.get(foundIndex).message().toString(), - event.substring(COMMIT.length() + 1)); - foundIndex++; - } - } - assertEquals(messages.size(), foundIndex); - } - } - } -} diff --git a/metadata/src/test/java/org/apache/kafka/metalog/metalog/LocalLogManagerTestEnv.java b/metadata/src/test/java/org/apache/kafka/metalog/metalog/LocalLogManagerTestEnv.java deleted file mode 100644 index 52aeea052bdde..0000000000000 --- a/metadata/src/test/java/org/apache/kafka/metalog/metalog/LocalLogManagerTestEnv.java +++ /dev/null @@ -1,143 +0,0 @@ -/* - * 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.metalog; - -import org.apache.kafka.common.utils.LogContext; -import org.apache.kafka.common.utils.Utils; -import org.apache.kafka.metalog.LocalLogManager.SharedLogData; -import org.apache.kafka.test.TestUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.File; -import java.io.IOException; -import java.util.ArrayList; -import java.util.List; -import java.util.concurrent.atomic.AtomicReference; - -public class LocalLogManagerTestEnv implements AutoCloseable { - private static final Logger log = - LoggerFactory.getLogger(LocalLogManagerTestEnv.class); - - /** - * The first error we encountered during this test, or the empty string if we have - * not encountered any. - */ - final AtomicReference firstError = new AtomicReference<>(null); - - /** - * The test directory, which we will delete once the test is over. - */ - private final File dir; - - /** - * The shared data for our LocalLogManager instances. - */ - private final SharedLogData shared; - - /** - * A list of log managers. - */ - private final List logManagers; - - public static LocalLogManagerTestEnv createWithMockListeners(int numManagers) throws Exception { - LocalLogManagerTestEnv testEnv = new LocalLogManagerTestEnv(numManagers); - try { - for (LocalLogManager logManager : testEnv.logManagers) { - logManager.register(new MockMetaLogManagerListener()); - } - } catch (Exception e) { - testEnv.close(); - throw e; - } - return testEnv; - } - - public LocalLogManagerTestEnv(int numManagers) throws Exception { - dir = TestUtils.tempDirectory(); - shared = new SharedLogData(); - List newLogManagers = new ArrayList<>(numManagers); - try { - for (int nodeId = 0; nodeId < numManagers; nodeId++) { - newLogManagers.add(new LocalLogManager( - new LogContext(String.format("[LocalLogManager %d] ", nodeId)), - nodeId, - shared, - String.format("LocalLogManager-%d_", nodeId))); - } - for (LocalLogManager logManager : newLogManagers) { - logManager.initialize(); - } - } catch (Throwable t) { - for (LocalLogManager logManager : newLogManagers) { - logManager.close(); - } - throw t; - } - this.logManagers = newLogManagers; - } - - AtomicReference firstError() { - return firstError; - } - - File dir() { - return dir; - } - - MetaLogLeader waitForLeader() throws InterruptedException { - AtomicReference value = new AtomicReference<>(null); - TestUtils.retryOnExceptionWithTimeout(3, 20000, () -> { - MetaLogLeader result = null; - for (LocalLogManager logManager : logManagers) { - MetaLogLeader leader = logManager.leader(); - if (leader.nodeId() == logManager.nodeId()) { - if (result != null) { - throw new RuntimeException("node " + leader.nodeId() + - " thinks it's the leader, but so does " + result.nodeId()); - } - result = leader; - } - } - if (result == null) { - throw new RuntimeException("No leader found."); - } - value.set(result); - }); - return value.get(); - } - - public List logManagers() { - return logManagers; - } - - @Override - public void close() throws InterruptedException { - try { - for (LocalLogManager logManager : logManagers) { - logManager.beginShutdown(); - } - for (LocalLogManager logManager : logManagers) { - logManager.close(); - } - Utils.delete(dir); - } catch (IOException e) { - log.error("Error deleting {}", dir.getAbsolutePath(), e); - } - } -} diff --git a/metadata/src/test/java/org/apache/kafka/metalog/metalog/MockMetaLogManagerListener.java b/metadata/src/test/java/org/apache/kafka/metalog/metalog/MockMetaLogManagerListener.java deleted file mode 100644 index fe61ec070285b..0000000000000 --- a/metadata/src/test/java/org/apache/kafka/metalog/metalog/MockMetaLogManagerListener.java +++ /dev/null @@ -1,77 +0,0 @@ -/* - * 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.metalog; - -import org.apache.kafka.common.protocol.ApiMessage; - -import java.util.ArrayList; -import java.util.List; - -public class MockMetaLogManagerListener implements MetaLogListener { - public static final String COMMIT = "COMMIT"; - public static final String LAST_COMMITTED_OFFSET = "LAST_COMMITTED_OFFSET"; - public static final String NEW_LEADER = "NEW_LEADER"; - public static final String RENOUNCE = "RENOUNCE"; - public static final String SHUTDOWN = "SHUTDOWN"; - - private final List serializedEvents = new ArrayList<>(); - - @Override - public synchronized void handleCommits(long lastCommittedOffset, List messages) { - for (ApiMessage message : messages) { - StringBuilder bld = new StringBuilder(); - bld.append(COMMIT).append(" ").append(message.toString()); - serializedEvents.add(bld.toString()); - } - StringBuilder bld = new StringBuilder(); - bld.append(LAST_COMMITTED_OFFSET).append(" ").append(lastCommittedOffset); - serializedEvents.add(bld.toString()); - } - - @Override - public void handleNewLeader(MetaLogLeader leader) { - StringBuilder bld = new StringBuilder(); - bld.append(NEW_LEADER).append(" "). - append(leader.nodeId()).append(" ").append(leader.epoch()); - synchronized (this) { - serializedEvents.add(bld.toString()); - } - } - - @Override - public void handleRenounce(long epoch) { - StringBuilder bld = new StringBuilder(); - bld.append(RENOUNCE).append(" ").append(epoch); - synchronized (this) { - serializedEvents.add(bld.toString()); - } - } - - @Override - public void beginShutdown() { - StringBuilder bld = new StringBuilder(); - bld.append(SHUTDOWN); - synchronized (this) { - serializedEvents.add(bld.toString()); - } - } - - public synchronized List serializedEvents() { - return new ArrayList<>(serializedEvents); - } -} From f3c433220d06de65379c951ad6b0041c56018661 Mon Sep 17 00:00:00 2001 From: David Arthur Date: Thu, 11 Feb 2021 14:58:21 -0500 Subject: [PATCH 4/5] Remove LocalConfigRepository (it was renamed on trunk) --- .../metadata/BrokerMetadataListener.scala | 2 +- .../metadata/LocalConfigRepository.scala | 93 ------------------- .../metadata/LocalConfigRepositoryTest.scala | 56 ----------- 3 files changed, 1 insertion(+), 150 deletions(-) delete mode 100644 core/src/main/scala/kafka/server/metadata/LocalConfigRepository.scala delete mode 100644 core/src/test/scala/kafka/server/metadata/LocalConfigRepositoryTest.scala diff --git a/core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala b/core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala index c74d8c3d9a240..6a901ff74d76b 100644 --- a/core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala +++ b/core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala @@ -41,7 +41,7 @@ object BrokerMetadataListener{ class BrokerMetadataListener(val brokerId: Int, val time: Time, val metadataCache: RaftMetadataCache, - val configRepository: LocalConfigRepository, + val configRepository: CachedConfigRepository, val groupCoordinator: GroupCoordinator, val replicaManager: RaftReplicaManager, val txnCoordinator: TransactionCoordinator, diff --git a/core/src/main/scala/kafka/server/metadata/LocalConfigRepository.scala b/core/src/main/scala/kafka/server/metadata/LocalConfigRepository.scala deleted file mode 100644 index a46993ba777cd..0000000000000 --- a/core/src/main/scala/kafka/server/metadata/LocalConfigRepository.scala +++ /dev/null @@ -1,93 +0,0 @@ -/** - * 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 kafka.server.metadata - -import java.util -import java.util.Properties -import java.util.concurrent.ConcurrentHashMap -import java.util.function.BiFunction - -import org.apache.kafka.common.config.ConfigResource -import org.apache.kafka.common.config.ConfigResource.Type - -import scala.jdk.CollectionConverters._ - -/** - * A ConfigRepository that stores configurations locally. - */ -class LocalConfigRepository extends ConfigRepository { - val configMap = new ConcurrentHashMap[ConfigResource, util.HashMap[String, String]] - - def setTopicConfig(topic: String, key: String, value: String): Unit = { - setConfig(new ConfigResource(Type.TOPIC, topic), key, value) - } - - def setBrokerConfig(id: Int, key: String, value: String): Unit = { - setConfig(new ConfigResource(Type.BROKER, id.toString()), key, value) - } - - def setConfig(configResource: ConfigResource, key: String, value: String): Unit = { - configMap.compute(configResource, new BiFunction[ConfigResource, util.HashMap[String, String], util.HashMap[String, String]] { - override def apply(resource: ConfigResource, - curConfigs: util.HashMap[String, String]): util.HashMap[String, String] = { - if (value == null) { - if (curConfigs == null) { - null - } else { - val newConfigs = new util.HashMap[String, String](curConfigs) - newConfigs.remove(key) - if (newConfigs.isEmpty) { - null - } else { - newConfigs - } - } - } else { - if (curConfigs == null) { - val newConfigs = new util.HashMap[String, String](1) - newConfigs.put(key, value) - newConfigs - } else { - val newConfigs = new util.HashMap[String, String](curConfigs.size() + 1) - newConfigs.putAll(curConfigs) - newConfigs.put(key, value) - newConfigs - } - } - } - }) - } - - def config(configResource: ConfigResource): Properties = { - val properties = new Properties() - Option(configMap.get(configResource)).foreach { - _.entrySet().iterator().asScala.foreach { case e => - properties.put(e.getKey, e.getValue) - } - } - properties - } - - override def topicConfig(topic: String): Properties = { - config(new ConfigResource(Type.TOPIC, topic)) - } - - override def brokerConfig(id: Int): Properties = { - config(new ConfigResource(Type.BROKER, id.toString())) - } -} diff --git a/core/src/test/scala/kafka/server/metadata/LocalConfigRepositoryTest.scala b/core/src/test/scala/kafka/server/metadata/LocalConfigRepositoryTest.scala deleted file mode 100644 index dd651ebbf4175..0000000000000 --- a/core/src/test/scala/kafka/server/metadata/LocalConfigRepositoryTest.scala +++ /dev/null @@ -1,56 +0,0 @@ -/** - * 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 kafka.server.metadata - -import org.junit.jupiter.api.Assertions._ -import org.junit.jupiter.api.{Test, Timeout} - -import java.util.Properties -import java.util.concurrent.TimeUnit - - -@Timeout(value = 120000, unit = TimeUnit.MILLISECONDS) -class LocalConfigRepositoryTest { - @Test - def testEmptyRepository(): Unit = { - val repository = new LocalConfigRepository() - assertEquals(new Properties(), repository.brokerConfig(0)) - assertEquals(new Properties(), repository.topicConfig("foo")) - } - - @Test - def testSetConfig(): Unit = { - val repository = new LocalConfigRepository() - repository.setBrokerConfig(0, "foo", null) - assertEquals(new Properties(), repository.brokerConfig(0)) - - repository.setBrokerConfig(1, "foo", "bar") - val brokerProperties = new Properties() - brokerProperties.put("foo", "bar") - assertEquals(brokerProperties, repository.brokerConfig(1)) - - val brokerProperties2 = new Properties() - brokerProperties2.put("foo", "bar") - brokerProperties2.put("foo2", "baz") - repository.setBrokerConfig(1, "foo2", "baz") - assertEquals(brokerProperties2, repository.brokerConfig(1)) - - repository.setBrokerConfig(1, "foo2", null) - assertEquals(brokerProperties, repository.brokerConfig(1)) - } -} From 1451071927103ae374aa00e105037a92ed0b855b Mon Sep 17 00:00:00 2001 From: David Arthur Date: Thu, 11 Feb 2021 15:11:31 -0500 Subject: [PATCH 5/5] Some small cleanup --- .../metadata/BrokerMetadataListener.scala | 32 +++++++++---------- 1 file changed, 16 insertions(+), 16 deletions(-) diff --git a/core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala b/core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala index 6a901ff74d76b..6185a35d3f35d 100644 --- a/core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala +++ b/core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala @@ -38,19 +38,19 @@ object BrokerMetadataListener{ val MetadataBatchSizes = "MetadataBatchSizes" } -class BrokerMetadataListener(val brokerId: Int, - val time: Time, - val metadataCache: RaftMetadataCache, - val configRepository: CachedConfigRepository, - val groupCoordinator: GroupCoordinator, - val replicaManager: RaftReplicaManager, - val txnCoordinator: TransactionCoordinator, - val logManager: LogManager, - val threadNamePrefix: Option[String], - val clientQuotaManager: ClientQuotaMetadataManager +class BrokerMetadataListener(brokerId: Int, + time: Time, + metadataCache: RaftMetadataCache, + configRepository: CachedConfigRepository, + groupCoordinator: GroupCoordinator, + replicaManager: RaftReplicaManager, + txnCoordinator: TransactionCoordinator, + logManager: LogManager, + threadNamePrefix: Option[String], + clientQuotaManager: ClientQuotaMetadataManager ) extends MetaLogListener with KafkaMetricsGroup { - val logContext = new LogContext(s"[BrokerMetadataListener id=${brokerId}] ") - val log = logContext.logger(classOf[BrokerMetadataListener]) + private val logContext = new LogContext(s"[BrokerMetadataListener id=${brokerId}] ") + private val log = logContext.logger(classOf[BrokerMetadataListener]) logIdent = logContext.logPrefix() /** @@ -91,11 +91,11 @@ class BrokerMetadataListener(val brokerId: Int, val startNs = time.nanoseconds() var index = 0 metadataBatchSizeHist.update(records.size()) - records.iterator().asScala.foreach { case record => + records.iterator().asScala.foreach { record => try { if (isTraceEnabled) { trace("Metadata batch %d: processing [%d/%d]: %s.".format(lastOffset, index + 1, - records.size(), record.toString())) + records.size(), record.toString)) } handleMessage(imageBuilder, record, lastOffset) } catch { @@ -211,13 +211,13 @@ class BrokerMetadataListener(val brokerId: Int, def handleFenceBrokerRecord(imageBuilder: MetadataImageBuilder, record: FenceBrokerRecord): Unit = { // TODO: add broker epoch to metadata cache, and check it here. - imageBuilder.brokersBuilder().changeFencing(record.id(), true) + imageBuilder.brokersBuilder().changeFencing(record.id(), fenced = true) } def handleUnfenceBrokerRecord(imageBuilder: MetadataImageBuilder, record: UnfenceBrokerRecord): Unit = { // TODO: add broker epoch to metadata cache, and check it here. - imageBuilder.brokersBuilder().changeFencing(record.id(), false) + imageBuilder.brokersBuilder().changeFencing(record.id(), fenced = false) } def handleRemoveTopicRecord(imageBuilder: MetadataImageBuilder,