You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by cm...@apache.org on 2022/05/12 21:48:41 UTC
[kafka] branch trunk updated: KAFKA-13649: Implement early.start.listeners and fix StandardAuthorizer loading (#11969)
This is an automated email from the ASF dual-hosted git repository.
cmccabe pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new fa59be4e770 KAFKA-13649: Implement early.start.listeners and fix StandardAuthorizer loading (#11969)
fa59be4e770 is described below
commit fa59be4e770627cd34cef85986b58ad7f606928d
Author: Colin Patrick McCabe <cm...@apache.org>
AuthorDate: Thu May 12 14:48:33 2022 -0700
KAFKA-13649: Implement early.start.listeners and fix StandardAuthorizer loading (#11969)
Since the StandardAuthorizer relies on the metadata log to store its ACLs, we need to be sure that
we have the latest metadata before allowing the authorizer to be used. However, if the authorizer
is not usable for controllers in the cluster, the latest metadata cannot be fetched, because
inter-node communication cannot occur. In the initial commit which introduced StandardAuthorizer,
we punted on the loading issue by allowing the authorizer to be used immediately. This commit fixes
that by implementing early.start.listeners as specified in KIP-801. This will allow in superusers
immediately, but throw the new AuthorizerNotReadyException if non-superusers try to use the
authorizer before StandardAuthorizer#completeInitialLoad is called.
For the broker, we call StandardAuthorizer#completeInitialLoad immediately after metadata catch-up
is complete, right before unfencing. For the controller, we call
StandardAuthorizer#completeInitialLoad when the node has caught up to the high water mark of the
cluster metadata partition.
This PR refactors the SocketServer so that it creates the configured acceptors and processors in
its constructor, rather than requiring a call to SocketServer#startup A new function,
SocketServer#enableRequestProcessing, then starts the threads and begins listening on the
configured ports. enableRequestProcessing uses an async model: we will start the acceptor and
processors associated with an endpoint as soon as that endpoint's authorizer future is completed.
Also fix a bug where the controller and listener were sharing an Authorizer when in co-located
mode, which was not intended.
Reviewers: Jason Gustafson <ja...@confluent.io>
---
.../common/errors/AuthorizerNotReadyException.java | 28 ++
.../server/authorizer/AuthorizerServerInfo.java | 5 +
core/src/main/scala/kafka/cluster/Broker.scala | 6 +-
core/src/main/scala/kafka/cluster/EndPoint.scala | 6 +
.../main/scala/kafka/network/SocketServer.scala | 533 +++++++++------------
.../kafka/server/BrokerLifecycleManager.scala | 10 +-
.../src/main/scala/kafka/server/BrokerServer.scala | 42 +-
.../main/scala/kafka/server/ControllerServer.scala | 23 +-
core/src/main/scala/kafka/server/KafkaConfig.scala | 29 +-
core/src/main/scala/kafka/server/KafkaServer.scala | 5 +-
.../main/scala/kafka/tools/TestRaftServer.scala | 4 +-
.../unit/kafka/network/SocketServerTest.scala | 63 +--
.../scala/unit/kafka/server/KafkaConfigTest.scala | 48 +-
.../apache/kafka/controller/QuorumController.java | 40 +-
.../authorizer/ClusterMetadataAuthorizer.java | 12 +
.../metadata/authorizer/StandardAuthorizer.java | 31 +-
.../authorizer/StandardAuthorizerData.java | 24 +
.../kafka/controller/AclControlManagerTest.java | 10 +
.../kafka/controller/QuorumControllerTest.java | 49 ++
.../authorizer/ClusterMetadataAuthorizerTest.java | 10 +
.../authorizer/StandardAuthorizerTest.java | 170 ++++++-
.../org/apache/kafka/metalog/LocalLogManager.java | 41 +-
.../kafka/metalog/LocalLogManagerTestEnv.java | 34 +-
.../java/org/apache/kafka/raft/RaftClient.java | 6 +
24 files changed, 837 insertions(+), 392 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/common/errors/AuthorizerNotReadyException.java b/clients/src/main/java/org/apache/kafka/common/errors/AuthorizerNotReadyException.java
new file mode 100644
index 00000000000..1c110ef2143
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/errors/AuthorizerNotReadyException.java
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.errors;
+
+/**
+ * An exception that indicates that the authorizer is not ready to receive the request yet.
+ */
+public class AuthorizerNotReadyException extends RetriableException {
+ private static final long serialVersionUID = 1L;
+
+ public AuthorizerNotReadyException() {
+ super();
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/server/authorizer/AuthorizerServerInfo.java b/clients/src/main/java/org/apache/kafka/server/authorizer/AuthorizerServerInfo.java
index 51e23fba57f..eb03c117b6c 100644
--- a/clients/src/main/java/org/apache/kafka/server/authorizer/AuthorizerServerInfo.java
+++ b/clients/src/main/java/org/apache/kafka/server/authorizer/AuthorizerServerInfo.java
@@ -48,4 +48,9 @@ public interface AuthorizerServerInfo {
* Returns the inter-broker endpoint. This is one of the endpoints returned by {@link #endpoints()}.
*/
Endpoint interBrokerEndpoint();
+
+ /**
+ * Returns the configured early start listeners.
+ */
+ Collection<String> earlyStartListeners();
}
diff --git a/core/src/main/scala/kafka/cluster/Broker.scala b/core/src/main/scala/kafka/cluster/Broker.scala
index 657d89b8fe7..9b1d741835c 100755
--- a/core/src/main/scala/kafka/cluster/Broker.scala
+++ b/core/src/main/scala/kafka/cluster/Broker.scala
@@ -35,7 +35,8 @@ object Broker {
private[kafka] case class ServerInfo(clusterResource: ClusterResource,
brokerId: Int,
endpoints: util.List[Endpoint],
- interBrokerEndpoint: Endpoint) extends AuthorizerServerInfo
+ interBrokerEndpoint: Endpoint,
+ earlyStartListeners: util.Set[String]) extends AuthorizerServerInfo
def apply(id: Int, endPoints: Seq[EndPoint], rack: Option[String]): Broker = {
new Broker(id, endPoints, rack, emptySupportedFeatures)
@@ -93,6 +94,7 @@ case class Broker(id: Int, endPoints: Seq[EndPoint], rack: Option[String], featu
val clusterResource: ClusterResource = new ClusterResource(clusterId)
val interBrokerEndpoint: Endpoint = endPoint(config.interBrokerListenerName).toJava
val brokerEndpoints: util.List[Endpoint] = endPoints.toList.map(_.toJava).asJava
- Broker.ServerInfo(clusterResource, id, brokerEndpoints, interBrokerEndpoint)
+ Broker.ServerInfo(clusterResource, id, brokerEndpoints, interBrokerEndpoint,
+ config.earlyStartListeners.map(_.value()).asJava)
}
}
diff --git a/core/src/main/scala/kafka/cluster/EndPoint.scala b/core/src/main/scala/kafka/cluster/EndPoint.scala
index 3e84f9ed834..89c9f5ec3d4 100644
--- a/core/src/main/scala/kafka/cluster/EndPoint.scala
+++ b/core/src/main/scala/kafka/cluster/EndPoint.scala
@@ -65,6 +65,12 @@ object EndPoint {
case _ => throw new KafkaException(s"Unable to parse a listener name from $connectionString")
}
}
+
+ def fromJava(endpoint: JEndpoint): EndPoint =
+ new EndPoint(endpoint.host(),
+ endpoint.port(),
+ new ListenerName(endpoint.listenerName().get()),
+ endpoint.securityProtocol())
}
/**
diff --git a/core/src/main/scala/kafka/network/SocketServer.scala b/core/src/main/scala/kafka/network/SocketServer.scala
index 88dfa15b3f5..e91c240415c 100644
--- a/core/src/main/scala/kafka/network/SocketServer.scala
+++ b/core/src/main/scala/kafka/network/SocketServer.scala
@@ -22,7 +22,6 @@ import java.net._
import java.nio.ByteBuffer
import java.nio.channels.{Selector => NSelector, _}
import java.util
-import java.util.Optional
import java.util.concurrent._
import java.util.concurrent.atomic._
@@ -34,7 +33,7 @@ import kafka.network.RequestChannel.{CloseConnectionResponse, EndThrottlingRespo
import kafka.network.SocketServer._
import kafka.security.CredentialProvider
import kafka.server.{ApiVersionManager, BrokerReconfigurable, KafkaConfig}
-import kafka.utils.Implicits._
+import org.apache.kafka.common.message.ApiMessageType.ListenerType
import kafka.utils._
import org.apache.kafka.common.config.ConfigException
import org.apache.kafka.common.config.internals.QuotaConfigs
@@ -104,51 +103,34 @@ class SocketServer(val config: KafkaConfig,
private[this] val nextProcessorId: AtomicInteger = new AtomicInteger(0)
val connectionQuotas = new ConnectionQuotas(config, time, metrics)
- private var startedProcessingRequests = false
- private var stoppedProcessingRequests = false
- // Processors are now created by each Acceptor. However to preserve compatibility, we need to number the processors
- // globally, so we keep the nextProcessorId counter in SocketServer
- def nextProcessorId(): Int = {
- nextProcessorId.getAndIncrement()
- }
+ /**
+ * A future which is completed once all the authorizer futures are complete.
+ */
+ private val allAuthorizerFuturesComplete = new CompletableFuture[Void]
/**
- * Starts the socket server and creates all the Acceptors and the Processors. The Acceptors
- * start listening at this stage so that the bound port is known when this method completes
- * even when ephemeral ports are used. Acceptors and Processors are started if `startProcessingRequests`
- * is true. If not, acceptors and processors are only started when [[kafka.network.SocketServer#startProcessingRequests()]]
- * is invoked. Delayed starting of acceptors and processors is used to delay processing client
- * connections until server is fully initialized, e.g. to ensure that all credentials have been
- * loaded before authentications are performed. Incoming connections on this server are processed
- * when processors start up and invoke [[org.apache.kafka.common.network.Selector#poll]].
- *
- * @param startProcessingRequests Flag indicating whether `Processor`s must be started.
- * @param controlPlaneListener The control plane listener, or None if there is none.
- * @param dataPlaneListeners The data plane listeners.
+ * True if the SocketServer is stopped. Must be accessed under the SocketServer lock.
*/
- def startup(startProcessingRequests: Boolean = true,
- controlPlaneListener: Option[EndPoint] = config.controlPlaneListener,
- dataPlaneListeners: Seq[EndPoint] = config.dataPlaneListeners): Unit = {
- this.synchronized {
- createControlPlaneAcceptorAndProcessor(controlPlaneListener)
- createDataPlaneAcceptorsAndProcessors(dataPlaneListeners)
- if (startProcessingRequests) {
- this.startProcessingRequests()
- }
- }
+ private var stopped = false
+ // Socket server metrics
+ newGauge(s"${DataPlaneAcceptor.MetricPrefix}NetworkProcessorAvgIdlePercent", () => SocketServer.this.synchronized {
val dataPlaneProcessors = dataPlaneAcceptors.asScala.values.flatMap(a => a.processors)
- val controlPlaneProcessorOpt = controlPlaneAcceptorOpt.map(a => a.processors(0))
- newGauge(s"${DataPlaneAcceptor.MetricPrefix}NetworkProcessorAvgIdlePercent", () => SocketServer.this.synchronized {
- val ioWaitRatioMetricNames = dataPlaneProcessors.map { p =>
- metrics.metricName("io-wait-ratio", MetricsGroup, p.metricTags)
- }
+ val ioWaitRatioMetricNames = dataPlaneProcessors.map { p =>
+ metrics.metricName("io-wait-ratio", MetricsGroup, p.metricTags)
+ }
+ if (dataPlaneProcessors.isEmpty) {
+ 1.0
+ } else {
ioWaitRatioMetricNames.map { metricName =>
Option(metrics.metric(metricName)).fold(0.0)(m => Math.min(m.metricValue.asInstanceOf[Double], 1.0))
}.sum / dataPlaneProcessors.size
- })
+ }
+ })
+ if (config.requiresZookeeper) {
newGauge(s"${ControlPlaneAcceptor.MetricPrefix}NetworkProcessorAvgIdlePercent", () => SocketServer.this.synchronized {
+ val controlPlaneProcessorOpt = controlPlaneAcceptorOpt.map(a => a.processors(0))
val ioWaitRatioMetricName = controlPlaneProcessorOpt.map { p =>
metrics.metricName("io-wait-ratio", MetricsGroup, p.metricTags)
}
@@ -156,17 +138,21 @@ class SocketServer(val config: KafkaConfig,
Option(metrics.metric(metricName)).fold(0.0)(m => Math.min(m.metricValue.asInstanceOf[Double], 1.0))
}.getOrElse(Double.NaN)
})
- newGauge("MemoryPoolAvailable", () => memoryPool.availableMemory)
- newGauge("MemoryPoolUsed", () => memoryPool.size() - memoryPool.availableMemory)
- newGauge(s"${DataPlaneAcceptor.MetricPrefix}ExpiredConnectionsKilledCount", () => SocketServer.this.synchronized {
- val expiredConnectionsKilledCountMetricNames = dataPlaneProcessors.map { p =>
- metrics.metricName("expired-connections-killed-count", MetricsGroup, p.metricTags)
- }
- expiredConnectionsKilledCountMetricNames.map { metricName =>
- Option(metrics.metric(metricName)).fold(0.0)(m => m.metricValue.asInstanceOf[Double])
- }.sum
- })
+ }
+ newGauge("MemoryPoolAvailable", () => memoryPool.availableMemory)
+ newGauge("MemoryPoolUsed", () => memoryPool.size() - memoryPool.availableMemory)
+ newGauge(s"${DataPlaneAcceptor.MetricPrefix}ExpiredConnectionsKilledCount", () => SocketServer.this.synchronized {
+ val dataPlaneProcessors = dataPlaneAcceptors.asScala.values.flatMap(a => a.processors)
+ val expiredConnectionsKilledCountMetricNames = dataPlaneProcessors.map { p =>
+ metrics.metricName("expired-connections-killed-count", MetricsGroup, p.metricTags)
+ }
+ expiredConnectionsKilledCountMetricNames.map { metricName =>
+ Option(metrics.metric(metricName)).fold(0.0)(m => m.metricValue.asInstanceOf[Double])
+ }.sum
+ })
+ if (config.requiresZookeeper) {
newGauge(s"${ControlPlaneAcceptor.MetricPrefix}ExpiredConnectionsKilledCount", () => SocketServer.this.synchronized {
+ val controlPlaneProcessorOpt = controlPlaneAcceptorOpt.map(a => a.processors(0))
val expiredConnectionsKilledCountMetricNames = controlPlaneProcessorOpt.map { p =>
metrics.metricName("expired-connections-killed-count", MetricsGroup, p.metricTags)
}
@@ -176,112 +162,86 @@ class SocketServer(val config: KafkaConfig,
})
}
- /**
- * Start processing requests and new connections. This method is used for delayed starting of
- * all the acceptors and processors if [[kafka.network.SocketServer#startup]] was invoked with
- * `startProcessingRequests=false`.
- *
- * Before starting processors for each endpoint, we ensure that authorizer has all the metadata
- * to authorize requests on that endpoint by waiting on the provided future. We start inter-broker
- * listener before other listeners. This allows authorization metadata for other listeners to be
- * stored in Kafka topics in this cluster.
- *
- * @param authorizerFutures Future per [[EndPoint]] used to wait before starting the processor
- * corresponding to the [[EndPoint]]
- */
- def startProcessingRequests(authorizerFutures: Map[Endpoint, CompletableFuture[Void]] = Map.empty): Unit = {
- info("Starting socket server acceptors and processors")
- this.synchronized {
- if (!startedProcessingRequests) {
- startControlPlaneProcessorAndAcceptor(authorizerFutures)
- startDataPlaneProcessorsAndAcceptors(authorizerFutures)
- startedProcessingRequests = true
- } else {
- info("Socket server acceptors and processors already started")
- }
- }
- info("Started socket server acceptors and processors")
+ // Create acceptors and processors for the statically configured endpoints when the
+ // SocketServer is constructed. Note that this just opens the ports and creates the data
+ // structures. It does not start the acceptors and processors or their associated JVM
+ // threads.
+ if (apiVersionManager.listenerType.equals(ListenerType.CONTROLLER)) {
+ config.controllerListeners.foreach(createDataPlaneAcceptorAndProcessors)
+ } else {
+ config.controlPlaneListener.foreach(createControlPlaneAcceptorAndProcessor)
+ config.dataPlaneListeners.foreach(createDataPlaneAcceptorAndProcessors)
}
- /**
- * Starts processors of the provided acceptor and the acceptor itself.
- *
- * Before starting them, we ensure that authorizer has all the metadata to authorize
- * requests on that endpoint by waiting on the provided future.
- */
- private def startAcceptorAndProcessors(acceptor: Acceptor,
- authorizerFutures: Map[Endpoint, CompletableFuture[Void]] = Map.empty): Unit = {
- val endpoint = acceptor.endPoint
- debug(s"Wait for authorizer to complete start up on listener ${endpoint.listenerName}")
- waitForAuthorizerFuture(acceptor, authorizerFutures)
- debug(s"Start processors on listener ${endpoint.listenerName}")
- acceptor.startProcessors()
- debug(s"Start acceptor thread on listener ${endpoint.listenerName}")
- if (!acceptor.isStarted()) {
- KafkaThread.nonDaemon(
- s"${acceptor.threadPrefix()}-kafka-socket-acceptor-${endpoint.listenerName}-${endpoint.securityProtocol}-${endpoint.port}",
- acceptor
- ).start()
- acceptor.awaitStartup()
- }
- info(s"Started ${acceptor.threadPrefix()} acceptor and processor(s) for endpoint : ${endpoint.listenerName}")
+ // Processors are now created by each Acceptor. However to preserve compatibility, we need to number the processors
+ // globally, so we keep the nextProcessorId counter in SocketServer
+ def nextProcessorId(): Int = {
+ nextProcessorId.getAndIncrement()
}
/**
- * Starts processors of all the data-plane acceptors and all the acceptors of this server.
+ * This method enables request processing for all endpoints managed by this SocketServer. Each
+ * endpoint will be brought up asynchronously as soon as its associated future is completed.
+ * Therefore, we do not know that any particular request processor will be running by the end of
+ * this function -- just that it might be running.
*
- * We start inter-broker listener before other listeners. This allows authorization metadata for
- * other listeners to be stored in Kafka topics in this cluster.
+ * @param authorizerFutures Future per [[EndPoint]] used to wait before starting the
+ * processor corresponding to the [[EndPoint]]. Any endpoint
+ * that does not appear in this map will be started once all
+ * authorizerFutures are complete.
*/
- private def startDataPlaneProcessorsAndAcceptors(authorizerFutures: Map[Endpoint, CompletableFuture[Void]]): Unit = {
- val interBrokerListener = dataPlaneAcceptors.asScala.keySet
- .find(_.listenerName == config.interBrokerListenerName)
- val orderedAcceptors = interBrokerListener match {
- case Some(interBrokerListener) => List(dataPlaneAcceptors.get(interBrokerListener)) ++
- dataPlaneAcceptors.asScala.filter { case (k, _) => k != interBrokerListener }.values
- case None => dataPlaneAcceptors.asScala.values
- }
- orderedAcceptors.foreach { acceptor =>
- startAcceptorAndProcessors(acceptor, authorizerFutures)
+ def enableRequestProcessing(
+ authorizerFutures: Map[Endpoint, CompletableFuture[Void]]
+ ): Unit = this.synchronized {
+ if (stopped) {
+ throw new RuntimeException("Can't enable request processing: SocketServer is stopped.")
+ }
+
+ def chainAcceptorFuture(acceptor: Acceptor): Unit = {
+ // Because of ephemeral ports, we need to match acceptors to futures by looking at
+ // the listener name, rather than the endpoint object.
+ authorizerFutures.find {
+ case (endpoint, _) => acceptor.endPoint.listenerName.value().equals(endpoint.listenerName().get())
+ } match {
+ case None => chainFuture(allAuthorizerFuturesComplete, acceptor.startFuture)
+ case Some((_, future)) => chainFuture(future, acceptor.startFuture)
+ }
}
- }
- /**
- * Start the processor of control-plane acceptor and the acceptor of this server.
- */
- private def startControlPlaneProcessorAndAcceptor(authorizerFutures: Map[Endpoint, CompletableFuture[Void]]): Unit = {
- controlPlaneAcceptorOpt.foreach { controlPlaneAcceptor =>
- startAcceptorAndProcessors(controlPlaneAcceptor, authorizerFutures)
- }
+ info("Enabling request processing.")
+ controlPlaneAcceptorOpt.foreach(chainAcceptorFuture)
+ dataPlaneAcceptors.values().forEach(chainAcceptorFuture)
+ chainFuture(CompletableFuture.allOf(authorizerFutures.values.toArray: _*),
+ allAuthorizerFuturesComplete)
}
- private def endpoints = config.listeners.map(l => l.listenerName -> l).toMap
-
- def createDataPlaneAcceptorsAndProcessors(endpoints: Seq[EndPoint]): Unit = {
- endpoints.foreach { endpoint =>
- val parsedConfigs = config.valuesFromThisConfigWithPrefixOverride(endpoint.listenerName.configPrefix)
- connectionQuotas.addListener(config, endpoint.listenerName)
-
- val isPrivilegedListener = controlPlaneRequestChannelOpt.isEmpty && config.interBrokerListenerName == endpoint.listenerName
-
- val dataPlaneAcceptor = createDataPlaneAcceptor(endpoint, isPrivilegedListener, dataPlaneRequestChannel)
- config.addReconfigurable(dataPlaneAcceptor)
- dataPlaneAcceptor.configure(parsedConfigs)
- dataPlaneAcceptors.put(endpoint, dataPlaneAcceptor)
- info(s"Created data-plane acceptor and processors for endpoint : ${endpoint.listenerName}")
+ def createDataPlaneAcceptorAndProcessors(endpoint: EndPoint): Unit = synchronized {
+ if (stopped) {
+ throw new RuntimeException("Can't create new data plane acceptor and processors: SocketServer is stopped.")
}
+ val parsedConfigs = config.valuesFromThisConfigWithPrefixOverride(endpoint.listenerName.configPrefix)
+ connectionQuotas.addListener(config, endpoint.listenerName)
+ val isPrivilegedListener = controlPlaneRequestChannelOpt.isEmpty &&
+ config.interBrokerListenerName == endpoint.listenerName
+ val dataPlaneAcceptor = createDataPlaneAcceptor(endpoint, isPrivilegedListener, dataPlaneRequestChannel)
+ config.addReconfigurable(dataPlaneAcceptor)
+ dataPlaneAcceptor.configure(parsedConfigs)
+ dataPlaneAcceptors.put(endpoint, dataPlaneAcceptor)
+ info(s"Created data-plane acceptor and processors for endpoint : ${endpoint.listenerName}")
}
- private def createControlPlaneAcceptorAndProcessor(endpointOpt: Option[EndPoint]): Unit = {
- endpointOpt.foreach { endpoint =>
- connectionQuotas.addListener(config, endpoint.listenerName)
- val controlPlaneAcceptor = createControlPlaneAcceptor(endpoint, controlPlaneRequestChannelOpt.get)
- controlPlaneAcceptor.addProcessors(1)
- controlPlaneAcceptorOpt = Some(controlPlaneAcceptor)
- info(s"Created control-plane acceptor and processor for endpoint : ${endpoint.listenerName}")
+ private def createControlPlaneAcceptorAndProcessor(endpoint: EndPoint): Unit = synchronized {
+ if (stopped) {
+ throw new RuntimeException("Can't create new control plane acceptor and processor: SocketServer is stopped.")
}
+ connectionQuotas.addListener(config, endpoint.listenerName)
+ val controlPlaneAcceptor = createControlPlaneAcceptor(endpoint, controlPlaneRequestChannelOpt.get)
+ controlPlaneAcceptor.addProcessors(1)
+ controlPlaneAcceptorOpt = Some(controlPlaneAcceptor)
+ info(s"Created control-plane acceptor and processor for endpoint : ${endpoint.listenerName}")
}
+ private def endpoints = config.listeners.map(l => l.listenerName -> l).toMap
protected def createDataPlaneAcceptor(endPoint: EndPoint, isPrivilegedListener: Boolean, requestChannel: RequestChannel): DataPlaneAcceptor = {
new DataPlaneAcceptor(this, endPoint, config, nodeId, connectionQuotas, time, isPrivilegedListener, requestChannel, metrics, credentialProvider, logContext, memoryPool, apiVersionManager)
@@ -294,18 +254,18 @@ class SocketServer(val config: KafkaConfig,
/**
* Stop processing requests and new connections.
*/
- def stopProcessingRequests(): Unit = {
- info("Stopping socket server request processors")
- this.synchronized {
- dataPlaneAcceptors.asScala.values.foreach(_.initiateShutdown())
- dataPlaneAcceptors.asScala.values.foreach(_.awaitShutdown())
- controlPlaneAcceptorOpt.foreach(_.initiateShutdown())
- controlPlaneAcceptorOpt.foreach(_.awaitShutdown())
+ def stopProcessingRequests(): Unit = synchronized {
+ if (!stopped) {
+ stopped = true
+ info("Stopping socket server request processors")
+ dataPlaneAcceptors.asScala.values.foreach(_.beginShutdown())
+ controlPlaneAcceptorOpt.foreach(_.beginShutdown())
+ dataPlaneAcceptors.asScala.values.foreach(_.close())
+ controlPlaneAcceptorOpt.foreach(_.close())
dataPlaneRequestChannel.clear()
controlPlaneRequestChannelOpt.foreach(_.clear())
- stoppedProcessingRequests = true
+ info("Stopped socket server request processors")
}
- info("Stopped socket server request processors")
}
/**
@@ -314,9 +274,10 @@ class SocketServer(val config: KafkaConfig,
*/
def shutdown(): Unit = {
info("Shutting down socket server")
+ allAuthorizerFuturesComplete.completeExceptionally(new TimeoutException("The socket " +
+ "server was shut down before the Authorizer could be completely initialized."))
this.synchronized {
- if (!stoppedProcessingRequests)
- stopProcessingRequests()
+ stopProcessingRequests()
dataPlaneRequestChannel.shutdown()
controlPlaneRequestChannelOpt.foreach(_.shutdown())
connectionQuotas.close()
@@ -338,12 +299,20 @@ class SocketServer(val config: KafkaConfig,
}
}
+ /**
+ * This method is called to dynamically add listeners.
+ */
def addListeners(listenersAdded: Seq[EndPoint]): Unit = synchronized {
+ if (stopped) {
+ throw new RuntimeException("can't add new listeners: SocketServer is stopped.")
+ }
info(s"Adding data-plane listeners for endpoints $listenersAdded")
- createDataPlaneAcceptorsAndProcessors(listenersAdded)
listenersAdded.foreach { endpoint =>
+ createDataPlaneAcceptorAndProcessors(endpoint)
val acceptor = dataPlaneAcceptors.get(endpoint)
- startAcceptorAndProcessors(acceptor)
+ // There is no authorizer future for this new listener endpoint. So start the
+ // listener once all authorizer futures are complete.
+ chainFuture(allAuthorizerFuturesComplete, acceptor.startFuture)
}
}
@@ -352,8 +321,8 @@ class SocketServer(val config: KafkaConfig,
listenersRemoved.foreach { endpoint =>
connectionQuotas.removeListener(config, endpoint.listenerName)
dataPlaneAcceptors.asScala.remove(endpoint).foreach { acceptor =>
- acceptor.initiateShutdown()
- acceptor.awaitShutdown()
+ acceptor.beginShutdown()
+ acceptor.close()
}
}
}
@@ -387,15 +356,6 @@ class SocketServer(val config: KafkaConfig,
}
}
- private def waitForAuthorizerFuture(acceptor: Acceptor,
- authorizerFutures: Map[Endpoint, CompletableFuture[Void]]): Unit = {
- //we can't rely on authorizerFutures.get() due to ephemeral ports. Get the future using listener name
- authorizerFutures.forKeyValue { (endpoint, future) =>
- if (endpoint.listenerName == Optional.of(acceptor.endPoint.listenerName.value))
- future.join()
- }
- }
-
// For test usage
private[network] def connectionCount(address: InetAddress): Int =
Option(connectionQuotas).fold(0)(_.get(address))
@@ -420,80 +380,22 @@ object SocketServer {
KafkaConfig.MaxConnectionCreationRateProp)
val ListenerReconfigurableConfigs = Set(KafkaConfig.MaxConnectionsProp, KafkaConfig.MaxConnectionCreationRateProp)
-}
-/**
- * A base class with some helper variables and methods
- */
-private[kafka] abstract class AbstractServerThread(connectionQuotas: ConnectionQuotas) extends Runnable with Logging {
-
- private val startupLatch = new CountDownLatch(1)
-
- // `shutdown()` is invoked before `startupComplete` and `shutdownComplete` if an exception is thrown in the constructor
- // (e.g. if the address is already in use). We want `shutdown` to proceed in such cases, so we first assign an open
- // latch and then replace it in `startupComplete()`.
- @volatile private var shutdownLatch = new CountDownLatch(0)
-
- private val alive = new AtomicBoolean(true)
-
- def wakeup(): Unit
-
- /**
- * Initiates a graceful shutdown by signaling to stop
- */
- def initiateShutdown(): Unit = {
- if (alive.getAndSet(false))
- wakeup()
+ def closeSocket(
+ channel: SocketChannel,
+ logging: Logging
+ ): Unit = {
+ CoreUtils.swallow(channel.socket().close(), logging, Level.ERROR)
+ CoreUtils.swallow(channel.close(), logging, Level.ERROR)
}
- /**
- * Wait for the thread to completely shutdown
- */
- def awaitShutdown(): Unit = shutdownLatch.await
-
- /**
- * Returns true if the thread is completely started
- */
- def isStarted(): Boolean = startupLatch.getCount == 0
-
- /**
- * Wait for the thread to completely start up
- */
- def awaitStartup(): Unit = startupLatch.await
-
- /**
- * Record that the thread startup is complete
- */
- protected def startupComplete(): Unit = {
- // Replace the open latch with a closed one
- shutdownLatch = new CountDownLatch(1)
- startupLatch.countDown()
- }
-
- /**
- * Record that the thread shutdown is complete
- */
- protected def shutdownComplete(): Unit = shutdownLatch.countDown()
-
- /**
- * Is the server still running?
- */
- protected def isRunning: Boolean = alive.get
-
- /**
- * Close `channel` and decrement the connection count.
- */
- def close(listenerName: ListenerName, channel: SocketChannel): Unit = {
- if (channel != null) {
- debug(s"Closing connection from ${channel.socket.getRemoteSocketAddress}")
- connectionQuotas.dec(listenerName, channel.socket.getInetAddress)
- closeSocket(channel)
- }
- }
-
- protected def closeSocket(channel: SocketChannel): Unit = {
- CoreUtils.swallow(channel.socket().close(), this, Level.ERROR)
- CoreUtils.swallow(channel.close(), this, Level.ERROR)
+ def chainFuture(sourceFuture: CompletableFuture[Void],
+ destinationFuture: CompletableFuture[Void]): Unit = {
+ sourceFuture.whenComplete((_, t) => if (t != null) {
+ destinationFuture.completeExceptionally(t)
+ } else {
+ destinationFuture.complete(null)
+ })
}
}
@@ -650,7 +552,7 @@ private[kafka] abstract class Acceptor(val socketServer: SocketServer,
val endPoint: EndPoint,
var config: KafkaConfig,
nodeId: Int,
- connectionQuotas: ConnectionQuotas,
+ val connectionQuotas: ConnectionQuotas,
time: Time,
isPrivilegedListener: Boolean,
requestChannel: RequestChannel,
@@ -659,7 +561,9 @@ private[kafka] abstract class Acceptor(val socketServer: SocketServer,
logContext: LogContext,
memoryPool: MemoryPool,
apiVersionManager: ApiVersionManager)
- extends AbstractServerThread(connectionQuotas) with KafkaMetricsGroup {
+ extends Runnable with Logging with KafkaMetricsGroup {
+
+ val shouldRun = new AtomicBoolean(true)
def metricPrefix(): String
def threadPrefix(): String
@@ -671,7 +575,6 @@ private[kafka] abstract class Acceptor(val socketServer: SocketServer,
private val nioSelector = NSelector.open()
private[network] val serverChannel = openServerSocket(endPoint.host, endPoint.port, listenBacklogSize)
private[network] val processors = new ArrayBuffer[Processor]()
- private val processorsStarted = new AtomicBoolean
// Build the metric name explicitly in order to keep the existing name for compatibility
private val blockedPercentMeterMetricName = explicitMetricName(
"kafka.network",
@@ -681,24 +584,27 @@ private[kafka] abstract class Acceptor(val socketServer: SocketServer,
private val blockedPercentMeter = newMeter(blockedPercentMeterMetricName,"blocked time", TimeUnit.NANOSECONDS)
private var currentProcessorIndex = 0
private[network] val throttledSockets = new mutable.PriorityQueue[DelayedCloseSocket]()
+ private var started = false
+ private[network] val startFuture = new CompletableFuture[Void]()
- private[network] case class DelayedCloseSocket(socket: SocketChannel, endThrottleTimeMs: Long) extends Ordered[DelayedCloseSocket] {
- override def compare(that: DelayedCloseSocket): Int = endThrottleTimeMs compare that.endThrottleTimeMs
- }
+ val thread = KafkaThread.nonDaemon(
+ s"${threadPrefix()}-kafka-socket-acceptor-${endPoint.listenerName}-${endPoint.securityProtocol}-${endPoint.port}",
+ this)
- private[network] def startProcessors(): Unit = synchronized {
- if (!processorsStarted.getAndSet(true)) {
- startProcessors(processors)
+ startFuture.thenRun(() => synchronized {
+ if (!shouldRun.get()) {
+ debug(s"Ignoring start future for ${endPoint.listenerName} since the acceptor has already been shut down.")
+ } else {
+ debug(s"Starting processors for listener ${endPoint.listenerName}")
+ started = true
+ processors.foreach(_.start())
+ debug(s"Starting acceptor thread for listener ${endPoint.listenerName}")
+ thread.start()
}
- }
+ })
- private def startProcessors(processors: Seq[Processor]): Unit = synchronized {
- processors.foreach { processor =>
- KafkaThread.nonDaemon(
- s"${threadPrefix()}-kafka-network-thread-$nodeId-${endPoint.listenerName}-${endPoint.securityProtocol}-${processor.id}",
- processor
- ).start()
- }
+ private[network] case class DelayedCloseSocket(socket: SocketChannel, endThrottleTimeMs: Long) extends Ordered[DelayedCloseSocket] {
+ override def compare(that: DelayedCloseSocket): Int = endThrottleTimeMs compare that.endThrottleTimeMs
}
private[network] def removeProcessors(removeCount: Int): Unit = synchronized {
@@ -707,33 +613,34 @@ private[kafka] abstract class Acceptor(val socketServer: SocketServer,
// The processors are then removed from `requestChannel` and any pending responses to these processors are dropped.
val toRemove = processors.takeRight(removeCount)
processors.remove(processors.size - removeCount, removeCount)
- toRemove.foreach(_.initiateShutdown())
- toRemove.foreach(_.awaitShutdown())
+ toRemove.foreach(_.close())
toRemove.foreach(processor => requestChannel.removeProcessor(processor.id))
}
- override def initiateShutdown(): Unit = {
- super.initiateShutdown()
- synchronized {
- processors.foreach(_.initiateShutdown())
+ def beginShutdown(): Unit = {
+ if (shouldRun.getAndSet(false)) {
+ wakeup()
+ synchronized {
+ processors.foreach(_.beginShutdown())
+ }
}
}
- override def awaitShutdown(): Unit = {
- super.awaitShutdown()
+ def close(): Unit = {
+ beginShutdown()
+ thread.join()
synchronized {
- processors.foreach(_.awaitShutdown())
+ processors.foreach(_.close())
}
}
/**
* Accept loop that checks for new connection attempts
*/
- def run(): Unit = {
+ override def run(): Unit = {
serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT)
- startupComplete()
try {
- while (isRunning) {
+ while (shouldRun.get()) {
try {
acceptNewConnections()
closeThrottledConnections()
@@ -750,9 +657,8 @@ private[kafka] abstract class Acceptor(val socketServer: SocketServer,
debug("Closing server socket, selector, and any throttled sockets.")
CoreUtils.swallow(serverChannel.close(), this, Level.ERROR)
CoreUtils.swallow(nioSelector.close(), this, Level.ERROR)
- throttledSockets.foreach(throttledSocket => closeSocket(throttledSocket.socket))
+ throttledSockets.foreach(throttledSocket => closeSocket(throttledSocket.socket, this))
throttledSockets.clear()
- shutdownComplete()
}
}
@@ -788,7 +694,7 @@ private[kafka] abstract class Acceptor(val socketServer: SocketServer,
if (ready > 0) {
val keys = nioSelector.selectedKeys()
val iter = keys.iterator()
- while (iter.hasNext && isRunning) {
+ while (iter.hasNext && shouldRun.get()) {
try {
val key = iter.next
iter.remove()
@@ -833,7 +739,7 @@ private[kafka] abstract class Acceptor(val socketServer: SocketServer,
} catch {
case e: TooManyConnectionsException =>
info(s"Rejected connection from ${e.ip}, address already has the configured maximum of ${e.count} connections.")
- close(endPoint.listenerName, socketChannel)
+ connectionQuotas.closeChannel(this, endPoint.listenerName, socketChannel)
None
case e: ConnectionThrottledException =>
val ip = socketChannel.socket.getInetAddress
@@ -843,7 +749,7 @@ private[kafka] abstract class Acceptor(val socketServer: SocketServer,
None
case e: IOException =>
error(s"Encountered an error while configuring the connection, closing it.", e)
- close(endPoint.listenerName, socketChannel)
+ connectionQuotas.closeChannel(this, endPoint.listenerName, socketChannel)
None
}
}
@@ -864,7 +770,7 @@ private[kafka] abstract class Acceptor(val socketServer: SocketServer,
while (throttledSockets.headOption.exists(_.endThrottleTimeMs < timeMs)) {
val closingSocket = throttledSockets.dequeue()
debug(s"Closing socket from ip ${closingSocket.socket.getRemoteAddress}")
- closeSocket(closingSocket.socket)
+ closeSocket(closingSocket.socket, this)
}
}
@@ -882,10 +788,9 @@ private[kafka] abstract class Acceptor(val socketServer: SocketServer,
/**
* Wakeup the thread for selection.
*/
- @Override
def wakeup(): Unit = nioSelector.wakeup()
- def addProcessors(toCreate: Int): Unit = {
+ def addProcessors(toCreate: Int): Unit = synchronized {
val listenerName = endPoint.listenerName
val securityProtocol = endPoint.securityProtocol
val listenerProcessors = new ArrayBuffer[Processor]()
@@ -894,14 +799,16 @@ private[kafka] abstract class Acceptor(val socketServer: SocketServer,
val processor = newProcessor(socketServer.nextProcessorId(), listenerName, securityProtocol)
listenerProcessors += processor
requestChannel.addProcessor(processor)
- }
+ if (started) {
+ processor.start()
+ }
+ }
processors ++= listenerProcessors
- if (processorsStarted.get)
- startProcessors(listenerProcessors)
}
def newProcessor(id: Int, listenerName: ListenerName, securityProtocol: SecurityProtocol): Processor = {
+ val name = s"${threadPrefix()}-kafka-network-thread-$nodeId-${endPoint.listenerName}-${endPoint.securityProtocol}-${id}"
new Processor(id,
time,
config.socketRequestMaxBytes,
@@ -918,9 +825,9 @@ private[kafka] abstract class Acceptor(val socketServer: SocketServer,
logContext,
Processor.ConnectionQueueSize,
isPrivilegedListener,
- apiVersionManager)
+ apiVersionManager,
+ name)
}
-
}
private[kafka] object Processor {
@@ -940,23 +847,29 @@ private[kafka] object Processor {
* forwarding requests; if the control plane is not defined, the processor
* relying on the inter broker listener would be acting as the privileged listener.
*/
-private[kafka] class Processor(val id: Int,
- time: Time,
- maxRequestSize: Int,
- requestChannel: RequestChannel,
- connectionQuotas: ConnectionQuotas,
- connectionsMaxIdleMs: Long,
- failedAuthenticationDelayMs: Int,
- listenerName: ListenerName,
- securityProtocol: SecurityProtocol,
- config: KafkaConfig,
- metrics: Metrics,
- credentialProvider: CredentialProvider,
- memoryPool: MemoryPool,
- logContext: LogContext,
- connectionQueueSize: Int,
- isPrivilegedListener: Boolean,
- apiVersionManager: ApiVersionManager) extends AbstractServerThread(connectionQuotas) with KafkaMetricsGroup {
+private[kafka] class Processor(
+ val id: Int,
+ time: Time,
+ maxRequestSize: Int,
+ requestChannel: RequestChannel,
+ connectionQuotas: ConnectionQuotas,
+ connectionsMaxIdleMs: Long,
+ failedAuthenticationDelayMs: Int,
+ listenerName: ListenerName,
+ securityProtocol: SecurityProtocol,
+ config: KafkaConfig,
+ metrics: Metrics,
+ credentialProvider: CredentialProvider,
+ memoryPool: MemoryPool,
+ logContext: LogContext,
+ connectionQueueSize: Int,
+ isPrivilegedListener: Boolean,
+ apiVersionManager: ApiVersionManager,
+ threadName: String
+) extends Runnable with KafkaMetricsGroup {
+ val shouldRun = new AtomicBoolean(true)
+
+ val thread = KafkaThread.nonDaemon(threadName, this)
private object ConnectionId {
def fromString(s: String): Option[ConnectionId] = s.split("-") match {
@@ -1036,9 +949,8 @@ private[kafka] class Processor(val id: Int,
private var nextConnectionIndex = 0
override def run(): Unit = {
- startupComplete()
try {
- while (isRunning) {
+ while (shouldRun.get()) {
try {
// setup any new connections that have been queued up
configureNewConnections()
@@ -1062,7 +974,6 @@ private[kafka] class Processor(val id: Int,
} finally {
debug(s"Closing selector - processor $id")
CoreUtils.swallow(closeAll(), this, Level.ERROR)
- shutdownComplete()
}
}
@@ -1325,7 +1236,7 @@ private[kafka] class Processor(val id: Int,
case e: Throwable =>
val remoteAddress = channel.socket.getRemoteSocketAddress
// need to close the channel here to avoid a socket leak.
- close(listenerName, channel)
+ connectionQuotas.closeChannel(this, listenerName, channel)
processException(s"Processor $id closed connection from $remoteAddress", e)
}
}
@@ -1392,15 +1303,27 @@ private[kafka] class Processor(val id: Int,
private[network] def channel(connectionId: String): Option[KafkaChannel] =
Option(selector.channel(connectionId))
+ def start(): Unit = thread.start()
+
/**
* Wakeup the thread for selection.
*/
- override def wakeup(): Unit = selector.wakeup()
+ def wakeup(): Unit = selector.wakeup()
- override def initiateShutdown(): Unit = {
- super.initiateShutdown()
- removeMetric("IdlePercent", Map("networkProcessor" -> id.toString))
- metrics.removeMetric(expiredConnectionsKilledCountMetricName)
+ def beginShutdown(): Unit = {
+ if (shouldRun.getAndSet(false)) {
+ wakeup()
+ }
+ }
+
+ def close(): Unit = {
+ try {
+ beginShutdown()
+ thread.join()
+ } finally {
+ removeMetric("IdlePercent", Map("networkProcessor" -> id.toString))
+ metrics.removeMetric(expiredConnectionsKilledCountMetricName)
+ }
}
}
@@ -1864,6 +1787,18 @@ class ConnectionQuotas(config: KafkaConfig, time: Time, metrics: Metrics) extend
sensor
}
}
+
+ /**
+ * Close `channel` and decrement the connection count.
+ */
+ def closeChannel(log: Logging, listenerName: ListenerName, channel: SocketChannel): Unit = {
+ if (channel != null) {
+ log.debug(s"Closing connection from ${channel.socket.getRemoteSocketAddress}")
+ dec(listenerName, channel.socket.getInetAddress)
+ closeSocket(channel, log)
+ }
+ }
+
}
class TooManyConnectionsException(val ip: InetAddress, val count: Int) extends KafkaException(s"Too many connections from $ip (maximum = $count)")
diff --git a/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala b/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
index 3ca2704eb3c..39dff71ad11 100644
--- a/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
+++ b/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
@@ -97,6 +97,11 @@ class BrokerLifecycleManager(val config: KafkaConfig,
*/
val initialCatchUpFuture = new CompletableFuture[Void]()
+ /**
+ * A future which is completed when the broker is unfenced for the first time.
+ */
+ val initialUnfenceFuture = new CompletableFuture[Void]()
+
/**
* A future which is completed when controlled shutdown is done.
*/
@@ -189,8 +194,9 @@ class BrokerLifecycleManager(val config: KafkaConfig,
channelManager, clusterId, advertisedListeners, supportedFeatures))
}
- def setReadyToUnfence(): Unit = {
+ def setReadyToUnfence(): CompletableFuture[Void] = {
eventQueue.append(new SetReadyToUnfenceEvent())
+ initialUnfenceFuture
}
def brokerEpoch: Long = _brokerEpoch
@@ -384,6 +390,7 @@ class BrokerLifecycleManager(val config: KafkaConfig,
case BrokerState.RECOVERY =>
if (!message.data().isFenced) {
info(s"The broker has been unfenced. Transitioning from RECOVERY to RUNNING.")
+ initialUnfenceFuture.complete(null)
_state = BrokerState.RUNNING
} else {
info(s"The broker is in RECOVERY.")
@@ -476,6 +483,7 @@ class BrokerLifecycleManager(val config: KafkaConfig,
_state = BrokerState.SHUTTING_DOWN
controlledShutdownFuture.complete(null)
initialCatchUpFuture.cancel(false)
+ initialUnfenceFuture.cancel(false)
if (_channelManager != null) {
_channelManager.shutdown()
_channelManager = null
diff --git a/core/src/main/scala/kafka/server/BrokerServer.scala b/core/src/main/scala/kafka/server/BrokerServer.scala
index 5447e298636..514b94bab15 100644
--- a/core/src/main/scala/kafka/server/BrokerServer.scala
+++ b/core/src/main/scala/kafka/server/BrokerServer.scala
@@ -43,6 +43,7 @@ import org.apache.kafka.common.security.scram.internals.ScramMechanism
import org.apache.kafka.common.security.token.delegation.internals.DelegationTokenCache
import org.apache.kafka.common.utils.{AppInfoParser, LogContext, Time, Utils}
import org.apache.kafka.common.{ClusterResource, Endpoint}
+import org.apache.kafka.metadata.authorizer.ClusterMetadataAuthorizer
import org.apache.kafka.metadata.{BrokerState, VersionRange}
import org.apache.kafka.raft.RaftConfig.AddressSpec
import org.apache.kafka.raft.{RaftClient, RaftConfig}
@@ -183,7 +184,9 @@ class BrokerServer(
config.dynamicConfig.initialize(zkClientOpt = None)
- lifecycleManager = new BrokerLifecycleManager(config, time, threadNamePrefix)
+ lifecycleManager = new BrokerLifecycleManager(config,
+ time,
+ threadNamePrefix)
/* start scheduler */
kafkaScheduler = new KafkaScheduler(config.backgroundThreads)
@@ -237,7 +240,6 @@ class BrokerServer(
// Delay starting processors until the end of the initialization sequence to ensure
// that credentials have been loaded before processing authentications.
socketServer = new SocketServer(config, metrics, time, credentialProvider, apiVersionManager)
- socketServer.startup(startProcessingRequests = false)
clientQuotaMetadataManager = new ClientQuotaMetadataManager(quotaManagers, socketServer.connectionQuotas)
@@ -365,10 +367,13 @@ class BrokerServer(
endpoints.asScala.map(ep => ep.listenerName().orElse("(none)")).mkString(", "))
}
val authorizerInfo = ServerInfo(new ClusterResource(clusterId),
- config.nodeId, endpoints, interBrokerListener)
+ config.nodeId,
+ endpoints,
+ interBrokerListener,
+ config.earlyStartListeners.map(_.value()).asJava)
- /* Get the authorizer and initialize it if one is specified.*/
- authorizer = config.authorizer
+ // Create and initialize an authorizer if one is configured.
+ authorizer = config.createNewAuthorizer()
authorizer.foreach(_.configure(config.originals))
val authorizerFutures: Map[Endpoint, CompletableFuture[Void]] = authorizer match {
case Some(authZ) =>
@@ -432,17 +437,36 @@ class BrokerServer(
// publish operation to complete. This first operation will initialize logManager,
// replicaManager, groupCoordinator, and txnCoordinator. The log manager may perform
// a potentially lengthy recovery-from-unclean-shutdown operation here, if required.
- metadataListener.startPublishing(metadataPublisher).get()
+ try {
+ metadataListener.startPublishing(metadataPublisher).get()
+ } catch {
+ case t: Throwable => throw new RuntimeException("Received a fatal error while " +
+ "waiting for the broker to catch up with the current cluster metadata.", t)
+ }
// Log static broker configurations.
new KafkaConfig(config.originals(), true)
- // Enable inbound TCP connections.
- socketServer.startProcessingRequests(authorizerFutures)
+ // Enable inbound TCP connections. Each endpoint will be started only once its matching
+ // authorizer future is completed.
+ socketServer.enableRequestProcessing(authorizerFutures)
+
+ // If we are using a ClusterMetadataAuthorizer which stores its ACLs in the metadata log,
+ // notify it that the loading process is complete.
+ authorizer match {
+ case Some(clusterMetadataAuthorizer: ClusterMetadataAuthorizer) =>
+ clusterMetadataAuthorizer.completeInitialLoad()
+ case _ => // nothing to do
+ }
// We're now ready to unfence the broker. This also allows this broker to transition
// from RECOVERY state to RUNNING state, once the controller unfences the broker.
- lifecycleManager.setReadyToUnfence()
+ try {
+ lifecycleManager.setReadyToUnfence().get()
+ } catch {
+ case t: Throwable => throw new RuntimeException("Received a fatal error while " +
+ "waiting for the broker to be unfenced.", t)
+ }
maybeChangeStatus(STARTING, STARTED)
} catch {
diff --git a/core/src/main/scala/kafka/server/ControllerServer.scala b/core/src/main/scala/kafka/server/ControllerServer.scala
index ba38aa17853..e004996bf74 100644
--- a/core/src/main/scala/kafka/server/ControllerServer.scala
+++ b/core/src/main/scala/kafka/server/ControllerServer.scala
@@ -103,7 +103,6 @@ class ControllerServer(
info("Starting controller")
maybeChangeStatus(STARTING, STARTED)
- // TODO: initialize the log dir(s)
this.logIdent = new LogContext(s"[ControllerServer id=${config.nodeId}] ").logPrefix()
newGauge("ClusterId", () => clusterId)
@@ -116,7 +115,7 @@ class ControllerServer(
}
val javaListeners = config.controllerListeners.map(_.toJava).asJava
- authorizer = config.authorizer
+ authorizer = config.createNewAuthorizer()
authorizer.foreach(_.configure(config.originals))
val authorizerFutures: Map[Endpoint, CompletableFuture[Void]] = authorizer match {
@@ -125,7 +124,11 @@ class ControllerServer(
// AuthorizerServerInfo, such as the assumption that there is an inter-broker
// listener, or that ID is named brokerId.
val controllerAuthorizerInfo = ServerInfo(
- new ClusterResource(clusterId), config.nodeId, javaListeners, javaListeners.get(0))
+ new ClusterResource(clusterId),
+ config.nodeId,
+ javaListeners,
+ javaListeners.get(0),
+ config.earlyStartListeners.map(_.value()).asJava)
authZ.start(controllerAuthorizerInfo).asScala.map { case (ep, cs) =>
ep -> cs.toCompletableFuture
}.toMap
@@ -144,7 +147,6 @@ class ControllerServer(
time,
credentialProvider,
apiVersionManager)
- socketServer.startup(startProcessingRequests = false, controlPlaneListener = None, config.controllerListeners)
if (config.controllerListeners.nonEmpty) {
socketServerFirstBoundPortFuture.complete(socketServer.boundPort(
@@ -213,7 +215,17 @@ class ControllerServer(
config.numIoThreads,
s"${DataPlaneAcceptor.MetricPrefix}RequestHandlerAvgIdlePercent",
DataPlaneAcceptor.ThreadPrefix)
- socketServer.startProcessingRequests(authorizerFutures)
+
+ /**
+ * Enable the controller endpoint(s). If we are using an authorizer which stores
+ * ACLs in the metadata log, such as StandardAuthorizer, we will be able to start
+ * accepting requests from principals included super.users right after this point,
+ * but we will not be able to process requests from non-superusers until the
+ * QuorumController declares that we have caught up to the high water mark of the
+ * metadata log. See @link{QuorumController#maybeCompleteAuthorizerInitialLoad}
+ * and KIP-801 for details.
+ */
+ socketServer.enableRequestProcessing(authorizerFutures)
} catch {
case e: Throwable =>
maybeChangeStatus(STARTING, STARTED)
@@ -241,6 +253,7 @@ class ControllerServer(
CoreUtils.swallow(quotaManagers.shutdown(), this)
if (controller != null)
controller.close()
+ CoreUtils.swallow(authorizer.foreach(_.close()), this)
createTopicPolicy.foreach(policy => CoreUtils.swallow(policy.close(), this))
alterConfigPolicy.foreach(policy => CoreUtils.swallow(policy.close(), this))
socketServerFirstBoundPortFuture.completeExceptionally(new RuntimeException("shutting down"))
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index 7d0d3a3d852..5942b164ea4 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -405,6 +405,8 @@ object KafkaConfig {
/************* Authorizer Configuration ***********/
val AuthorizerClassNameProp = "authorizer.class.name"
+ val EarlyStartListenersProp = "early.start.listeners"
+
/** ********* Socket Server Configuration ***********/
val ListenersProp = "listeners"
val AdvertisedListenersProp = "advertised.listeners"
@@ -725,7 +727,12 @@ object KafkaConfig {
/************* Authorizer Configuration ***********/
val AuthorizerClassNameDoc = s"The fully qualified name of a class that implements <code>${classOf[Authorizer].getName}</code>" +
- " interface, which is used by the broker for authorization."
+ " interface, which is used by the broker for authorization."
+ val EarlyStartListenersDoc = "A comma-separated list of listener names which may be started before the authorizer has finished " +
+ "initialization. This is useful when the authorizer is dependent on the cluster itself for bootstrapping, as is the case for " +
+ "the StandardAuthorizer (which stores ACLs in the metadata log.) By default, all listeners included in controller.listener.names " +
+ "will also be early start listeners. A listener should not appear in this list if it accepts external traffic."
+
/** ********* Socket Server Configuration ***********/
val ListenersDoc = "Listener List - Comma-separated list of URIs we will listen on and the listener names." +
s" If the listener name is not a security protocol, <code>$ListenerSecurityProtocolMapProp</code> must also be set.\n" +
@@ -1142,6 +1149,7 @@ object KafkaConfig {
/************* Authorizer Configuration ***********/
.define(AuthorizerClassNameProp, STRING, Defaults.AuthorizerClassName, new ConfigDef.NonNullValidator(), LOW, AuthorizerClassNameDoc)
+ .define(EarlyStartListenersProp, STRING, null, HIGH, EarlyStartListenersDoc)
/** ********* Socket Server Configuration ***********/
.define(ListenersProp, STRING, Defaults.Listeners, HIGH, ListenersDoc)
@@ -1653,7 +1661,7 @@ class KafkaConfig private(doLog: Boolean, val props: java.util.Map[_, _], dynami
val metadataSnapshotMaxNewRecordBytes = getLong(KafkaConfig.MetadataSnapshotMaxNewRecordBytesProp)
/************* Authorizer Configuration ***********/
- val authorizer: Option[Authorizer] = {
+ def createNewAuthorizer(): Option[Authorizer] = {
val className = getString(KafkaConfig.AuthorizerClassNameProp)
if (className == null || className.isEmpty)
None
@@ -1662,6 +1670,23 @@ class KafkaConfig private(doLog: Boolean, val props: java.util.Map[_, _], dynami
}
}
+ val earlyStartListeners: Set[ListenerName] = {
+ val listenersSet = listeners.map(_.listenerName).toSet
+ val controllerListenersSet = controllerListeners.map(_.listenerName).toSet
+ Option(getString(KafkaConfig.EarlyStartListenersProp)) match {
+ case None => controllerListenersSet
+ case Some(str) =>
+ str.split(",").map(_.trim()).filter(!_.isEmpty).map { str =>
+ val listenerName = new ListenerName(str)
+ if (!listenersSet.contains(listenerName) && !controllerListenersSet.contains(listenerName))
+ throw new ConfigException(s"${KafkaConfig.EarlyStartListenersProp} contains " +
+ s"listener ${listenerName.value()}, but this is not contained in " +
+ s"${KafkaConfig.ListenersProp} or ${KafkaConfig.ControllerListenerNamesProp}")
+ listenerName
+ }.toSet
+ }
+ }
+
/** ********* Socket Server Configuration ***********/
val socketSendBufferBytes = getInt(KafkaConfig.SocketSendBufferBytesProp)
val socketReceiveBufferBytes = getInt(KafkaConfig.SocketReceiveBufferBytesProp)
diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala
index c69f43c9eff..b1273ed628f 100755
--- a/core/src/main/scala/kafka/server/KafkaServer.scala
+++ b/core/src/main/scala/kafka/server/KafkaServer.scala
@@ -307,7 +307,6 @@ class KafkaServer(
// Note that we allow the use of KRaft mode controller APIs when forwarding is enabled
// so that the Envelope request is exposed. This is only used in testing currently.
socketServer = new SocketServer(config, metrics, time, credentialProvider, apiVersionManager)
- socketServer.startup(startProcessingRequests = false)
// Start alter partition manager based on the IBP version
alterIsrManager = if (config.interBrokerProtocolVersion.isAlterPartitionSupported) {
@@ -381,7 +380,7 @@ class KafkaServer(
)
/* Get the authorizer and initialize it if one is specified.*/
- authorizer = config.authorizer
+ authorizer = config.createNewAuthorizer()
authorizer.foreach(_.configure(config.originals))
val authorizerFutures: Map[Endpoint, CompletableFuture[Void]] = authorizer match {
case Some(authZ) =>
@@ -450,7 +449,7 @@ class KafkaServer(
dynamicConfigManager = new ZkConfigManager(zkClient, dynamicConfigHandlers)
dynamicConfigManager.startup()
- socketServer.startProcessingRequests(authorizerFutures)
+ socketServer.enableRequestProcessing(authorizerFutures)
_brokerState = BrokerState.RUNNING
shutdownLatch = new CountDownLatch(1)
diff --git a/core/src/main/scala/kafka/tools/TestRaftServer.scala b/core/src/main/scala/kafka/tools/TestRaftServer.scala
index 0b27f7fcb52..56104df8212 100644
--- a/core/src/main/scala/kafka/tools/TestRaftServer.scala
+++ b/core/src/main/scala/kafka/tools/TestRaftServer.scala
@@ -19,6 +19,7 @@ package kafka.tools
import java.util.concurrent.atomic.{AtomicInteger, AtomicLong}
import java.util.concurrent.{CompletableFuture, CountDownLatch, LinkedBlockingDeque, TimeUnit}
+
import joptsimple.OptionException
import kafka.network.{DataPlaneAcceptor, SocketServer}
import kafka.raft.{KafkaRaftManager, RaftManager}
@@ -74,7 +75,6 @@ class TestRaftServer(
val apiVersionManager = new SimpleApiVersionManager(ListenerType.CONTROLLER)
socketServer = new SocketServer(config, metrics, time, credentialProvider, apiVersionManager)
- socketServer.startup(startProcessingRequests = false)
val metaProperties = MetaProperties(
clusterId = Uuid.ZERO_UUID.toString,
@@ -119,7 +119,7 @@ class TestRaftServer(
workloadGenerator.start()
raftManager.startup()
- socketServer.startProcessingRequests(Map.empty)
+ socketServer.enableRequestProcessing(Map.empty)
}
def shutdown(): Unit = {
diff --git a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
index 9b4e81ab39e..98f92d61ff2 100644
--- a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
+++ b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
@@ -79,7 +79,7 @@ class SocketServerTest {
private val apiVersionManager = new SimpleApiVersionManager(ListenerType.ZK_BROKER)
val server = new SocketServer(config, metrics, Time.SYSTEM, credentialProvider, apiVersionManager)
- server.startup()
+ server.enableRequestProcessing(Map.empty)
val sockets = new ArrayBuffer[Socket]
private val kafkaLogger = org.apache.log4j.LogManager.getLogger("kafka")
@@ -296,20 +296,18 @@ class SocketServerTest {
shutdownServerAndMetrics(server)
val testProps = new Properties
testProps ++= props
- testProps.put("listeners", "EXTERNAL://localhost:0,INTERNAL://localhost:0,CONTROLLER://localhost:0")
- testProps.put("listener.security.protocol.map", "EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT")
- testProps.put("control.plane.listener.name", "CONTROLLER")
+ testProps.put("listeners", "EXTERNAL://localhost:0,INTERNAL://localhost:0,CONTROL_PLANE://localhost:0")
+ testProps.put("listener.security.protocol.map", "EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT,CONTROL_PLANE:PLAINTEXT")
+ testProps.put("control.plane.listener.name", "CONTROL_PLANE")
testProps.put("inter.broker.listener.name", "INTERNAL")
val config = KafkaConfig.fromProps(testProps)
val testableServer = new TestableSocketServer(config)
- testableServer.startup(startProcessingRequests = false)
val updatedEndPoints = config.effectiveAdvertisedListeners.map { endpoint =>
endpoint.copy(port = testableServer.boundPort(endpoint.listenerName))
}.map(_.toJava)
val externalReadyFuture = new CompletableFuture[Void]()
- val executor = Executors.newSingleThreadExecutor()
def controlPlaneListenerStarted() = {
try {
@@ -334,18 +332,19 @@ class SocketServerTest {
try {
val externalListener = new ListenerName("EXTERNAL")
val externalEndpoint = updatedEndPoints.find(e => e.listenerName.get == externalListener.value).get
- val futures = Map(externalEndpoint -> externalReadyFuture)
- val startFuture = executor.submit((() => testableServer.startProcessingRequests(futures)): Runnable)
+ val controlPlaneListener = new ListenerName("CONTROL_PLANE")
+ val controlPlaneEndpoint = updatedEndPoints.find(e => e.listenerName.get == controlPlaneListener.value).get
+ val futures = Map(
+ externalEndpoint -> externalReadyFuture,
+ controlPlaneEndpoint -> CompletableFuture.completedFuture[Void](null))
+ testableServer.enableRequestProcessing(futures)
TestUtils.waitUntilTrue(() => controlPlaneListenerStarted(), "Control plane listener not started")
- TestUtils.waitUntilTrue(() => listenerStarted(config.interBrokerListenerName), "Inter-broker listener not started")
- assertFalse(startFuture.isDone, "Socket server startup did not wait for future to complete")
-
+ assertFalse(listenerStarted(config.interBrokerListenerName))
assertFalse(listenerStarted(externalListener))
-
externalReadyFuture.complete(null)
+ TestUtils.waitUntilTrue(() => listenerStarted(config.interBrokerListenerName), "Inter-broker listener not started")
TestUtils.waitUntilTrue(() => listenerStarted(externalListener), "External listener not started")
} finally {
- executor.shutdownNow()
shutdownServerAndMetrics(testableServer)
}
}
@@ -362,7 +361,6 @@ class SocketServerTest {
val config = KafkaConfig.fromProps(testProps)
val connectionQueueSize = 1
val testableServer = new TestableSocketServer(config, connectionQueueSize)
- testableServer.startup(startProcessingRequests = false)
val socket1 = connect(testableServer, new ListenerName("EXTERNAL"), localAddr = InetAddress.getLocalHost)
sendRequest(socket1, producerRequestBytes())
@@ -468,7 +466,7 @@ class SocketServerTest {
time, credentialProvider, apiVersionManager)
try {
- overrideServer.startup()
+ overrideServer.enableRequestProcessing(Map.empty)
val serializedBytes = producerRequestBytes()
// Connection with no outstanding requests
@@ -536,7 +534,7 @@ class SocketServerTest {
}
try {
- overrideServer.startup()
+ overrideServer.enableRequestProcessing(Map.empty)
overrideServer.testableProcessor.setConnectionId(overrideConnectionId)
val socket1 = connectAndWaitForConnectionRegister()
TestUtils.waitUntilTrue(() => connectionCount == 1 && openChannel.isDefined, "Failed to create channel")
@@ -805,7 +803,7 @@ class SocketServerTest {
val server = new SocketServer(KafkaConfig.fromProps(newProps), new Metrics(),
Time.SYSTEM, credentialProvider, apiVersionManager)
try {
- server.startup()
+ server.enableRequestProcessing(Map.empty)
// make the maximum allowable number of connections
val conns = (0 until 5).map(_ => connect(server))
// now try one more (should fail)
@@ -844,7 +842,7 @@ class SocketServerTest {
val overrideServer = new SocketServer(KafkaConfig.fromProps(overrideProps), serverMetrics,
Time.SYSTEM, credentialProvider, apiVersionManager)
try {
- overrideServer.startup()
+ overrideServer.enableRequestProcessing(Map.empty)
// make the maximum allowable number of connections
val conns = (0 until overrideNum).map(_ => connect(overrideServer))
@@ -884,7 +882,7 @@ class SocketServerTest {
}
try {
- overrideServer.startup()
+ overrideServer.enableRequestProcessing(Map.empty)
val conn = connect(overrideServer)
conn.setSoTimeout(3000)
assertEquals(-1, conn.getInputStream.read())
@@ -907,7 +905,7 @@ class SocketServerTest {
// update the connection rate to 5
overrideServer.connectionQuotas.updateIpConnectionRateQuota(None, Some(connectionRate))
try {
- overrideServer.startup()
+ overrideServer.enableRequestProcessing(Map.empty)
// make the (maximum allowable number + 1) of connections
(0 to connectionRate).map(_ => connect(overrideServer))
@@ -956,7 +954,7 @@ class SocketServerTest {
val overrideServer = new SocketServer(KafkaConfig.fromProps(overrideProps), new Metrics(),
time, credentialProvider, apiVersionManager)
overrideServer.connectionQuotas.updateIpConnectionRateQuota(None, Some(connectionRate))
- overrideServer.startup()
+ overrideServer.enableRequestProcessing(Map.empty)
// make the maximum allowable number of connections
(0 until connectionRate).map(_ => connect(overrideServer))
// now try one more (should get throttled)
@@ -979,7 +977,7 @@ class SocketServerTest {
val overrideServer = new SocketServer(KafkaConfig.fromProps(sslServerProps), serverMetrics,
Time.SYSTEM, credentialProvider, apiVersionManager)
try {
- overrideServer.startup()
+ overrideServer.enableRequestProcessing(Map.empty)
val sslContext = SSLContext.getInstance(TestSslUtils.DEFAULT_TLS_PROTOCOL_FOR_TESTS)
sslContext.init(null, Array(TestUtils.trustAllCerts), new java.security.SecureRandom())
val socketFactory = sslContext.getSocketFactory
@@ -1038,7 +1036,7 @@ class SocketServerTest {
val time = new MockTime()
val overrideServer = new TestableSocketServer(KafkaConfig.fromProps(overrideProps), time = time)
try {
- overrideServer.startup()
+ overrideServer.enableRequestProcessing(Map.empty)
val socket = connect(overrideServer, ListenerName.forSecurityProtocol(SecurityProtocol.SASL_PLAINTEXT))
val correlationId = -1
@@ -1118,7 +1116,7 @@ class SocketServerTest {
val overrideServer = new TestableSocketServer(KafkaConfig.fromProps(props))
try {
- overrideServer.startup()
+ overrideServer.enableRequestProcessing(Map.empty)
val conn: Socket = connect(overrideServer)
overrideServer.testableProcessor.closeSocketOnSendResponse(conn)
val serializedBytes = producerRequestBytes()
@@ -1150,7 +1148,7 @@ class SocketServerTest {
val overrideServer = new TestableSocketServer(KafkaConfig.fromProps(props))
try {
- overrideServer.startup()
+ overrideServer.enableRequestProcessing(Map.empty)
val selector = overrideServer.testableSelector
// Create a channel, send some requests and close socket. Receive one pending request after socket was closed.
@@ -1178,7 +1176,7 @@ class SocketServerTest {
val overrideServer = new SocketServer(KafkaConfig.fromProps(props), serverMetrics,
Time.SYSTEM, credentialProvider, apiVersionManager)
try {
- overrideServer.startup()
+ overrideServer.enableRequestProcessing(Map.empty)
conn = connect(overrideServer)
val serializedBytes = producerRequestBytes()
sendRequest(conn, serializedBytes)
@@ -1559,7 +1557,7 @@ class SocketServerTest {
props.put(KafkaConfig.ConnectionsMaxIdleMsProp, idleTimeMs.toString)
props ++= sslServerProps
val testableServer = new TestableSocketServer(time = time)
- testableServer.startup()
+ testableServer.enableRequestProcessing(Map.empty)
assertTrue(testableServer.controlPlaneRequestChannelOpt.isEmpty)
@@ -1595,7 +1593,7 @@ class SocketServerTest {
val time = new MockTime()
props ++= sslServerProps
val testableServer = new TestableSocketServer(time = time)
- testableServer.startup()
+ testableServer.enableRequestProcessing(Map.empty)
val proxyServer = new ProxyServer(testableServer)
try {
val testableSelector = testableServer.testableSelector
@@ -1741,7 +1739,7 @@ class SocketServerTest {
val numConnections = 5
props.put("max.connections.per.ip", numConnections.toString)
val testableServer = new TestableSocketServer(KafkaConfig.fromProps(props), connectionQueueSize = 1)
- testableServer.startup()
+ testableServer.enableRequestProcessing(Map.empty)
val testableSelector = testableServer.testableSelector
val errors = new mutable.HashSet[String]
@@ -1893,7 +1891,9 @@ class SocketServerTest {
startProcessingRequests: Boolean = true): Unit = {
shutdownServerAndMetrics(server)
val testableServer = new TestableSocketServer(config)
- testableServer.startup(startProcessingRequests = startProcessingRequests)
+ if (startProcessingRequests) {
+ testableServer.enableRequestProcessing(Map.empty)
+ }
try {
testWithServer(testableServer)
} finally {
@@ -1998,7 +1998,8 @@ class SocketServerTest {
new LogContext(),
connectionQueueSize,
isPrivilegedListener,
- apiVersionManager) {
+ apiVersionManager,
+ s"TestableProcessor${id}") {
private var connectionId: Option[String] = None
private var conn: Option[Socket] = None
diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
index e5ff0cd1080..e879a7f6ffc 100755
--- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
@@ -1486,13 +1486,19 @@ class KafkaConfigTest {
assertEquals("3", originals.get(KafkaConfig.NodeIdProp))
}
- @Test
- def testBrokerIdIsInferredByNodeIdWithKraft(): Unit = {
+ def kraftProps(): Properties = {
val props = new Properties()
props.setProperty(KafkaConfig.ProcessRolesProp, "broker")
- props.put(KafkaConfig.ControllerListenerNamesProp, "SSL")
+ props.setProperty(KafkaConfig.ControllerListenerNamesProp, "CONTROLLER")
props.setProperty(KafkaConfig.NodeIdProp, "3")
props.setProperty(KafkaConfig.QuorumVotersProp, "1@localhost:9093")
+ props
+ }
+
+ @Test
+ def testBrokerIdIsInferredByNodeIdWithKraft(): Unit = {
+ val props = new Properties(kraftProps())
+ props.putAll(kraftProps())
val config = KafkaConfig.fromProps(props)
assertEquals(3, config.brokerId)
assertEquals(3, config.nodeId)
@@ -1527,4 +1533,40 @@ class KafkaConfigTest {
assertTrue(ce.getMessage.contains(KafkaConfig.InterBrokerSecurityProtocolProp))
}
+ @Test
+ def testEarlyStartListenersDefault(): Unit = {
+ val props = new Properties()
+ props.setProperty(KafkaConfig.ProcessRolesProp, "controller")
+ props.setProperty(KafkaConfig.ControllerListenerNamesProp, "CONTROLLER")
+ props.setProperty(KafkaConfig.ListenersProp, "CONTROLLER://:8092")
+ props.setProperty(KafkaConfig.NodeIdProp, "1")
+ props.setProperty(KafkaConfig.QuorumVotersProp, "1@localhost:9093")
+ val config = new KafkaConfig(props)
+ assertEquals(Set("CONTROLLER"), config.earlyStartListeners.map(_.value()))
+ }
+
+ @Test
+ def testEarlyStartListeners(): Unit = {
+ val props = new Properties()
+ props.putAll(kraftProps())
+ props.setProperty(KafkaConfig.EarlyStartListenersProp, "INTERNAL,INTERNAL2")
+ props.setProperty(KafkaConfig.InterBrokerListenerNameProp, "INTERNAL")
+ props.setProperty(KafkaConfig.ListenerSecurityProtocolMapProp,
+ "INTERNAL:PLAINTEXT,INTERNAL2:PLAINTEXT,CONTROLLER:PLAINTEXT")
+ props.setProperty(KafkaConfig.ListenersProp,
+ "INTERNAL://127.0.0.1:9092,INTERNAL2://127.0.0.1:9093")
+ val config = new KafkaConfig(props)
+ assertEquals(Set(new ListenerName("INTERNAL"), new ListenerName("INTERNAL2")),
+ config.earlyStartListeners)
+ }
+
+ @Test
+ def testEarlyStartListenersMustBeListeners(): Unit = {
+ val props = new Properties()
+ props.putAll(kraftProps())
+ props.setProperty(KafkaConfig.EarlyStartListenersProp, "INTERNAL")
+ assertEquals("early.start.listeners contains listener INTERNAL, but this is not " +
+ "contained in listeners or controller.listener.names",
+ assertThrows(classOf[ConfigException], () => new KafkaConfig(props)).getMessage)
+ }
}
diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
index 9ff6a599577..cea62c8f00b 100644
--- a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
+++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
@@ -167,6 +167,10 @@ public final class QuorumController implements Controller {
this.clusterId = clusterId;
}
+ public int nodeId() {
+ return nodeId;
+ }
+
public Builder setTime(Time time) {
this.time = time;
return this;
@@ -753,11 +757,11 @@ public final class QuorumController implements Controller {
}
class QuorumMetaLogListener implements RaftClient.Listener<ApiMessageAndVersion> {
-
@Override
public void handleCommit(BatchReader<ApiMessageAndVersion> reader) {
appendRaftEvent("handleCommit[baseOffset=" + reader.baseOffset() + "]", () -> {
try {
+ maybeCompleteAuthorizerInitialLoad();
boolean isActiveController = curClaimEpoch != -1;
long processedRecordsSize = 0;
while (reader.hasNext()) {
@@ -921,12 +925,35 @@ public final class QuorumController implements Controller {
if (this != metaLogListener) {
log.debug("Ignoring {} raft event from an old registration", name);
} else {
- runnable.run();
+ try {
+ runnable.run();
+ } finally {
+ maybeCompleteAuthorizerInitialLoad();
+ }
}
});
}
}
+ private void maybeCompleteAuthorizerInitialLoad() {
+ if (!needToCompleteAuthorizerLoad) return;
+ OptionalLong highWatermark = raftClient.highWatermark();
+ if (highWatermark.isPresent()) {
+ if (lastCommittedOffset + 1 >= highWatermark.getAsLong()) {
+ log.info("maybeCompleteAuthorizerInitialLoad: completing authorizer " +
+ "initial load at last committed offset {}.", lastCommittedOffset);
+ authorizer.get().completeInitialLoad();
+ needToCompleteAuthorizerLoad = false;
+ } else {
+ log.trace("maybeCompleteAuthorizerInitialLoad: can't proceed because " +
+ "lastCommittedOffset = {}, but highWatermark = {}.",
+ lastCommittedOffset, highWatermark.getAsLong());
+ }
+ } else {
+ log.trace("maybeCompleteAuthorizerInitialLoad: highWatermark not set.");
+ }
+ }
+
private void renounce() {
curClaimEpoch = -1;
controllerMetrics.setActive(false);
@@ -1276,6 +1303,12 @@ public final class QuorumController implements Controller {
*/
private long lastCommittedTimestamp = -1;
+ /**
+ * True if we need to complete the authorizer initial load.
+ * This must be accessed only by the event queue thread.
+ */
+ private boolean needToCompleteAuthorizerLoad;
+
/**
* If we have called scheduleWrite, this is the last offset we got back from it.
*/
@@ -1384,9 +1417,12 @@ public final class QuorumController implements Controller {
this.metaLogListener = new QuorumMetaLogListener();
this.curClaimEpoch = -1;
this.writeOffset = -1L;
+ this.needToCompleteAuthorizerLoad = authorizer.isPresent();
resetState();
+ log.info("Creating new QuorumController with clusterId {}, authorizer {}.", clusterId, authorizer);
+
this.raftClient.register(metaLogListener);
}
diff --git a/metadata/src/main/java/org/apache/kafka/metadata/authorizer/ClusterMetadataAuthorizer.java b/metadata/src/main/java/org/apache/kafka/metadata/authorizer/ClusterMetadataAuthorizer.java
index cae35283335..d7673022897 100644
--- a/metadata/src/main/java/org/apache/kafka/metadata/authorizer/ClusterMetadataAuthorizer.java
+++ b/metadata/src/main/java/org/apache/kafka/metadata/authorizer/ClusterMetadataAuthorizer.java
@@ -56,6 +56,18 @@ public interface ClusterMetadataAuthorizer extends Authorizer {
*/
AclMutator aclMutatorOrException();
+ /**
+ * Complete the initial load of the cluster metadata authorizer, so that all
+ * principals can use it.
+ */
+ void completeInitialLoad();
+
+ /**
+ * Complete the initial load of the cluster metadata authorizer with an exception,
+ * indicating that the loading process has failed.
+ */
+ void completeInitialLoad(Exception e);
+
/**
* Load the ACLs in the given map. Anything not in the map will be removed.
* The authorizer will also wait for this initial snapshot load to complete when
diff --git a/metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizer.java b/metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizer.java
index 510a6f87075..611625a4f3a 100644
--- a/metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizer.java
+++ b/metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizer.java
@@ -54,10 +54,9 @@ public class StandardAuthorizer implements ClusterMetadataAuthorizer {
public final static String ALLOW_EVERYONE_IF_NO_ACL_IS_FOUND_CONFIG = "allow.everyone.if.no.acl.found";
/**
- * A future which is completed once we have loaded a snapshot.
- * TODO: KAFKA-13649: StandardAuthorizer should not finish loading until it reads up to the high water mark.
+ * A future which is completed once we have loaded up to the initial high water mark.
*/
- private final CompletableFuture<Void> initialLoadFuture = CompletableFuture.completedFuture(null);
+ private final CompletableFuture<Void> initialLoadFuture = new CompletableFuture<>();
/**
* The current data. Can be read without a lock. Must be written while holding the object lock.
@@ -78,6 +77,24 @@ public class StandardAuthorizer implements ClusterMetadataAuthorizer {
return aclMutator;
}
+ @Override
+ public synchronized void completeInitialLoad() {
+ data = data.copyWithNewLoadingComplete(true);
+ data.log.info("Completed initial ACL load process.");
+ initialLoadFuture.complete(null);
+ }
+
+ // Visible for testing
+ public CompletableFuture<Void> initialLoadFuture() {
+ return initialLoadFuture;
+ }
+
+ @Override
+ public void completeInitialLoad(Exception e) {
+ data.log.error("Failed to complete initial ACL load process.", e);
+ initialLoadFuture.completeExceptionally(e);
+ }
+
@Override
public void addAcl(Uuid id, StandardAcl acl) {
data.addAcl(id, acl);
@@ -98,7 +115,12 @@ public class StandardAuthorizer implements ClusterMetadataAuthorizer {
AuthorizerServerInfo serverInfo) {
Map<Endpoint, CompletableFuture<Void>> result = new HashMap<>();
for (Endpoint endpoint : serverInfo.endpoints()) {
- result.put(endpoint, initialLoadFuture);
+ if (serverInfo.earlyStartListeners().contains(
+ endpoint.listenerName().orElseGet(() -> ""))) {
+ result.put(endpoint, CompletableFuture.completedFuture(null));
+ } else {
+ result.put(endpoint, initialLoadFuture);
+ }
}
return result;
}
@@ -131,7 +153,6 @@ public class StandardAuthorizer implements ClusterMetadataAuthorizer {
// Complete the initialLoadFuture, if it hasn't been completed already.
initialLoadFuture.completeExceptionally(new TimeoutException("The authorizer was " +
"closed before the initial load could complete."));
- // Nothing else to do here.
}
@Override
diff --git a/metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerData.java b/metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerData.java
index 27cca4271b8..b46c839e4fa 100644
--- a/metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerData.java
+++ b/metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerData.java
@@ -22,6 +22,7 @@ import org.apache.kafka.common.acl.AclBinding;
import org.apache.kafka.common.acl.AclBindingFilter;
import org.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.acl.AclPermissionType;
+import org.apache.kafka.common.errors.AuthorizerNotReadyException;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.resource.PatternType;
import org.apache.kafka.common.resource.ResourcePattern;
@@ -92,6 +93,11 @@ public class StandardAuthorizerData {
*/
final AclMutator aclMutator;
+ /**
+ * True if the authorizer loading process is complete.
+ */
+ final boolean loadingComplete;
+
/**
* A statically configured set of users that are authorized to do anything.
*/
@@ -123,6 +129,7 @@ public class StandardAuthorizerData {
static StandardAuthorizerData createEmpty() {
return new StandardAuthorizerData(createLogger(-1),
null,
+ false,
Collections.emptySet(),
DENIED,
new ConcurrentSkipListSet<>(), new ConcurrentHashMap<>());
@@ -130,6 +137,7 @@ public class StandardAuthorizerData {
private StandardAuthorizerData(Logger log,
AclMutator aclMutator,
+ boolean loadingComplete,
Set<String> superUsers,
AuthorizationResult defaultResult,
ConcurrentSkipListSet<StandardAcl> aclsByResource,
@@ -137,6 +145,7 @@ public class StandardAuthorizerData {
this.log = log;
this.auditLog = auditLogger();
this.aclMutator = aclMutator;
+ this.loadingComplete = loadingComplete;
this.superUsers = superUsers;
this.defaultRule = new DefaultRule(defaultResult);
this.aclsByResource = aclsByResource;
@@ -147,6 +156,17 @@ public class StandardAuthorizerData {
return new StandardAuthorizerData(
log,
newAclMutator,
+ loadingComplete,
+ superUsers,
+ defaultRule.result,
+ aclsByResource,
+ aclsById);
+ }
+
+ StandardAuthorizerData copyWithNewLoadingComplete(boolean newLoadingComplete) {
+ return new StandardAuthorizerData(log,
+ aclMutator,
+ newLoadingComplete,
superUsers,
defaultRule.result,
aclsByResource,
@@ -159,6 +179,7 @@ public class StandardAuthorizerData {
return new StandardAuthorizerData(
createLogger(nodeId),
aclMutator,
+ loadingComplete,
newSuperUsers,
newDefaultResult,
aclsByResource,
@@ -169,6 +190,7 @@ public class StandardAuthorizerData {
StandardAuthorizerData newData = new StandardAuthorizerData(
log,
aclMutator,
+ loadingComplete,
superUsers,
defaultRule.result,
new ConcurrentSkipListSet<>(),
@@ -250,6 +272,8 @@ public class StandardAuthorizerData {
// Superusers are authorized to do anything.
if (superUsers.contains(principal.toString())) {
rule = SuperUserRule.INSTANCE;
+ } else if (!loadingComplete) {
+ throw new AuthorizerNotReadyException();
} else {
MatchingAclRule aclRule = findAclRule(
matchingPrincipals(requestContext),
diff --git a/metadata/src/test/java/org/apache/kafka/controller/AclControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/AclControlManagerTest.java
index 08b362a7276..fdc03276451 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/AclControlManagerTest.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/AclControlManagerTest.java
@@ -148,6 +148,16 @@ public class AclControlManagerTest {
throw new NotControllerException("The current node is not the active controller.");
}
+ @Override
+ public void completeInitialLoad() {
+ // do nothing
+ }
+
+ @Override
+ public void completeInitialLoad(Exception e) {
+ // do nothing
+ }
+
@Override
public void loadSnapshot(Map<Uuid, StandardAcl> acls) {
this.acls = new HashMap<>(acls);
diff --git a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
index 9c679ad7d1e..148c5720b6d 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
@@ -17,6 +17,7 @@
package org.apache.kafka.controller;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -81,6 +82,7 @@ import org.apache.kafka.metadata.BrokerRegistrationReply;
import org.apache.kafka.metadata.MetadataRecordSerde;
import org.apache.kafka.metadata.PartitionRegistration;
import org.apache.kafka.metadata.RecordTestUtils;
+import org.apache.kafka.metadata.authorizer.StandardAuthorizer;
import org.apache.kafka.metalog.LocalLogManagerTestEnv;
import org.apache.kafka.raft.Batch;
import org.apache.kafka.server.common.ApiMessageAndVersion;
@@ -91,6 +93,7 @@ import org.apache.kafka.test.TestUtils;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
+import static java.util.function.Function.identity;
import static org.apache.kafka.clients.admin.AlterConfigOp.OpType.SET;
import static org.apache.kafka.common.config.ConfigResource.Type.BROKER;
import static org.apache.kafka.common.config.ConfigResource.Type.TOPIC;
@@ -1071,4 +1074,50 @@ public class QuorumControllerTest {
}
}
}
+
+ private static final Uuid FOO_ID = Uuid.fromString("igRktLOnR8ektWHr79F8mw");
+
+ private static final Map<Integer, Long> ALL_ZERO_BROKER_EPOCHS =
+ IntStream.of(0, 1, 2, 3).boxed().collect(Collectors.toMap(identity(), __ -> 0L));
+
+ @Test
+ public void testQuorumControllerCompletesAuthorizerInitialLoad() throws Throwable {
+ final int numControllers = 3;
+ List<StandardAuthorizer> authorizers = new ArrayList<>(numControllers);
+ for (int i = 0; i < numControllers; i++) {
+ StandardAuthorizer authorizer = new StandardAuthorizer();
+ authorizer.configure(Collections.emptyMap());
+ authorizers.add(authorizer);
+ }
+ try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(
+ numControllers,
+ Optional.empty(),
+ shared -> {
+ shared.setInitialMaxReadOffset(2);
+ }
+ )) {
+ logEnv.appendInitialRecords(expectedSnapshotContent(FOO_ID, ALL_ZERO_BROKER_EPOCHS));
+ logEnv.logManagers().forEach(m -> m.setMaxReadOffset(2));
+ try (QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv(logEnv, b -> {
+ b.setAuthorizer(authorizers.get(b.nodeId()));
+ })) {
+ assertInitialLoadFuturesNotComplete(authorizers);
+ logEnv.logManagers().get(0).setMaxReadOffset(Long.MAX_VALUE);
+ QuorumController active = controlEnv.activeController();
+ active.unregisterBroker(ANONYMOUS_CONTEXT, 3).get();
+ assertInitialLoadFuturesNotComplete(authorizers.stream().skip(1).collect(Collectors.toList()));
+ logEnv.logManagers().forEach(m -> m.setMaxReadOffset(Long.MAX_VALUE));
+ TestUtils.waitForCondition(() -> {
+ return authorizers.stream().allMatch(a -> a.initialLoadFuture().isDone());
+ }, "Failed to complete initial authorizer load for all controllers.");
+ }
+ }
+ }
+
+ private static void assertInitialLoadFuturesNotComplete(List<StandardAuthorizer> authorizers) {
+ for (int i = 0; i < authorizers.size(); i++) {
+ assertFalse(authorizers.get(i).initialLoadFuture().isDone(),
+ "authorizer " + i + " should not have completed loading.");
+ }
+ }
}
diff --git a/metadata/src/test/java/org/apache/kafka/metadata/authorizer/ClusterMetadataAuthorizerTest.java b/metadata/src/test/java/org/apache/kafka/metadata/authorizer/ClusterMetadataAuthorizerTest.java
index 19b07abed2c..3242170bcdf 100644
--- a/metadata/src/test/java/org/apache/kafka/metadata/authorizer/ClusterMetadataAuthorizerTest.java
+++ b/metadata/src/test/java/org/apache/kafka/metadata/authorizer/ClusterMetadataAuthorizerTest.java
@@ -106,6 +106,16 @@ public class ClusterMetadataAuthorizerTest {
return aclMutator;
}
+ @Override
+ public void completeInitialLoad() {
+ // do nothing
+ }
+
+ @Override
+ public void completeInitialLoad(Exception e) {
+ // do nothing
+ }
+
@Override
public void loadSnapshot(Map<Uuid, StandardAcl> acls) {
// do nothing
diff --git a/metadata/src/test/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerTest.java b/metadata/src/test/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerTest.java
index 7ed37785c42..987c00155c4 100644
--- a/metadata/src/test/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerTest.java
+++ b/metadata/src/test/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerTest.java
@@ -17,19 +17,25 @@
package org.apache.kafka.metadata.authorizer;
+import org.apache.kafka.common.ClusterResource;
+import org.apache.kafka.common.Endpoint;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.acl.AccessControlEntryFilter;
import org.apache.kafka.common.acl.AclBinding;
import org.apache.kafka.common.acl.AclBindingFilter;
import org.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.acl.AclPermissionType;
+import org.apache.kafka.common.errors.AuthorizerNotReadyException;
+import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.resource.PatternType;
import org.apache.kafka.common.resource.ResourcePattern;
import org.apache.kafka.common.resource.ResourcePatternFilter;
import org.apache.kafka.common.resource.ResourceType;
import org.apache.kafka.common.security.auth.KafkaPrincipal;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.server.authorizer.Action;
import org.apache.kafka.server.authorizer.AuthorizableRequestContext;
+import org.apache.kafka.server.authorizer.AuthorizerServerInfo;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.params.ParameterizedTest;
@@ -40,12 +46,17 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.InetAddress;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.CompletionStage;
import static java.util.Arrays.asList;
import static java.util.Collections.singletonList;
@@ -82,6 +93,56 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
@Timeout(value = 40)
public class StandardAuthorizerTest {
+ public static final Endpoint PLAINTEXT = new Endpoint("PLAINTEXT",
+ SecurityProtocol.PLAINTEXT,
+ "127.0.0.1",
+ 9020);
+
+ public static final Endpoint CONTROLLER = new Endpoint("CONTROLLER",
+ SecurityProtocol.PLAINTEXT,
+ "127.0.0.1",
+ 9020);
+
+ static class AuthorizerTestServerInfo implements AuthorizerServerInfo {
+ private final Collection<Endpoint> endpoints;
+
+ AuthorizerTestServerInfo(Collection<Endpoint> endpoints) {
+ assertFalse(endpoints.isEmpty());
+ this.endpoints = endpoints;
+ }
+
+ @Override
+ public ClusterResource clusterResource() {
+ return new ClusterResource(Uuid.fromString("r7mqHQrxTNmzbKvCvWZzLQ").toString());
+ }
+
+ @Override
+ public int brokerId() {
+ return 0;
+ }
+
+ @Override
+ public Collection<Endpoint> endpoints() {
+ return endpoints;
+ }
+
+ @Override
+ public Endpoint interBrokerEndpoint() {
+ return endpoints.iterator().next();
+ }
+
+ @Override
+ public Collection<String> earlyStartListeners() {
+ List<String> result = new ArrayList<>();
+ for (Endpoint endpoint : endpoints) {
+ if (endpoint.listenerName().get().equals("CONTROLLER")) {
+ result.add(endpoint.listenerName().get());
+ }
+ }
+ return result;
+ }
+ }
+
@Test
public void testGetConfiguredSuperUsers() {
assertEquals(Collections.emptySet(),
@@ -124,6 +185,16 @@ public class StandardAuthorizerTest {
new ResourcePattern(resourceType, resourceName, LITERAL), 1, false, false);
}
+ static StandardAuthorizer createAndInitializeStandardAuthorizer() {
+ StandardAuthorizer authorizer = new StandardAuthorizer();
+ authorizer.configure(Collections.singletonMap(SUPER_USERS_CONFIG, "User:superman"));
+ authorizer.start(new AuthorizerTestServerInfo(Collections.singletonList(PLAINTEXT)));
+ authorizer.completeInitialLoad();
+ return authorizer;
+ }
+
+ private final static AtomicLong NEXT_ID = new AtomicLong(0);
+
static StandardAcl newFooAcl(AclOperation op, AclPermissionType permission) {
return new StandardAcl(
TOPIC,
@@ -225,8 +296,7 @@ public class StandardAuthorizerTest {
@Test
public void testListAcls() throws Exception {
- StandardAuthorizer authorizer = new StandardAuthorizer();
- authorizer.configure(Collections.emptyMap());
+ StandardAuthorizer authorizer = createAndInitializeStandardAuthorizer();
List<StandardAclWithId> fooAcls = asList(
withId(newFooAcl(READ, ALLOW)),
withId(newFooAcl(WRITE, ALLOW)));
@@ -247,8 +317,7 @@ public class StandardAuthorizerTest {
@Test
public void testSimpleAuthorizations() throws Exception {
- StandardAuthorizer authorizer = new StandardAuthorizer();
- authorizer.configure(Collections.emptyMap());
+ StandardAuthorizer authorizer = createAndInitializeStandardAuthorizer();
List<StandardAclWithId> fooAcls = asList(
withId(newFooAcl(READ, ALLOW)),
withId(newFooAcl(WRITE, ALLOW)));
@@ -269,20 +338,17 @@ public class StandardAuthorizerTest {
@Test
public void testDenyPrecedenceWithOperationAll() throws Exception {
- StandardAuthorizer authorizer = new StandardAuthorizer();
- authorizer.configure(Collections.emptyMap());
+ StandardAuthorizer authorizer = createAndInitializeStandardAuthorizer();
List<StandardAcl> acls = Arrays.asList(
new StandardAcl(TOPIC, "foo", LITERAL, "User:alice", "*", ALL, DENY),
new StandardAcl(TOPIC, "foo", PREFIXED, "User:alice", "*", READ, ALLOW),
new StandardAcl(TOPIC, "foo", LITERAL, "User:*", "*", ALL, DENY),
new StandardAcl(TOPIC, "foo", PREFIXED, "User:*", "*", DESCRIBE, ALLOW)
);
-
acls.forEach(acl -> {
StandardAclWithId aclWithId = withId(acl);
authorizer.addAcl(aclWithId.id(), aclWithId.acl());
});
-
assertEquals(Arrays.asList(DENIED, DENIED, DENIED, ALLOWED), authorizer.authorize(
newRequestContext("alice"),
Arrays.asList(
@@ -290,7 +356,6 @@ public class StandardAuthorizerTest {
newAction(READ, TOPIC, "foo"),
newAction(DESCRIBE, TOPIC, "foo"),
newAction(READ, TOPIC, "foobar"))));
-
assertEquals(Arrays.asList(DENIED, DENIED, DENIED, ALLOWED, DENIED), authorizer.authorize(
newRequestContext("bob"),
Arrays.asList(
@@ -303,19 +368,16 @@ public class StandardAuthorizerTest {
@Test
public void testTopicAclWithOperationAll() throws Exception {
- StandardAuthorizer authorizer = new StandardAuthorizer();
- authorizer.configure(Collections.emptyMap());
+ StandardAuthorizer authorizer = createAndInitializeStandardAuthorizer();
List<StandardAcl> acls = Arrays.asList(
new StandardAcl(TOPIC, "foo", LITERAL, "User:*", "*", ALL, ALLOW),
new StandardAcl(TOPIC, "bar", PREFIXED, "User:alice", "*", ALL, ALLOW),
new StandardAcl(TOPIC, "baz", LITERAL, "User:bob", "*", ALL, ALLOW)
);
-
acls.forEach(acl -> {
StandardAclWithId aclWithId = withId(acl);
authorizer.addAcl(aclWithId.id(), aclWithId.acl());
});
-
assertEquals(Arrays.asList(ALLOWED, ALLOWED, DENIED), authorizer.authorize(
newRequestContext("alice"),
Arrays.asList(
@@ -349,8 +411,7 @@ public class StandardAuthorizerTest {
InetAddress host1 = InetAddress.getByName("192.168.1.1");
InetAddress host2 = InetAddress.getByName("192.168.1.2");
- StandardAuthorizer authorizer = new StandardAuthorizer();
- authorizer.configure(Collections.emptyMap());
+ StandardAuthorizer authorizer = createAndInitializeStandardAuthorizer();
List<StandardAcl> acls = Arrays.asList(
new StandardAcl(TOPIC, "foo", LITERAL, "User:alice", host1.getHostAddress(), READ, DENY),
new StandardAcl(TOPIC, "foo", LITERAL, "User:alice", "*", READ, ALLOW),
@@ -395,9 +456,7 @@ public class StandardAuthorizerTest {
.build();
}
- private static StandardAuthorizer createAuthorizerWithManyAcls() {
- StandardAuthorizer authorizer = new StandardAuthorizer();
- authorizer.configure(Collections.emptyMap());
+ private static void addManyAcls(StandardAuthorizer authorizer) {
List<StandardAcl> acls = Arrays.asList(
new StandardAcl(TOPIC, "green2", LITERAL, "User:*", "*", READ, ALLOW),
new StandardAcl(TOPIC, "green", PREFIXED, "User:bob", "*", READ, ALLOW),
@@ -413,12 +472,12 @@ public class StandardAuthorizerTest {
StandardAclWithId aclWithId = withId(acl);
authorizer.addAcl(aclWithId.id(), aclWithId.acl());
});
- return authorizer;
}
@Test
public void testAuthorizationWithManyAcls() throws Exception {
- StandardAuthorizer authorizer = createAuthorizerWithManyAcls();
+ StandardAuthorizer authorizer = createAndInitializeStandardAuthorizer();
+ addManyAcls(authorizer);
assertEquals(Arrays.asList(ALLOWED, DENIED),
authorizer.authorize(new MockAuthorizableRequestContext.Builder().
setPrincipal(new KafkaPrincipal(USER_TYPE, "bob")).build(),
@@ -449,7 +508,8 @@ public class StandardAuthorizerTest {
Mockito.when(auditLog.isDebugEnabled()).thenReturn(true);
Mockito.when(auditLog.isTraceEnabled()).thenReturn(true);
- StandardAuthorizer authorizer = createAuthorizerWithManyAcls();
+ StandardAuthorizer authorizer = createAndInitializeStandardAuthorizer();
+ addManyAcls(authorizer);
ResourcePattern topicResource = new ResourcePattern(TOPIC, "alpha", LITERAL);
Action action = new Action(READ, topicResource, 1, false, logIfDenied);
MockAuthorizableRequestContext requestContext = new MockAuthorizableRequestContext.Builder()
@@ -490,7 +550,8 @@ public class StandardAuthorizerTest {
Mockito.when(auditLog.isDebugEnabled()).thenReturn(true);
Mockito.when(auditLog.isTraceEnabled()).thenReturn(true);
- StandardAuthorizer authorizer = createAuthorizerWithManyAcls();
+ StandardAuthorizer authorizer = createAndInitializeStandardAuthorizer();
+ addManyAcls(authorizer);
ResourcePattern topicResource = new ResourcePattern(TOPIC, "green1", LITERAL);
Action action = new Action(READ, topicResource, 1, logIfAllowed, false);
MockAuthorizableRequestContext requestContext = new MockAuthorizableRequestContext.Builder()
@@ -514,4 +575,69 @@ public class StandardAuthorizerTest {
}
}
+ /**
+ * Test that StandardAuthorizer#start returns a completed future for early start
+ * listeners.
+ */
+ @Test
+ public void testStartWithEarlyStartListeners() throws Exception {
+ StandardAuthorizer authorizer = new StandardAuthorizer();
+ authorizer.configure(Collections.singletonMap(SUPER_USERS_CONFIG, "User:superman"));
+ Map<Endpoint, ? extends CompletionStage<Void>> futures2 = authorizer.
+ start(new AuthorizerTestServerInfo(Arrays.asList(PLAINTEXT, CONTROLLER)));
+ assertEquals(new HashSet<>(Arrays.asList(PLAINTEXT, CONTROLLER)), futures2.keySet());
+ assertFalse(futures2.get(PLAINTEXT).toCompletableFuture().isDone());
+ assertTrue(futures2.get(CONTROLLER).toCompletableFuture().isDone());
+ }
+
+ /**
+ * Test attempts to authorize prior to completeInitialLoad. During this time, only
+ * superusers can be authorized. Other users will get an AuthorizerNotReadyException
+ * exception. Not even an authorization result, just an exception thrown for the whole
+ * batch.
+ */
+ @Test
+ public void testAuthorizationPriorToCompleteInitialLoad() throws Exception {
+ StandardAuthorizer authorizer = new StandardAuthorizer();
+ authorizer.configure(Collections.singletonMap(SUPER_USERS_CONFIG, "User:superman"));
+ assertThrows(AuthorizerNotReadyException.class, () ->
+ authorizer.authorize(new MockAuthorizableRequestContext.Builder().
+ setPrincipal(new KafkaPrincipal(USER_TYPE, "bob")).build(),
+ Arrays.asList(newAction(READ, TOPIC, "green1"),
+ newAction(READ, TOPIC, "green2"))));
+ assertEquals(Arrays.asList(ALLOWED, ALLOWED),
+ authorizer.authorize(new MockAuthorizableRequestContext.Builder().
+ setPrincipal(new KafkaPrincipal(USER_TYPE, "superman")).build(),
+ Arrays.asList(newAction(READ, TOPIC, "green1"),
+ newAction(WRITE, GROUP, "wheel"))));
+ }
+
+ @Test
+ public void testCompleteInitialLoad() throws Exception {
+ StandardAuthorizer authorizer = new StandardAuthorizer();
+ authorizer.configure(Collections.singletonMap(SUPER_USERS_CONFIG, "User:superman"));
+ Map<Endpoint, ? extends CompletionStage<Void>> futures = authorizer.
+ start(new AuthorizerTestServerInfo(Collections.singleton(PLAINTEXT)));
+ assertEquals(Collections.singleton(PLAINTEXT), futures.keySet());
+ assertFalse(futures.get(PLAINTEXT).toCompletableFuture().isDone());
+ authorizer.completeInitialLoad();
+ assertTrue(futures.get(PLAINTEXT).toCompletableFuture().isDone());
+ assertFalse(futures.get(PLAINTEXT).toCompletableFuture().isCompletedExceptionally());
+ }
+
+ @Test
+ public void testCompleteInitialLoadWithException() throws Exception {
+ StandardAuthorizer authorizer = new StandardAuthorizer();
+ authorizer.configure(Collections.singletonMap(SUPER_USERS_CONFIG, "User:superman"));
+ Map<Endpoint, ? extends CompletionStage<Void>> futures = authorizer.
+ start(new AuthorizerTestServerInfo(Arrays.asList(PLAINTEXT, CONTROLLER)));
+ assertEquals(new HashSet<>(Arrays.asList(PLAINTEXT, CONTROLLER)), futures.keySet());
+ assertFalse(futures.get(PLAINTEXT).toCompletableFuture().isDone());
+ assertTrue(futures.get(CONTROLLER).toCompletableFuture().isDone());
+ authorizer.completeInitialLoad(new TimeoutException("timed out"));
+ assertTrue(futures.get(PLAINTEXT).toCompletableFuture().isDone());
+ assertTrue(futures.get(PLAINTEXT).toCompletableFuture().isCompletedExceptionally());
+ assertTrue(futures.get(CONTROLLER).toCompletableFuture().isDone());
+ assertFalse(futures.get(CONTROLLER).toCompletableFuture().isCompletedExceptionally());
+ }
}
diff --git a/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java b/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java
index 855fd468cba..3c495adff92 100644
--- a/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java
+++ b/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java
@@ -77,10 +77,10 @@ public final class LocalLogManager implements RaftClient<ApiMessageAndVersion>,
int size();
}
- static class LeaderChangeBatch implements LocalBatch {
+ public static class LeaderChangeBatch implements LocalBatch {
private final LeaderAndEpoch newLeader;
- LeaderChangeBatch(LeaderAndEpoch newLeader) {
+ public LeaderChangeBatch(LeaderAndEpoch newLeader) {
this.newLeader = newLeader;
}
@@ -113,12 +113,12 @@ public final class LocalLogManager implements RaftClient<ApiMessageAndVersion>,
}
}
- static class LocalRecordBatch implements LocalBatch {
+ public static class LocalRecordBatch implements LocalBatch {
private final int leaderEpoch;
private final long appendTimestamp;
private final List<ApiMessageAndVersion> records;
- LocalRecordBatch(int leaderEpoch, long appendTimestamp, List<ApiMessageAndVersion> records) {
+ public LocalRecordBatch(int leaderEpoch, long appendTimestamp, List<ApiMessageAndVersion> records) {
this.leaderEpoch = leaderEpoch;
this.appendTimestamp = appendTimestamp;
this.records = records;
@@ -184,6 +184,11 @@ public final class LocalLogManager implements RaftClient<ApiMessageAndVersion>,
*/
private long prevOffset;
+ /**
+ * The initial max read offset which LocalLog instances will be configured with.
+ */
+ private long initialMaxReadOffset = Long.MAX_VALUE;
+
/**
* Maps committed offset to snapshot reader.
*/
@@ -237,7 +242,12 @@ public final class LocalLogManager implements RaftClient<ApiMessageAndVersion>,
return offset;
}
- synchronized long append(LocalBatch batch) {
+ public synchronized long append(LocalBatch batch) {
+ try {
+ throw new RuntimeException("foo");
+ } catch (Exception e) {
+ log.info("WATERMELON: appending {}", batch, e);
+ }
prevOffset += batch.size();
log.debug("append(batch={}, prevOffset={})", batch, prevOffset);
batches.put(prevOffset, batch);
@@ -352,6 +362,15 @@ public final class LocalLogManager implements RaftClient<ApiMessageAndVersion>,
})
.sum();
}
+
+ public SharedLogData setInitialMaxReadOffset(long initialMaxReadOffset) {
+ this.initialMaxReadOffset = initialMaxReadOffset;
+ return this;
+ }
+
+ public long initialMaxReadOffset() {
+ return initialMaxReadOffset;
+ }
}
private static class MetaLogListenerData {
@@ -454,6 +473,7 @@ public final class LocalLogManager implements RaftClient<ApiMessageAndVersion>,
this.log = logContext.logger(LocalLogManager.class);
this.nodeId = nodeId;
this.shared = shared;
+ this.maxReadOffset = shared.initialMaxReadOffset();
this.eventQueue = new KafkaEventQueue(Time.SYSTEM, logContext, threadNamePrefix);
shared.registerLogManager(this);
}
@@ -502,6 +522,8 @@ public final class LocalLogManager implements RaftClient<ApiMessageAndVersion>,
// Only notify the listener if it equals the shared leader state
LeaderAndEpoch sharedLeader = shared.leaderAndEpoch();
if (batch.newLeader.equals(sharedLeader)) {
+ log.debug("Node {}: Executing handleLeaderChange {}",
+ nodeId, sharedLeader);
listenerData.handleLeaderChange(entryOffset, batch.newLeader);
if (batch.newLeader.epoch() > leader.epoch()) {
leader = batch.newLeader;
@@ -658,6 +680,15 @@ public final class LocalLogManager implements RaftClient<ApiMessageAndVersion>,
});
}
+ @Override
+ public synchronized OptionalLong highWatermark() {
+ if (shared.prevOffset > 0) {
+ return OptionalLong.of(shared.prevOffset);
+ } else {
+ return OptionalLong.empty();
+ }
+ }
+
@Override
public long scheduleAppend(int epoch, List<ApiMessageAndVersion> batch) {
if (batch.isEmpty()) {
diff --git a/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTestEnv.java b/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTestEnv.java
index ed18ab6053d..17c9c467124 100644
--- a/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTestEnv.java
+++ b/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTestEnv.java
@@ -20,8 +20,11 @@ package org.apache.kafka.metalog;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.metalog.LocalLogManager.LeaderChangeBatch;
+import org.apache.kafka.metalog.LocalLogManager.LocalRecordBatch;
import org.apache.kafka.metalog.LocalLogManager.SharedLogData;
import org.apache.kafka.raft.LeaderAndEpoch;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
import org.apache.kafka.snapshot.RawSnapshotReader;
import org.apache.kafka.test.TestUtils;
import org.slf4j.Logger;
@@ -32,7 +35,9 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
+import java.util.OptionalInt;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
public class LocalLogManagerTestEnv implements AutoCloseable {
private static final Logger log =
@@ -77,10 +82,15 @@ public class LocalLogManagerTestEnv implements AutoCloseable {
return testEnv;
}
- public LocalLogManagerTestEnv(int numManagers, Optional<RawSnapshotReader> snapshot) throws Exception {
+ public LocalLogManagerTestEnv(
+ int numManagers,
+ Optional<RawSnapshotReader> snapshot,
+ Consumer<SharedLogData> dataSetup
+ ) throws Exception {
clusterId = Uuid.randomUuid().toString();
dir = TestUtils.tempDirectory();
shared = new SharedLogData(snapshot);
+ dataSetup.accept(shared);
List<LocalLogManager> newLogManagers = new ArrayList<>(numManagers);
try {
for (int nodeId = 0; nodeId < numManagers; nodeId++) {
@@ -102,6 +112,28 @@ public class LocalLogManagerTestEnv implements AutoCloseable {
this.logManagers = newLogManagers;
}
+ public LocalLogManagerTestEnv(
+ int numManagers,
+ Optional<RawSnapshotReader> snapshot
+ ) throws Exception {
+ this(numManagers, snapshot, __ -> { });
+ }
+
+ /**
+ * Append some records to the log. This method is meant to be called before the
+ * controllers are started, to simulate a pre-existing metadata log.
+ *
+ * @param records The records to be appended. Will be added in a single batch.
+ */
+ public void appendInitialRecords(List<ApiMessageAndVersion> records) {
+ int initialLeaderEpoch = 1;
+ shared.append(new LeaderChangeBatch(
+ new LeaderAndEpoch(OptionalInt.empty(), initialLeaderEpoch + 1)));
+ shared.append(new LocalRecordBatch(initialLeaderEpoch + 1, 0, records));
+ shared.append(new LeaderChangeBatch(
+ new LeaderAndEpoch(OptionalInt.of(0), initialLeaderEpoch + 2)));
+ }
+
public String clusterId() {
return clusterId;
}
diff --git a/raft/src/main/java/org/apache/kafka/raft/RaftClient.java b/raft/src/main/java/org/apache/kafka/raft/RaftClient.java
index 8e4f50e7488..952cc60a387 100644
--- a/raft/src/main/java/org/apache/kafka/raft/RaftClient.java
+++ b/raft/src/main/java/org/apache/kafka/raft/RaftClient.java
@@ -24,6 +24,7 @@ import org.apache.kafka.snapshot.SnapshotWriter;
import java.util.List;
import java.util.Optional;
import java.util.OptionalInt;
+import java.util.OptionalLong;
import java.util.concurrent.CompletableFuture;
public interface RaftClient<T> extends AutoCloseable {
@@ -113,6 +114,11 @@ public interface RaftClient<T> extends AutoCloseable {
*/
void unregister(Listener<T> listener);
+ /**
+ * Returns the current high water mark, or OptionalLong.empty if it is not known.
+ */
+ OptionalLong highWatermark();
+
/**
* Return the current {@link LeaderAndEpoch}.
*