KAFKA-12381: remove live broker checks for forwarding topic creation

core/src/main/scala/kafka/server/AutoTopicCreationManager.scala

@@ -24,7 +24,6 @@ import java.util.concurrent.atomic.AtomicReference
 import kafka.controller.KafkaController
 import kafka.coordinator.group.GroupCoordinator
 import kafka.coordinator.transaction.TransactionCoordinator
-import kafka.server.metadata.MetadataBroker
 import kafka.utils.Logging
 import org.apache.kafka.clients.ClientResponse
 import org.apache.kafka.common.errors.InvalidTopicException
@@ -82,14 +81,13 @@ object AutoTopicCreationManager {
         ))
       else
         None
-    new DefaultAutoTopicCreationManager(config, metadataCache, channelManager, adminManager,
+    new DefaultAutoTopicCreationManager(config, channelManager, adminManager,
       controller, groupCoordinator, txnCoordinator)
   }
 }
 
 class DefaultAutoTopicCreationManager(
   config: KafkaConfig,
-  metadataCache: MetadataCache,
   channelManager: Option[BrokerToControllerChannelManager],
   adminManager: Option[ZkAdminManager],
   controller: Option[KafkaController],
@@ -266,16 +264,13 @@ class DefaultAutoTopicCreationManager(
     topics: Set[String]
   ): (Map[String, CreatableTopic], Seq[MetadataResponseTopic]) = {
 
-    val aliveBrokers = metadataCache.getAliveBrokers
     val creatableTopics = mutable.Map.empty[String, CreatableTopic]
     val uncreatableTopics = mutable.Buffer.empty[MetadataResponseTopic]
 
     topics.foreach { topic =>
       // Attempt basic topic validation before sending any requests to the controller.
       val validationError: Option[Errors] = if (!isValidTopicName(topic)) {
         Some(Errors.INVALID_TOPIC_EXCEPTION)
-      } else if (!hasEnoughLiveBrokers(topic, aliveBrokers)) {
-        Some(Errors.INVALID_REPLICATION_FACTOR)
       } else if (!inflightTopics.add(topic)) {
         Some(Errors.UNKNOWN_TOPIC_OR_PARTITION)
       } else {
@@ -295,30 +290,4 @@ class DefaultAutoTopicCreationManager(
 
     (creatableTopics, uncreatableTopics)
   }
-
-  private def hasEnoughLiveBrokers(
-    topicName: String,
-    aliveBrokers: Seq[MetadataBroker]
-  ): Boolean = {
-    val (replicationFactor, replicationFactorConfig) = topicName match {
-      case GROUP_METADATA_TOPIC_NAME =>
-        (config.offsetsTopicReplicationFactor.intValue, KafkaConfig.OffsetsTopicReplicationFactorProp)
-
-      case TRANSACTION_STATE_TOPIC_NAME =>
-        (config.transactionTopicReplicationFactor.intValue, KafkaConfig.TransactionsTopicReplicationFactorProp)
-
-      case _ =>
-        (config.defaultReplicationFactor, KafkaConfig.DefaultReplicationFactorProp)
-    }
-
-    if (aliveBrokers.size < replicationFactor) {
-      error(s"Number of alive brokers '${aliveBrokers.size}' does not meet the required replication factor " +
-        s"'$replicationFactor' for auto creation of topic '$topicName' which is configured by $replicationFactorConfig. " +
-        "This error can be ignored if the cluster is starting up and not all brokers are up yet.")
-      false
-    } else {
-      true
-    }
-  }
-
 }

core/src/main/scala/kafka/server/BrokerServer.scala

@@ -248,7 +248,7 @@ class BrokerServer(
       val autoTopicCreationChannelManager = BrokerToControllerChannelManager(controllerNodeProvider,
         time, metrics, config, "autocreate", threadNamePrefix, 60000)
       autoTopicCreationManager = new DefaultAutoTopicCreationManager(
-        config, metadataCache, Some(autoTopicCreationChannelManager), None, None,
+        config, Some(autoTopicCreationChannelManager), None, None,
         groupCoordinator, transactionCoordinator)
       autoTopicCreationManager.start()
 

core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala

@@ -24,6 +24,7 @@ import kafka.coordinator.group.GroupCoordinator
 import kafka.coordinator.transaction.TransactionCoordinator
 import kafka.utils.TestUtils
 import kafka.utils.TestUtils.createBroker
+import org.apache.kafka.common.internals.Topic
 import org.apache.kafka.common.internals.Topic.{GROUP_METADATA_TOPIC_NAME, TRANSACTION_STATE_TOPIC_NAME}
 import org.apache.kafka.common.message.CreateTopicsRequestData
 import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic
@@ -33,6 +34,7 @@ import org.apache.kafka.common.requests._
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.{BeforeEach, Test}
 import org.mockito.ArgumentMatchers.any
+import org.mockito.invocation.InvocationOnMock
 import org.mockito.{ArgumentMatchers, Mockito}
 
 import scala.collection.{Map, Seq}
@@ -94,7 +96,6 @@ class AutoTopicCreationManagerTest {
                               replicationFactor: Short = 1): Unit = {
     autoTopicCreationManager = new DefaultAutoTopicCreationManager(
       config,
-      metadataCache,
       Some(brokerToController),
       Some(adminManager),
       Some(controller),
@@ -123,7 +124,6 @@ class AutoTopicCreationManagerTest {
   def testCreateTopicsWithForwardingDisabled(): Unit = {
     autoTopicCreationManager = new DefaultAutoTopicCreationManager(
       config,
-      metadataCache,
       None,
       Some(adminManager),
       Some(controller),
@@ -146,29 +146,118 @@ class AutoTopicCreationManagerTest {
   }
 
   @Test
-  def testNotEnoughLiveBrokers(): Unit = {
-    val props = TestUtils.createBrokerConfig(1, "localhost")
-    props.setProperty(KafkaConfig.DefaultReplicationFactorProp, 3.toString)
-    config = KafkaConfig.fromProps(props)
+  def testInvalidReplicationFactorForNonInternalTopics(): Unit = {
+    testErrorWithCreationInZk(Errors.INVALID_REPLICATION_FACTOR, "topic", isInternal = false)
+  }
+
+  @Test
+  def testInvalidReplicationFactorForConsumerOffsetsTopic(): Unit = {
+    Mockito.when(groupCoordinator.offsetsTopicConfigs).thenReturn(new Properties)
+    testErrorWithCreationInZk(Errors.INVALID_REPLICATION_FACTOR, Topic.GROUP_METADATA_TOPIC_NAME, isInternal = true)
+  }
+
+  @Test
+  def testInvalidReplicationFactorForTxnOffsetTopic(): Unit = {
+    Mockito.when(transactionCoordinator.transactionTopicConfigs).thenReturn(new Properties)
+    testErrorWithCreationInZk(Errors.INVALID_REPLICATION_FACTOR, Topic.TRANSACTION_STATE_TOPIC_NAME, isInternal = true)
+  }
+
+  @Test
+  def testTopicExistsErrorSwapForNonInternalTopics(): Unit = {
+    testErrorWithCreationInZk(Errors.TOPIC_ALREADY_EXISTS, "topic", isInternal = false,
+      expectedError = Errors.LEADER_NOT_AVAILABLE)
+  }
+
+  @Test
+  def testTopicExistsErrorSwapForConsumerOffsetsTopic(): Unit = {
+    Mockito.when(groupCoordinator.offsetsTopicConfigs).thenReturn(new Properties)
+    testErrorWithCreationInZk(Errors.TOPIC_ALREADY_EXISTS, Topic.GROUP_METADATA_TOPIC_NAME, isInternal = true,
+      expectedError = Errors.LEADER_NOT_AVAILABLE)
+  }
+
+  @Test
+  def testTopicExistsErrorSwapForTxnOffsetTopic(): Unit = {
+    Mockito.when(transactionCoordinator.transactionTopicConfigs).thenReturn(new Properties)
+    testErrorWithCreationInZk(Errors.TOPIC_ALREADY_EXISTS, Topic.TRANSACTION_STATE_TOPIC_NAME, isInternal = true,
+      expectedError = Errors.LEADER_NOT_AVAILABLE)
+  }
+
+  @Test
+  def testRequestTimeoutErrorSwapForNonInternalTopics(): Unit = {
+    testErrorWithCreationInZk(Errors.REQUEST_TIMED_OUT, "topic", isInternal = false,
+      expectedError = Errors.LEADER_NOT_AVAILABLE)
+  }
+
+  @Test
+  def testRequestTimeoutErrorSwapForConsumerOffsetTopic(): Unit = {
+    Mockito.when(groupCoordinator.offsetsTopicConfigs).thenReturn(new Properties)
+    testErrorWithCreationInZk(Errors.REQUEST_TIMED_OUT, Topic.GROUP_METADATA_TOPIC_NAME, isInternal = true,
+      expectedError = Errors.LEADER_NOT_AVAILABLE)
+  }
+
+  @Test
+  def testRequestTimeoutErrorSwapForTxnOffsetTopic(): Unit = {
+    Mockito.when(transactionCoordinator.transactionTopicConfigs).thenReturn(new Properties)
+    testErrorWithCreationInZk(Errors.REQUEST_TIMED_OUT, Topic.TRANSACTION_STATE_TOPIC_NAME, isInternal = true,
+      expectedError = Errors.LEADER_NOT_AVAILABLE)
+  }
+
+  @Test
+  def testUnknownTopicPartitionForNonIntervalTopic(): Unit = {
+    testErrorWithCreationInZk(Errors.UNKNOWN_TOPIC_OR_PARTITION, "topic", isInternal = false)
+  }
 
+  @Test
+  def testUnknownTopicPartitionForConsumerOffsetTopic(): Unit = {
+    Mockito.when(groupCoordinator.offsetsTopicConfigs).thenReturn(new Properties)
+    testErrorWithCreationInZk(Errors.UNKNOWN_TOPIC_OR_PARTITION, Topic.GROUP_METADATA_TOPIC_NAME, isInternal = true)
+  }
+
+  @Test
+  def testUnknownTopicPartitionForTxnOffsetTopic(): Unit = {
+    Mockito.when(transactionCoordinator.transactionTopicConfigs).thenReturn(new Properties)
+    testErrorWithCreationInZk(Errors.UNKNOWN_TOPIC_OR_PARTITION, Topic.TRANSACTION_STATE_TOPIC_NAME, isInternal = true)
+  }
+
+  private def testErrorWithCreationInZk(error: Errors,
+                                        topicName: String,
+                                        isInternal: Boolean,
+                                        expectedError: Errors = null): Unit = {
     autoTopicCreationManager = new DefaultAutoTopicCreationManager(
       config,
-      metadataCache,
-      Some(brokerToController),
+      None,
       Some(adminManager),
       Some(controller),
       groupCoordinator,
       transactionCoordinator)
 
-    val topicName = "topic"
-
     Mockito.when(controller.isActive).thenReturn(false)
+    val newTopic = if (isInternal) {
+      topicName match {
+        case Topic.GROUP_METADATA_TOPIC_NAME => getNewTopic(topicName,
+          numPartitions = config.offsetsTopicPartitions, replicationFactor = config.offsetsTopicReplicationFactor)
+        case Topic.TRANSACTION_STATE_TOPIC_NAME => getNewTopic(topicName,
+          numPartitions = config.transactionTopicPartitions, replicationFactor = config.transactionTopicReplicationFactor)
+      }
+    } else {
+      getNewTopic(topicName)
+    }
+
+    val topicErrors = if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) null else
+      Map(topicName -> new ApiError(error))
+    Mockito.when(adminManager.createTopics(
+      ArgumentMatchers.eq(0),
+      ArgumentMatchers.eq(false),
+      ArgumentMatchers.eq(Map(topicName -> newTopic)),
+      ArgumentMatchers.eq(Map.empty),
+      any(classOf[ControllerMutationQuota]),
+      any(classOf[Map[String, ApiError] => Unit]))).thenAnswer((invocation: InvocationOnMock) => {
+      invocation.getArgument(5).asInstanceOf[Map[String, ApiError] => Unit]
+        .apply(topicErrors)
+    })
 
-    createTopicAndVerifyResult(Errors.INVALID_REPLICATION_FACTOR, topicName, false)
-
-    Mockito.verify(brokerToController, Mockito.never()).sendRequest(
-      any(),
-      any(classOf[ControllerRequestCompletionHandler]))
+    val errorToVerify = if (expectedError != null) expectedError else error
+    createTopicAndVerifyResult(errorToVerify, topicName, isInternal = isInternal)
   }
 
   private def createTopicAndVerifyResult(error: Errors,

tests/kafkatest/tests/core/security_test.py

@@ -65,7 +65,10 @@ def test_client_ssl_endpoint_validation_failure(self, security_protocol, interbr
         Test that invalid hostname in certificate results in connection failures.
         When security_protocol=SSL, client SSL handshakes are expected to fail due to hostname verification failure.
         When security_protocol=PLAINTEXT and interbroker_security_protocol=SSL, controller connections fail
-        with hostname verification failure. Hence clients are expected to fail with LEADER_NOT_AVAILABLE.
+        with hostname verification failure. Since metadata cannot be propagated in the cluster without a valid certificate,
+        the broker's metadata caches will be empty. Hence we expect Metadata requests to fail with an INVALID_REPLICATION_FACTOR
+        error since the broker will attempt to create the topic automatically as it does not exist in the metadata cache,
+        and there will be no online brokers.
         """
 
         # Start Kafka with valid hostnames in the certs' SANs so that we can create the test topic via the admin client
@@ -111,7 +114,7 @@ def test_client_ssl_endpoint_validation_failure(self, security_protocol, interbr
                 # expected
                 pass
 
-            error = 'SSLHandshakeException' if security_protocol == 'SSL' else 'LEADER_NOT_AVAILABLE'
+            error = 'SSLHandshakeException' if security_protocol == 'SSL' else 'INVALID_REPLICATION_FACTOR'
             wait_until(lambda: self.producer_consumer_have_expected_error(error), timeout_sec=30)
             self.producer.stop()
             self.consumer.stop()