From a2ad74b959e6f4954f88da85f8fbda735acc1be8 Mon Sep 17 00:00:00 2001 From: abbccdda Date: Mon, 1 Mar 2021 18:05:19 -0800 Subject: [PATCH 1/5] only return leader not available for internal topic creation --- .../server/AutoTopicCreationManager.scala | 6 +++- .../server/AutoTopicCreationManagerTest.scala | 32 ++++++++++++++++--- 2 files changed, 32 insertions(+), 6 deletions(-) diff --git a/core/src/main/scala/kafka/server/AutoTopicCreationManager.scala b/core/src/main/scala/kafka/server/AutoTopicCreationManager.scala index 01dabedf75428..b82a8a32adbac 100644 --- a/core/src/main/scala/kafka/server/AutoTopicCreationManager.scala +++ b/core/src/main/scala/kafka/server/AutoTopicCreationManager.scala @@ -275,7 +275,11 @@ class DefaultAutoTopicCreationManager( val validationError: Option[Errors] = if (!isValidTopicName(topic)) { Some(Errors.INVALID_TOPIC_EXCEPTION) } else if (!hasEnoughLiveBrokers(topic, aliveBrokers)) { - Some(Errors.INVALID_REPLICATION_FACTOR) + if (Topic.isInternal(topic)) { + Some(Errors.INVALID_REPLICATION_FACTOR) + } else { + Some(Errors.LEADER_NOT_AVAILABLE) + } } else if (!inflightTopics.add(topic)) { Some(Errors.UNKNOWN_TOPIC_OR_PARTITION) } else { diff --git a/core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala b/core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala index 9f9749bba66dc..4a18e67414027 100644 --- a/core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala +++ b/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 @@ -146,9 +147,26 @@ class AutoTopicCreationManagerTest { } @Test - def testNotEnoughLiveBrokers(): Unit = { + def testNotEnoughLiveBrokersForNonInternalTopics(): Unit = { + testNotEnoughLiveBrokers("topic", KafkaConfig.DefaultReplicationFactorProp) + } + + @Test + def testNotEnoughLiveBrokersForConsumerOffsetsTopic(): Unit = { + Mockito.when(groupCoordinator.offsetsTopicConfigs).thenReturn(new Properties) + testNotEnoughLiveBrokers(Topic.GROUP_METADATA_TOPIC_NAME, + KafkaConfig.OffsetsTopicReplicationFactorProp) + } + + @Test + def testNotEnoughLiveBrokersForTxnOffsetTopic(): Unit = { + testNotEnoughLiveBrokers(Topic.TRANSACTION_STATE_TOPIC_NAME, + KafkaConfig.TransactionsTopicReplicationFactorProp) + } + + private def testNotEnoughLiveBrokers(topic: String, replicationConfig: String): Unit = { val props = TestUtils.createBrokerConfig(1, "localhost") - props.setProperty(KafkaConfig.DefaultReplicationFactorProp, 3.toString) + props.setProperty(replicationConfig, 3.toString) config = KafkaConfig.fromProps(props) autoTopicCreationManager = new DefaultAutoTopicCreationManager( @@ -160,11 +178,15 @@ class AutoTopicCreationManagerTest { groupCoordinator, transactionCoordinator) - val topicName = "topic" - Mockito.when(controller.isActive).thenReturn(false) - createTopicAndVerifyResult(Errors.INVALID_REPLICATION_FACTOR, topicName, false) + val isInternal = Topic.isInternal(topic) + val expectedError = if (isInternal) + Errors.INVALID_REPLICATION_FACTOR + else + Errors.LEADER_NOT_AVAILABLE + + createTopicAndVerifyResult(expectedError, topic, isInternal) Mockito.verify(brokerToController, Mockito.never()).sendRequest( any(), From d3b5160de58d4fe5ad7f8488504530202163af55 Mon Sep 17 00:00:00 2001 From: abbccdda Date: Wed, 3 Mar 2021 09:58:13 -0800 Subject: [PATCH 2/5] remove entire logic --- .../server/AutoTopicCreationManager.scala | 37 +----- .../scala/kafka/server/BrokerServer.scala | 2 +- .../server/AutoTopicCreationManagerTest.scala | 121 ++++++++++++++---- 3 files changed, 96 insertions(+), 64 deletions(-) diff --git a/core/src/main/scala/kafka/server/AutoTopicCreationManager.scala b/core/src/main/scala/kafka/server/AutoTopicCreationManager.scala index b82a8a32adbac..f8063a820d57b 100644 --- a/core/src/main/scala/kafka/server/AutoTopicCreationManager.scala +++ b/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,7 +264,6 @@ 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] @@ -274,12 +271,6 @@ class DefaultAutoTopicCreationManager( // 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)) { - if (Topic.isInternal(topic)) { - Some(Errors.INVALID_REPLICATION_FACTOR) - } else { - Some(Errors.LEADER_NOT_AVAILABLE) - } } else if (!inflightTopics.add(topic)) { Some(Errors.UNKNOWN_TOPIC_OR_PARTITION) } else { @@ -299,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 - } - } - } diff --git a/core/src/main/scala/kafka/server/BrokerServer.scala b/core/src/main/scala/kafka/server/BrokerServer.scala index 4c0aaab27aefc..e7f7a125146d2 100644 --- a/core/src/main/scala/kafka/server/BrokerServer.scala +++ b/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() diff --git a/core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala b/core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala index 4a18e67414027..9bb13eabfb7d3 100644 --- a/core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala +++ b/core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala @@ -34,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} @@ -95,7 +96,6 @@ class AutoTopicCreationManagerTest { replicationFactor: Short = 1): Unit = { autoTopicCreationManager = new DefaultAutoTopicCreationManager( config, - metadataCache, Some(brokerToController), Some(adminManager), Some(controller), @@ -124,7 +124,6 @@ class AutoTopicCreationManagerTest { def testCreateTopicsWithForwardingDisabled(): Unit = { autoTopicCreationManager = new DefaultAutoTopicCreationManager( config, - metadataCache, None, Some(adminManager), Some(controller), @@ -147,50 +146,118 @@ class AutoTopicCreationManagerTest { } @Test - def testNotEnoughLiveBrokersForNonInternalTopics(): Unit = { - testNotEnoughLiveBrokers("topic", KafkaConfig.DefaultReplicationFactorProp) + def testInvalidReplicationFactorForNonInternalTopics(): Unit = { + testErrorWithCreationInZk(Errors.INVALID_REPLICATION_FACTOR, "topic", isInternal = false) } @Test - def testNotEnoughLiveBrokersForConsumerOffsetsTopic(): Unit = { + def testInvalidReplicationFactorForConsumerOffsetsTopic(): Unit = { Mockito.when(groupCoordinator.offsetsTopicConfigs).thenReturn(new Properties) - testNotEnoughLiveBrokers(Topic.GROUP_METADATA_TOPIC_NAME, - KafkaConfig.OffsetsTopicReplicationFactorProp) + testErrorWithCreationInZk(Errors.INVALID_REPLICATION_FACTOR, Topic.GROUP_METADATA_TOPIC_NAME, isInternal = true) } @Test - def testNotEnoughLiveBrokersForTxnOffsetTopic(): Unit = { - testNotEnoughLiveBrokers(Topic.TRANSACTION_STATE_TOPIC_NAME, - KafkaConfig.TransactionsTopicReplicationFactorProp) + def testInvalidReplicationFactorForTxnOffsetTopic(): Unit = { + Mockito.when(transactionCoordinator.transactionTopicConfigs).thenReturn(new Properties) + testErrorWithCreationInZk(Errors.INVALID_REPLICATION_FACTOR, Topic.TRANSACTION_STATE_TOPIC_NAME, isInternal = true) } - private def testNotEnoughLiveBrokers(topic: String, replicationConfig: String): Unit = { - val props = TestUtils.createBrokerConfig(1, "localhost") - props.setProperty(replicationConfig, 3.toString) - config = KafkaConfig.fromProps(props) + @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) 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) + }) - val isInternal = Topic.isInternal(topic) - val expectedError = if (isInternal) - Errors.INVALID_REPLICATION_FACTOR - else - Errors.LEADER_NOT_AVAILABLE - - createTopicAndVerifyResult(expectedError, topic, isInternal) - - 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, From 435920fd36fd80331d7c7ffe381c86cc7b851653 Mon Sep 17 00:00:00 2001 From: abbccdda Date: Thu, 4 Mar 2021 10:06:45 -0800 Subject: [PATCH 3/5] test with invalid replication factor --- tests/kafkatest/tests/core/security_test.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/kafkatest/tests/core/security_test.py b/tests/kafkatest/tests/core/security_test.py index 0ce12c9c71ba7..cb7e485a4f782 100644 --- a/tests/kafkatest/tests/core/security_test.py +++ b/tests/kafkatest/tests/core/security_test.py @@ -65,7 +65,7 @@ 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. Hence clients are expected to fail with INVALID_REPLICATION_FACTOR. """ # Start Kafka with valid hostnames in the certs' SANs so that we can create the test topic via the admin client @@ -111,7 +111,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() From 62bfcfc0c97b5bb6ad3a798829c90378ad8e1ba9 Mon Sep 17 00:00:00 2001 From: abbccdda Date: Thu, 4 Mar 2021 19:49:40 -0800 Subject: [PATCH 4/5] additional comments for system test --- tests/kafkatest/tests/core/security_test.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/tests/kafkatest/tests/core/security_test.py b/tests/kafkatest/tests/core/security_test.py index cb7e485a4f782..8e2b695074d97 100644 --- a/tests/kafkatest/tests/core/security_test.py +++ b/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 INVALID_REPLICATION_FACTOR. + 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 From 33288941d44e0e2543d7523ac90c038bd661dada Mon Sep 17 00:00:00 2001 From: abbccdda Date: Fri, 5 Mar 2021 11:29:45 -0800 Subject: [PATCH 5/5] test nit --- .../server/AutoTopicCreationManagerTest.scala | 17 ++++++++--------- 1 file changed, 8 insertions(+), 9 deletions(-) diff --git a/core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala b/core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala index 9bb13eabfb7d3..0bd97c36e258d 100644 --- a/core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala +++ b/core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala @@ -165,41 +165,41 @@ class AutoTopicCreationManagerTest { @Test def testTopicExistsErrorSwapForNonInternalTopics(): Unit = { testErrorWithCreationInZk(Errors.TOPIC_ALREADY_EXISTS, "topic", isInternal = false, - expectedError = Errors.LEADER_NOT_AVAILABLE) + expectedError = Some(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) + expectedError = Some(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) + expectedError = Some(Errors.LEADER_NOT_AVAILABLE)) } @Test def testRequestTimeoutErrorSwapForNonInternalTopics(): Unit = { testErrorWithCreationInZk(Errors.REQUEST_TIMED_OUT, "topic", isInternal = false, - expectedError = Errors.LEADER_NOT_AVAILABLE) + expectedError = Some(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) + expectedError = Some(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) + expectedError = Some(Errors.LEADER_NOT_AVAILABLE)) } @Test @@ -222,7 +222,7 @@ class AutoTopicCreationManagerTest { private def testErrorWithCreationInZk(error: Errors, topicName: String, isInternal: Boolean, - expectedError: Errors = null): Unit = { + expectedError: Option[Errors] = None): Unit = { autoTopicCreationManager = new DefaultAutoTopicCreationManager( config, None, @@ -256,8 +256,7 @@ class AutoTopicCreationManagerTest { .apply(topicErrors) }) - val errorToVerify = if (expectedError != null) expectedError else error - createTopicAndVerifyResult(errorToVerify, topicName, isInternal = isInternal) + createTopicAndVerifyResult(expectedError.getOrElse(error), topicName, isInternal = isInternal) } private def createTopicAndVerifyResult(error: Errors,