diff --git a/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java b/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java index c295713df2370..8bfde0b6db9ef 100644 --- a/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java +++ b/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java @@ -117,6 +117,7 @@ public static List parseAndValidateAddresses(List url public static ChannelBuilder createChannelBuilder(AbstractConfig config, Time time, LogContext logContext) { SecurityProtocol securityProtocol = SecurityProtocol.forName(config.getString(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG)); String clientSaslMechanism = config.getString(SaslConfigs.SASL_MECHANISM); + log.info("!!! createChannelBuilder:" + securityProtocol + " " + clientSaslMechanism); return ChannelBuilders.clientChannelBuilder(securityProtocol, JaasContext.Type.CLIENT, config, null, clientSaslMechanism, time, logContext); } diff --git a/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java b/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java index a3a0baf1e39db..36bbccf789bf2 100644 --- a/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java +++ b/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java @@ -22,6 +22,7 @@ import org.apache.kafka.common.config.SslClientAuth; import org.apache.kafka.common.config.SslConfigs; import org.apache.kafka.common.security.auth.KafkaPrincipalBuilder; +import org.apache.kafka.common.security.auth.SecurityProtocol; import org.apache.kafka.common.security.authenticator.DefaultKafkaPrincipalBuilder; import org.apache.kafka.common.utils.Utils; @@ -173,6 +174,12 @@ public class BrokerSecurityConfigs { .define(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG, CLASS, null, LOW, SslConfigs.SSL_ENGINE_FACTORY_CLASS_DOC) // Sasl Configuration + .define("security.protocol", ConfigDef.Type.STRING, "PLAINTEXT", + ConfigDef.CaseInsensitiveValidString + .in(Utils.enumOptions(SecurityProtocol.class)), + ConfigDef.Importance.MEDIUM, + "Protocol used to communicate with brokers.") + .define(SaslConfigs.SASL_MECHANISM, ConfigDef.Type.STRING, SaslConfigs.DEFAULT_SASL_MECHANISM, ConfigDef.Importance.MEDIUM, SaslConfigs.SASL_MECHANISM_DOC) .define(BrokerSecurityConfigs.SASL_MECHANISM_INTER_BROKER_PROTOCOL_CONFIG, STRING, SaslConfigs.DEFAULT_SASL_MECHANISM, MEDIUM, BrokerSecurityConfigs.SASL_MECHANISM_INTER_BROKER_PROTOCOL_DOC) .define(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG, LIST, BrokerSecurityConfigs.DEFAULT_SASL_ENABLED_MECHANISMS, MEDIUM, BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_DOC) .define(BrokerSecurityConfigs.SASL_SERVER_CALLBACK_HANDLER_CLASS_CONFIG, CLASS, null, MEDIUM, BrokerSecurityConfigs.SASL_SERVER_CALLBACK_HANDLER_CLASS_DOC) diff --git a/core/src/main/java/kafka/server/RemoteClusterMetadataManager.java b/core/src/main/java/kafka/server/RemoteClusterMetadataManager.java new file mode 100644 index 0000000000000..d22f02a270640 --- /dev/null +++ b/core/src/main/java/kafka/server/RemoteClusterMetadataManager.java @@ -0,0 +1,229 @@ +/* + * 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; + +import org.apache.kafka.clients.ClientResponse; +import org.apache.kafka.clients.ClientUtils; +import org.apache.kafka.common.Node; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.message.CreatePartitionsRequestData; +import org.apache.kafka.common.metrics.Metrics; +import org.apache.kafka.common.requests.CreatePartitionsRequest; +import org.apache.kafka.common.requests.MetadataRequest; +import org.apache.kafka.common.requests.MetadataResponse; +import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.common.utils.Time; +import org.apache.kafka.image.LocalReplicaChanges; +import org.apache.kafka.image.MetadataDelta; +import org.apache.kafka.image.MetadataImage; +import org.apache.kafka.server.common.ControllerRequestCompletionHandler; +import org.apache.kafka.server.common.NodeToControllerChannelManager; +import org.apache.kafka.server.network.BrokerEndPoint; +import org.apache.kafka.server.util.Scheduler; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Random; +import java.util.Set; + +/** + * A manager to handle metadata related to remote clusters. It watches topic leader changes, + * topic partitions changes, and topic configuration changes in remote clusters and updates + * the local metadata accordingly. + */ +public class RemoteClusterMetadataManager implements AutoCloseable { + private static final Logger log = LoggerFactory.getLogger(RemoteClusterMetadataManager.class); + private final KafkaConfig brokerConfig; + private final int nodeId; + // Mapping from remote bootstrap servers to its corresponding broker sender and topics. + // TODO: A better key might be a cluster id or cluster-link. For now, we use remote bootstrap servers for demo. + private final Map remoteBrokers; + private final Map> topics; + private final Map> remoteClusterNodes; + private final Map> remotePartitionLeaders; + private final Metrics metrics; + private final Time time; + private final NodeToControllerChannelManager channelManager; + private final Random random; + private MetadataImage metadataImage; + + public RemoteClusterMetadataManager( + KafkaConfig config, + Metrics metrics, + Time time, + Scheduler scheduler, + NodeToControllerChannelManager channelManager + ) { + this.brokerConfig = config; + this.nodeId = config.nodeId(); + this.remoteBrokers = new HashMap<>(); + this.topics = new HashMap<>(); + this.remoteClusterNodes = new HashMap<>(); + this.remotePartitionLeaders = new HashMap<>(); + this.metrics = metrics; + this.time = time; + this.channelManager = channelManager; + this.metadataImage = MetadataImage.EMPTY; + this.random = new Random(); + } + + public void onMetadataUpdate(MetadataDelta delta, MetadataImage newImage) { + // TODO: Use ClusterLinkDelta to manage remote brokers / topics. + metadataImage = newImage; + if (delta.topicsDelta() == null) { + return; + } + var localChanges = delta.topicsDelta().localChanges(nodeId); + if (!localChanges.followers().isEmpty()) { + handleFollowerChanges(localChanges.followers()); + } + if (!localChanges.readOnlyLeaders().isEmpty()) { + handleReadOnlyLeadersChanges(localChanges.readOnlyLeaders()); + } + } + + @Override + public void close() throws Exception { + + } + + private void handleFollowerChanges(Map followers) { + followers.forEach((tp, info) -> { + var remoteBrokerTopics = topics.get(info.partition().remoteBootstrapServers); + if (remoteBrokerTopics != null) { + remoteBrokerTopics.remove(tp.topic()); + if (remoteBrokerTopics.isEmpty()) { + var sender = remoteBrokers.remove(info.partition().remoteBootstrapServers); + if (sender != null) { + sender.close(); + } + topics.remove(info.partition().remoteBootstrapServers); + } + } + }); + } + + private void handleReadOnlyLeadersChanges(Map readOnlyLeaders) { + var updateRemoteBootstrapServers = new HashSet(); + readOnlyLeaders.forEach((tp, info) -> { + remoteBrokers.computeIfAbsent( + info.partition().remoteBootstrapServers, + k -> { + var remoteBootstrapServers = Arrays.stream(k.split(",")).toList(); + var addresses = ClientUtils.parseAndValidateAddresses(remoteBootstrapServers, "use_all_dns_ips"); + // Use random node id here because we don't know node id of remote brokers. + var brokerEndpoint = new BrokerEndPoint(random.nextInt(), addresses.get(0).getHostString(), addresses.get(0).getPort()); + var logContext = new LogContext("[" + RemoteClusterMetadataManager.class.getName() + " replicaId=" + nodeId + ", remoteBootstrapServers=" + k + ", " + + "readOnly=true] "); + return new RemoteBrokerBlockingSender( + brokerEndpoint, + brokerConfig, + metrics, + time, + brokerEndpoint.id(), + "broker-" + nodeId + "-remote-cluster-metadata-manager-" + k.replace(":", "-"), + logContext + ); + }); + updateRemoteBootstrapServers.add(info.partition().remoteBootstrapServers); + topics.computeIfAbsent(info.partition().remoteBootstrapServers, k -> new HashSet<>()).add(tp.topic()); + }); + + log.info("!!! Updating remote cluster metadata for bootstrap servers: {}", updateRemoteBootstrapServers); + updateRemoteBootstrapServers.forEach(remoteBootstrapServers -> { + var sender = remoteBrokers.get(remoteBootstrapServers); + var updatedTopics = topics.get(remoteBootstrapServers); + var response = sender.sendRequest(MetadataRequest.Builder.forTopicNames(updatedTopics.stream().toList(), false)); + if (response.responseBody() instanceof MetadataResponse metadataResponse) { + log.info("!!! metadataResponse: {}", metadataResponse); + metadataResponse.brokers().forEach(broker -> { + remoteClusterNodes.computeIfAbsent(remoteBootstrapServers, k -> new HashMap<>()).put(broker.id(), broker); + }); + metadataResponse.topicMetadata().forEach(topicMetadata -> { + var partitionLeaders = remotePartitionLeaders.computeIfAbsent(remoteBootstrapServers, k -> new HashMap<>()); + topicMetadata.partitionMetadata().forEach(partitionMetadata -> { + partitionLeaders.put(partitionMetadata.topicPartition, remoteClusterNodes.get(remoteBootstrapServers).get(partitionMetadata.leaderId.get())); + }); + }); + } + }); + } + + public Node getRemotePartitionLeader(String remoteBootstrapServers, TopicPartition tp) { + var partitionLeaders = remotePartitionLeaders.get(remoteBootstrapServers); + if (partitionLeaders != null) { + return partitionLeaders.get(tp); + } + return null; + } + + public void refreshRemoteMetadata() { + remoteBrokers.forEach((remoteBootstrapServers, sender) -> { + var response = sender.sendRequest(MetadataRequest.Builder.forTopicNames(topics.get(remoteBootstrapServers).stream().toList(), false)); + if (response.responseBody() instanceof MetadataResponse metadataResponse) { + log.info("!!! periodic metadataResponse: {}", metadataResponse); + metadataResponse.brokers().forEach(broker -> { + remoteClusterNodes.computeIfAbsent(remoteBootstrapServers, k -> new HashMap<>()).put(broker.id(), broker); + }); + + var createPartitionsTopics = new CreatePartitionsRequestData.CreatePartitionsTopicCollection(); + metadataResponse.topicMetadata().forEach(topicMetadata -> { + var partitionLeaders = remotePartitionLeaders.computeIfAbsent(remoteBootstrapServers, k -> new HashMap<>()); + topicMetadata.partitionMetadata().forEach(partitionMetadata -> { + partitionLeaders.put(partitionMetadata.topicPartition, remoteClusterNodes.get(remoteBootstrapServers).get(partitionMetadata.leaderId.get())); + }); + + if (metadataImage.topics().getTopic(topicMetadata.topicId()) != null && + metadataImage.topics().getTopic(topicMetadata.topicId()).partitions().size() < partitionLeaders.size()) { + createPartitionsTopics.add(new CreatePartitionsRequestData.CreatePartitionsTopic() + .setName(topicMetadata.topic()) + .setCount(partitionLeaders.size()) + .setAssignments(null) + ); + } + }); + + if (!createPartitionsTopics.isEmpty()) { + log.info("!!! Detected partition count change, sending CreatePartitionsRequest: {}", createPartitionsTopics); + channelManager.sendRequest(new CreatePartitionsRequest.Builder( + new CreatePartitionsRequestData() + .setTopics(createPartitionsTopics) + .setValidateOnly(false) + .setTimeoutMs(3000) + ), new TimeoutHandler()); + } + } + }); + } + + private static class TimeoutHandler implements ControllerRequestCompletionHandler { + @Override + public void onTimeout() { + log.info("!!! Timed out"); + } + + @Override + public void onComplete(ClientResponse response) { + log.info("!!! Update topics: {}", response); + } + } +} diff --git a/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java b/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java index 5426d55a64da3..402d4dea5db44 100644 --- a/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java +++ b/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java @@ -129,6 +129,7 @@ public ReplicaManager build() { () -> -1L, Option.empty(), DirectoryEventHandler.NOOP, - new DelayedActionQueue()); + new DelayedActionQueue(), + Option.empty()); } } diff --git a/core/src/main/scala/kafka/server/AbstractFetcherManager.scala b/core/src/main/scala/kafka/server/AbstractFetcherManager.scala index 1cdd087aa87b7..c18c3767beb46 100755 --- a/core/src/main/scala/kafka/server/AbstractFetcherManager.scala +++ b/core/src/main/scala/kafka/server/AbstractFetcherManager.scala @@ -123,17 +123,17 @@ abstract class AbstractFetcherManager[T <: AbstractFetcherThread](val name: Stri } // to be defined in subclass to create a specific fetcher - def createFetcherThread(fetcherId: Int, sourceBroker: BrokerEndPoint): T + def createFetcherThread(fetcherId: Int, sourceBroker: BrokerEndPoint, readOnly: Boolean): T def addFetcherForPartitions(partitionAndOffsets: Map[TopicPartition, InitialFetchState]): Unit = { lock synchronized { val partitionsPerFetcher = partitionAndOffsets.groupBy { case (topicPartition, brokerAndInitialFetchOffset) => - BrokerAndFetcherId(brokerAndInitialFetchOffset.leader, getFetcherId(topicPartition)) + BrokerAndFetcherId(brokerAndInitialFetchOffset.leader, getFetcherId(topicPartition), brokerAndInitialFetchOffset.readOnly) } def addAndStartFetcherThread(brokerAndFetcherId: BrokerAndFetcherId, brokerIdAndFetcherId: BrokerIdAndFetcherId): T = { - val fetcherThread = createFetcherThread(brokerAndFetcherId.fetcherId, brokerAndFetcherId.broker) + val fetcherThread = createFetcherThread(brokerAndFetcherId.fetcherId, brokerAndFetcherId.broker, brokerAndFetcherId.readOnly) fetcherThreadMap.put(brokerIdAndFetcherId, fetcherThread) fetcherThread.start() fetcherThread @@ -263,7 +263,7 @@ class FailedPartitions { } } -case class BrokerAndFetcherId(broker: BrokerEndPoint, fetcherId: Int) +case class BrokerAndFetcherId(broker: BrokerEndPoint, fetcherId: Int, readOnly: Boolean = false) case class InitialFetchState(topicId: Option[Uuid], leader: BrokerEndPoint, currentLeaderEpoch: Int, initOffset: Long, readOnly: Boolean = false) diff --git a/core/src/main/scala/kafka/server/BrokerBlockingSender.scala b/core/src/main/scala/kafka/server/BrokerBlockingSender.scala index c8405be6c6e6f..4d85f0867b967 100644 --- a/core/src/main/scala/kafka/server/BrokerBlockingSender.scala +++ b/core/src/main/scala/kafka/server/BrokerBlockingSender.scala @@ -132,3 +132,74 @@ class BrokerBlockingSender(sourceBroker: BrokerEndPoint, s"BrokerBlockingSender(sourceBroker=$sourceBroker, fetcherId=$fetcherId)" } } + +class RemoteBrokerBlockingSender(sourceBroker: BrokerEndPoint, + brokerConfig: KafkaConfig, + metrics: Metrics, + time: Time, + fetcherId: Int, + clientId: String, + logContext: LogContext) extends BlockingSend { + + private val sourceNode = new Node(sourceBroker.id, sourceBroker.host, sourceBroker.port) + private val socketTimeout: Int = brokerConfig.replicaSocketTimeoutMs + + private val (networkClient, reconfigurableChannelBuilder) = { + + val channelBuilder = ClientUtils.createChannelBuilder(brokerConfig, time, logContext) + val selector = new Selector( + brokerConfig.getLong(CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_CONFIG), + metrics, time, "remote-readonly-" + clientId, channelBuilder, logContext) + val networkClient = new NetworkClient( + selector, + new ManualMetadataUpdater(), + clientId, + 1, + 0, + 0, + Selectable.USE_DEFAULT_BUFFER_SIZE, + brokerConfig.replicaSocketReceiveBufferBytes, + brokerConfig.requestTimeoutMs, + brokerConfig.connectionSetupTimeoutMs, + brokerConfig.connectionSetupTimeoutMaxMs, + time, + false, + new ApiVersions, + logContext, + MetadataRecoveryStrategy.NONE + ) + (networkClient, None) + } + + override def brokerEndPoint(): BrokerEndPoint = sourceBroker + + override def sendRequest(requestBuilder: Builder[_ <: AbstractRequest]): ClientResponse = { + try { + if (!NetworkClientUtils.awaitReady(networkClient, sourceNode, time, socketTimeout)) + throw new SocketTimeoutException(s"Failed to connect within $socketTimeout ms") + else { + val clientRequest = networkClient.newClientRequest(sourceBroker.id.toString, requestBuilder, + time.milliseconds(), true) + NetworkClientUtils.sendAndReceive(networkClient, clientRequest, time) + } + } + catch { + case e: Throwable => + networkClient.close(sourceBroker.id.toString) + throw e + } + } + + override def initiateClose(): Unit = { + reconfigurableChannelBuilder.foreach(brokerConfig.removeReconfigurable) + networkClient.initiateClose() + } + + def close(): Unit = { + networkClient.close() + } + + override def toString: String = { + s"RemoteBrokerBlockingSender(sourceBroker=$sourceBroker, fetcherId=$fetcherId)" + } +} diff --git a/core/src/main/scala/kafka/server/BrokerServer.scala b/core/src/main/scala/kafka/server/BrokerServer.scala index dfb4427b8eda3..ea48aa37d63ab 100644 --- a/core/src/main/scala/kafka/server/BrokerServer.scala +++ b/core/src/main/scala/kafka/server/BrokerServer.scala @@ -336,6 +336,15 @@ class BrokerServer( */ val defaultActionQueue = new DelayedActionQueue + val remoteClusterMetadataManager = new RemoteClusterMetadataManager( + config, + metrics, + time, + kafkaScheduler, + clientToControllerChannelManager + ) + kafkaScheduler.schedule("remote-cluster-metadata-refresh", () => remoteClusterMetadataManager.refreshRemoteMetadata(), 5000L, 30000L) + this._replicaManager = new ReplicaManager( config = config, metrics = metrics, @@ -352,7 +361,8 @@ class BrokerServer( brokerEpochSupplier = () => lifecycleManager.brokerEpoch, addPartitionsToTxnManager = Some(addPartitionsToTxnManager), directoryEventHandler = directoryEventHandler, - defaultActionQueue = defaultActionQueue + defaultActionQueue = defaultActionQueue, + remoteClusterMetadataManager = Some(remoteClusterMetadataManager) ) /* start token manager */ @@ -514,7 +524,8 @@ class BrokerServer( authorizerPlugin.toJava ), sharedServer.initialBrokerMetadataLoadFaultHandler, - sharedServer.metadataPublishingFaultHandler + sharedServer.metadataPublishingFaultHandler, + remoteClusterMetadataManager ) // If the BrokerLifecycleManager's initial catch-up future fails, it means we timed out // or are shutting down before we could catch up. Therefore, also fail the firstPublishFuture. diff --git a/core/src/main/scala/kafka/server/ReplicaAlterLogDirsManager.scala b/core/src/main/scala/kafka/server/ReplicaAlterLogDirsManager.scala index e0473166b365d..184c2990ee0cd 100644 --- a/core/src/main/scala/kafka/server/ReplicaAlterLogDirsManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaAlterLogDirsManager.scala @@ -34,7 +34,7 @@ class ReplicaAlterLogDirsManager(brokerConfig: KafkaConfig, clientId = "ReplicaAlterLogDirs", numFetchers = brokerConfig.getNumReplicaAlterLogDirsThreads) { - override def createFetcherThread(fetcherId: Int, sourceBroker: BrokerEndPoint): ReplicaAlterLogDirsThread = { + override def createFetcherThread(fetcherId: Int, sourceBroker: BrokerEndPoint, readOnly: Boolean): ReplicaAlterLogDirsThread = { val threadName = s"ReplicaAlterLogDirsThread-$fetcherId" val leader: LeaderEndPoint = new LocalLeaderEndPoint(sourceBroker, brokerConfig, replicaManager, quotaManager) new ReplicaAlterLogDirsThread(threadName, leader, failedPartitions, replicaManager, diff --git a/core/src/main/scala/kafka/server/ReplicaFetcherManager.scala b/core/src/main/scala/kafka/server/ReplicaFetcherManager.scala index 2b02b4a72133f..77eb6f6b8445e 100644 --- a/core/src/main/scala/kafka/server/ReplicaFetcherManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaFetcherManager.scala @@ -36,12 +36,17 @@ class ReplicaFetcherManager(brokerConfig: KafkaConfig, clientId = "Replica", numFetchers = brokerConfig.numReplicaFetchers) { - override def createFetcherThread(fetcherId: Int, sourceBroker: BrokerEndPoint): ReplicaFetcherThread = { - info("!!! createFetcherThread: sourceBroker = " + sourceBroker + " fetcherId = " + fetcherId) + override def createFetcherThread(fetcherId: Int, sourceBroker: BrokerEndPoint, readOnly: Boolean): ReplicaFetcherThread = { + info("!!! createFetcherThread: sourceBroker = " + sourceBroker + " fetcherId = " + fetcherId + " readOnly = " + readOnly) val threadName = s"ReplicaFetcherThread-$fetcherId-${sourceBroker.id}" val logContext = new LogContext(s"[ReplicaFetcher replicaId=${brokerConfig.brokerId}, leaderId=${sourceBroker.id}, " + - s"fetcherId=$fetcherId] ") - val endpoint = new BrokerBlockingSender(sourceBroker, brokerConfig, metrics, time, fetcherId, + s"fetcherId=$fetcherId, readOnly=$readOnly] ") + + val endpoint = if (readOnly) + new RemoteBrokerBlockingSender(sourceBroker, brokerConfig, metrics, time, fetcherId, + s"broker-${brokerConfig.brokerId}-remote-fetcher-$fetcherId", logContext) + else + new BrokerBlockingSender(sourceBroker, brokerConfig, metrics, time, fetcherId, s"broker-${brokerConfig.brokerId}-fetcher-$fetcherId", logContext) val fetchSessionHandler = new FetchSessionHandler(logContext, sourceBroker.id) val leader: LeaderEndPoint = new RemoteLeaderEndPoint(logContext.logPrefix, endpoint, fetchSessionHandler, brokerConfig, diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index 7dcec414f0f01..82be96a392e8b 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaManager.scala @@ -26,7 +26,6 @@ import kafka.server.ReplicaManager.{AtMinIsrPartitionCountMetricName, FailedIsrU import kafka.server.share.DelayedShareFetch import kafka.utils._ import org.apache.kafka.common.config.{ConfigResource, TopicConfig} -import org.apache.kafka.clients.{ClientUtils} import org.apache.kafka.common.{IsolationLevel, Node, TopicIdPartition, TopicPartition, Uuid} import org.apache.kafka.common.errors._ import org.apache.kafka.common.internals.{Plugin, Topic} @@ -230,7 +229,8 @@ class ReplicaManager(val config: KafkaConfig, val brokerEpochSupplier: () => Long = () => -1, addPartitionsToTxnManager: Option[AddPartitionsToTxnManager] = None, val directoryEventHandler: DirectoryEventHandler = DirectoryEventHandler.NOOP, - val defaultActionQueue: ActionQueue = new DelayedActionQueue + val defaultActionQueue: ActionQueue = new DelayedActionQueue, + val remoteClusterMetadataManager: Option[RemoteClusterMetadataManager] = None ) extends Logging { private val metricsGroup = new KafkaMetricsGroup(this.getClass) private val addPartitionsToTxnConfig = new AddPartitionsToTxnConfig(config) @@ -2541,36 +2541,12 @@ class ReplicaManager(val config: KafkaConfig, // TODO: the remote leader host should get from metadata request, currently, using localhost:9092 partitionsToStartFetching.foreachEntry { (topicPartition, partition) => -// val logContext = new LogContext(String.format("[remote follower clientId=%s] ", localBrokerId)) -//// val metadata = new Metadata(100, 1000, 300000, logContext, new ClusterResourceListeners) -// val addresses = ClientUtils.parseAndValidateAddresses(remoteBootstrapServers, "use_all_dns_ips") -//// -//// metadata.bootstrap(addresses) -// val metadataManager: AdminMetadataManager = new AdminMetadataManager(logContext, 100, 1000, false) -// metadataManager.update(Cluster.bootstrap(addresses), time.milliseconds) -// -// val networkClient = ClientUtils.createNetworkClient(config, "clientId", metrics, "admin-client", logContext, new ApiVersions, time, 1, TimeUnit.HOURS.toMillis(1).toInt, null, metadataManager.updater, new DefaultHostResolver -// , null, null) -// -// val requestBuilder = new MetadataRequest.Builder( -// new MetadataRequestData().setTopics(util.Arrays.asList(new MetadataRequestData.MetadataRequestTopic().setName(partition.topic))).setAllowAutoTopicCreation(true)) -// val clientRequest = networkClient.newClientRequest("3000", requestBuilder, -// time.milliseconds(), true) -// val response: MetadataResponse = NetworkClientUtils.sendAndReceive(networkClient, clientRequest, time) -// val now = time.milliseconds -// -// if (response.topLevelError eq Errors.REBOOTSTRAP_REQUIRED) metadataManager.initiateRebootstrap() -// else metadataManager.update(response.buildCluster, now) -// metadataManager.fetchMetadataSnapshot().cluster.leaderFor(topicPartition) - val nodeOpt = if (!remoteLeader) partition.leaderReplicaIdOpt .flatMap(leaderId => Option(newImage.cluster.broker(leaderId))) .flatMap(_.node(listenerName).toScala) else { - val remoteBootstrapServers = partition.remoteBootstrapServer.split(",").map(_.trim).toList.asJava - val addresses = ClientUtils.parseAndValidateAddresses(remoteBootstrapServers, "use_all_dns_ips") - Option.apply(new Node(3000, addresses.get(0).getHostString, addresses.get(0).getPort)) + Some(remoteClusterMetadataManager.get.getRemotePartitionLeader(partition.remoteBootstrapServer, partition.topicPartition)) } nodeOpt match { diff --git a/core/src/main/scala/kafka/server/metadata/BrokerMetadataPublisher.scala b/core/src/main/scala/kafka/server/metadata/BrokerMetadataPublisher.scala index 7876163f9ea98..62d1fffa2e2db 100644 --- a/core/src/main/scala/kafka/server/metadata/BrokerMetadataPublisher.scala +++ b/core/src/main/scala/kafka/server/metadata/BrokerMetadataPublisher.scala @@ -21,7 +21,7 @@ import java.util.OptionalInt import kafka.coordinator.transaction.TransactionCoordinator import kafka.log.LogManager import kafka.server.share.SharePartitionManager -import kafka.server.{KafkaConfig, ReplicaManager} +import kafka.server.{KafkaConfig, ReplicaManager, RemoteClusterMetadataManager} import kafka.utils.Logging import org.apache.kafka.common.TopicPartition import org.apache.kafka.common.errors.TimeoutException @@ -83,7 +83,8 @@ class BrokerMetadataPublisher( delegationTokenPublisher: DelegationTokenPublisher, aclPublisher: AclPublisher, fatalFaultHandler: FaultHandler, - metadataPublishingFaultHandler: FaultHandler + metadataPublishingFaultHandler: FaultHandler, + remoteClusterMetadataManager: RemoteClusterMetadataManager ) extends MetadataPublisher with Logging { logIdent = s"[BrokerMetadataPublisher id=${config.nodeId}] " @@ -143,6 +144,8 @@ class BrokerMetadataPublisher( debug(s"Publishing metadata at offset $highestOffsetAndEpoch with $metadataVersionLogMsg.") } + remoteClusterMetadataManager.onMetadataUpdate(delta, newImage) + // Apply topic deltas. Option(delta.topicsDelta()).foreach { topicsDelta => try { diff --git a/core/src/test/scala/unit/kafka/server/AbstractFetcherManagerTest.scala b/core/src/test/scala/unit/kafka/server/AbstractFetcherManagerTest.scala index 20ea24e0bb4fc..95c7686771047 100644 --- a/core/src/test/scala/unit/kafka/server/AbstractFetcherManagerTest.scala +++ b/core/src/test/scala/unit/kafka/server/AbstractFetcherManagerTest.scala @@ -59,7 +59,7 @@ class AbstractFetcherManagerTest { def testAddAndRemovePartition(): Unit = { val fetcher: AbstractFetcherThread = mock(classOf[AbstractFetcherThread]) val fetcherManager = new AbstractFetcherManager[AbstractFetcherThread]("fetcher-manager", "fetcher-manager", 2) { - override def createFetcherThread(fetcherId: Int, sourceBroker: BrokerEndPoint): AbstractFetcherThread = { + override def createFetcherThread(fetcherId: Int, sourceBroker: BrokerEndPoint, readOnly: Boolean): AbstractFetcherThread = { fetcher } } @@ -96,7 +96,7 @@ class AbstractFetcherManagerTest { def testMetricFailedPartitionCount(): Unit = { val fetcher: AbstractFetcherThread = mock(classOf[AbstractFetcherThread]) val fetcherManager = new AbstractFetcherManager[AbstractFetcherThread]("fetcher-manager", "fetcher-manager", 2) { - override def createFetcherThread(fetcherId: Int, sourceBroker: BrokerEndPoint): AbstractFetcherThread = { + override def createFetcherThread(fetcherId: Int, sourceBroker: BrokerEndPoint, readOnly: Boolean): AbstractFetcherThread = { fetcher } } @@ -121,7 +121,7 @@ class AbstractFetcherManagerTest { def testDeadThreadCountMetric(): Unit = { val fetcher: AbstractFetcherThread = mock(classOf[AbstractFetcherThread]) val fetcherManager = new AbstractFetcherManager[AbstractFetcherThread]("fetcher-manager", "fetcher-manager", 2) { - override def createFetcherThread(fetcherId: Int, sourceBroker: BrokerEndPoint): AbstractFetcherThread = { + override def createFetcherThread(fetcherId: Int, sourceBroker: BrokerEndPoint, readOnly: Boolean): AbstractFetcherThread = { fetcher } } @@ -155,7 +155,7 @@ class AbstractFetcherManagerTest { def testMaybeUpdateTopicIds(): Unit = { val fetcher: AbstractFetcherThread = mock(classOf[AbstractFetcherThread]) val fetcherManager = new AbstractFetcherManager[AbstractFetcherThread]("fetcher-manager", "fetcher-manager", 2) { - override def createFetcherThread(fetcherId: Int, sourceBroker: BrokerEndPoint): AbstractFetcherThread = { + override def createFetcherThread(fetcherId: Int, sourceBroker: BrokerEndPoint, readOnly: Boolean): AbstractFetcherThread = { fetcher } } @@ -242,7 +242,7 @@ class AbstractFetcherManagerTest { val fetchingTopicPartitions = makeTopicPartition(10, 100) val failedTopicPartitions = makeTopicPartition(2, 5, "topic_failed") val fetcherManager = new AbstractFetcherManager[AbstractFetcherThread]("fetcher-manager", "fetcher-manager", currentFetcherSize) { - override def createFetcherThread(fetcherId: Int, sourceBroker: BrokerEndPoint): AbstractFetcherThread = { + override def createFetcherThread(fetcherId: Int, sourceBroker: BrokerEndPoint, readOnly: Boolean): AbstractFetcherThread = { new TestResizeFetcherThread(sourceBroker, failedPartitions, new MockResizeFetcherTierStateMachine) } } diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala index 10a16656836c0..d96bba4d96335 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala @@ -2623,7 +2623,7 @@ class ReplicaManagerTest { val rm = this new ReplicaFetcherManager(this.config, rm, metrics, time, replicationQuotaManager, () => this.metadataCache.metadataVersion(), () => 1) { - override def createFetcherThread(fetcherId: Int, sourceBroker: BrokerEndPoint): ReplicaFetcherThread = { + override def createFetcherThread(fetcherId: Int, sourceBroker: BrokerEndPoint, readOnly: Boolean): ReplicaFetcherThread = { val logContext = new LogContext(s"[ReplicaFetcher replicaId=${rm.config.brokerId}, leaderId=${sourceBroker.id}, " + s"fetcherId=$fetcherId] ") val fetchSessionHandler = new FetchSessionHandler(logContext, sourceBroker.id) @@ -3069,7 +3069,7 @@ class ReplicaManagerTest { val config = this.config val metadataCache = this.metadataCache new ReplicaFetcherManager(config, this, metrics, time, quotaManager, () => metadataCache.metadataVersion(), () => 1) { - override def createFetcherThread(fetcherId: Int, sourceBroker: BrokerEndPoint): ReplicaFetcherThread = { + override def createFetcherThread(fetcherId: Int, sourceBroker: BrokerEndPoint, readOnly: Boolean): ReplicaFetcherThread = { val threadName = s"ReplicaFetcherThread-$fetcherId-${sourceBroker.id}" val tp = new TopicPartition(topic, 0) diff --git a/core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataPublisherTest.scala b/core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataPublisherTest.scala index 3bb3f5fc3f77e..420ffb2b7d7d6 100644 --- a/core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataPublisherTest.scala +++ b/core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataPublisherTest.scala @@ -24,7 +24,7 @@ import java.util.{OptionalInt, Properties} import java.util.concurrent.atomic.{AtomicInteger, AtomicReference} import kafka.log.LogManager import kafka.server.share.SharePartitionManager -import kafka.server.{BrokerServer, KafkaConfig, ReplicaManager} +import kafka.server.{BrokerServer, KafkaConfig, RemoteClusterMetadataManager, ReplicaManager} import kafka.utils.TestUtils import org.apache.kafka.clients.admin.AlterConfigOp.OpType.SET import org.apache.kafka.clients.admin.{Admin, AlterConfigOp, ConfigEntry, NewTopic} @@ -211,7 +211,8 @@ class BrokerMetadataPublisherTest { mock(classOf[DelegationTokenPublisher]), mock(classOf[AclPublisher]), faultHandler, - faultHandler + faultHandler, + mock(classOf[RemoteClusterMetadataManager]) ) val topicId = Uuid.randomUuid() @@ -276,6 +277,7 @@ class BrokerMetadataPublisherTest { mock(classOf[AclPublisher]), faultHandler, faultHandler, + mock(classOf[RemoteClusterMetadataManager]) ) val image = MetadataImage.EMPTY @@ -316,7 +318,8 @@ class BrokerMetadataPublisherTest { mock(classOf[DelegationTokenPublisher]), mock(classOf[AclPublisher]), faultHandler, - faultHandler + faultHandler, + mock(classOf[RemoteClusterMetadataManager]) ) val featuresImage = new FeaturesImage(