Add remote cluster metadata manager

clients/src/main/java/org/apache/kafka/clients/ClientUtils.java

@@ -117,6 +117,7 @@ public static List<InetSocketAddress> parseAndValidateAddresses(List<String> 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);
     }

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)

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<String, RemoteBrokerBlockingSender> remoteBrokers;
+    private final Map<String, Set<String>> topics;
+    private final Map<String, Map<Integer, Node>> remoteClusterNodes;
+    private final Map<String, Map<TopicPartition, Node>> 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<TopicPartition, LocalReplicaChanges.PartitionInfo> 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<TopicPartition, LocalReplicaChanges.PartitionInfo> readOnlyLeaders) {
+        var updateRemoteBootstrapServers = new HashSet<String>();
+        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);
+        }
+    }
+}

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());
     }
 }

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)
 

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