diff --git a/core/src/test/java/kafka/server/handlers/DescribeTopicPartitionsRequestHandlerTest.java b/core/src/test/java/kafka/server/handlers/DescribeTopicPartitionsRequestHandlerTest.java index 4db64639df0de..e48e830175bdd 100644 --- a/core/src/test/java/kafka/server/handlers/DescribeTopicPartitionsRequestHandlerTest.java +++ b/core/src/test/java/kafka/server/handlers/DescribeTopicPartitionsRequestHandlerTest.java @@ -535,7 +535,6 @@ KafkaConfig createKafkaDefaultConfig() { int brokerId = 1; Properties properties = TestUtils.createBrokerConfig( brokerId, - null, true, true, TestUtils.RandomPort(), diff --git a/core/src/test/scala/integration/kafka/admin/ListOffsetsIntegrationTest.scala b/core/src/test/scala/integration/kafka/admin/ListOffsetsIntegrationTest.scala index 54d73c9d1e68d..37348c0657862 100644 --- a/core/src/test/scala/integration/kafka/admin/ListOffsetsIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/admin/ListOffsetsIntegrationTest.scala @@ -276,7 +276,7 @@ class ListOffsetsIntegrationTest extends KafkaServerTestHarness { } def generateConfigs: Seq[KafkaConfig] = { - TestUtils.createBrokerConfigs(2, null).zipWithIndex.map{ case (props, index) => + TestUtils.createBrokerConfigs(2).zipWithIndex.map{ case (props, index) => // We use mock timer so the records can get removed if the test env is too busy to complete // tests before kafka-log-retention. Hence, we disable the retention to avoid failed tests props.setProperty(ServerLogConfigs.LOG_RETENTION_TIME_MILLIS_CONFIG, "-1") diff --git a/core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala b/core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala index e24731261ac63..d89a83c7750f2 100644 --- a/core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala +++ b/core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala @@ -472,7 +472,7 @@ class RemoteTopicCrudTest extends IntegrationTestHarness { TestUtils.createTopicWithAdmin(createAdminClient(), testTopicName, brokers, controllerServers, numPartitions, brokerCount, topicConfig = topicConfig) - val tsDisabledProps = TestUtils.createBrokerConfigs(1, null).head + val tsDisabledProps = TestUtils.createBrokerConfigs(1).head instanceConfigs = List(KafkaConfig.fromProps(tsDisabledProps)) recreateBrokers(startup = true) @@ -490,7 +490,7 @@ class RemoteTopicCrudTest extends IntegrationTestHarness { TestUtils.createTopicWithAdmin(createAdminClient(), testTopicName, brokers, controllerServers, numPartitions, brokerCount, topicConfig = topicConfig) - val tsDisabledProps = TestUtils.createBrokerConfigs(1, null).head + val tsDisabledProps = TestUtils.createBrokerConfigs(1).head instanceConfigs = List(KafkaConfig.fromProps(tsDisabledProps)) recreateBrokers(startup = true) diff --git a/core/src/test/scala/integration/kafka/api/AdminClientWithPoliciesIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/AdminClientWithPoliciesIntegrationTest.scala index d6d74c5b94156..b11cb96ef8ce0 100644 --- a/core/src/test/scala/integration/kafka/api/AdminClientWithPoliciesIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/api/AdminClientWithPoliciesIntegrationTest.scala @@ -64,7 +64,7 @@ class AdminClientWithPoliciesIntegrationTest extends KafkaServerTestHarness with Map[String, Object](AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG -> bootstrapServers()).asJava override def generateConfigs: collection.Seq[KafkaConfig] = { - val configs = TestUtils.createBrokerConfigs(brokerCount, null) + val configs = TestUtils.createBrokerConfigs(brokerCount) configs.foreach(overrideNodeConfigs) configs.map(KafkaConfig.fromProps) } diff --git a/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala b/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala index c8c36730f2cd4..74111e319b0ab 100644 --- a/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala +++ b/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala @@ -55,7 +55,6 @@ abstract class BaseProducerSendTest extends KafkaServerTestHarness { overridingProps.put(ServerLogConfigs.NUM_PARTITIONS_CONFIG, 4.toString) TestUtils.createBrokerConfigs( numServers, - null, interBrokerSecurityProtocol = Some(securityProtocol), trustStoreFile = trustStoreFile, saslProperties = serverSaslProperties diff --git a/core/src/test/scala/integration/kafka/api/ConsumerBounceTest.scala b/core/src/test/scala/integration/kafka/api/ConsumerBounceTest.scala index a3825d28afefc..a8dbe0ecdaa5d 100644 --- a/core/src/test/scala/integration/kafka/api/ConsumerBounceTest.scala +++ b/core/src/test/scala/integration/kafka/api/ConsumerBounceTest.scala @@ -77,7 +77,7 @@ class ConsumerBounceTest extends AbstractConsumerTest with Logging { private def generateKafkaConfigs(maxGroupSize: String = maxGroupSize.toString): Seq[KafkaConfig] = { val properties = new Properties testConfigs.foreachEntry((k, v) => properties.setProperty(k, v)) - FixedPortTestUtils.createBrokerConfigs(brokerCount, null, enableControlledShutdown = false) + FixedPortTestUtils.createBrokerConfigs(brokerCount, enableControlledShutdown = false) .map(KafkaConfig.fromProps(_, properties)) } diff --git a/core/src/test/scala/integration/kafka/api/CustomQuotaCallbackTest.scala b/core/src/test/scala/integration/kafka/api/CustomQuotaCallbackTest.scala index a18c03fb593f9..fd383bf0809ad 100644 --- a/core/src/test/scala/integration/kafka/api/CustomQuotaCallbackTest.scala +++ b/core/src/test/scala/integration/kafka/api/CustomQuotaCallbackTest.scala @@ -38,6 +38,7 @@ import java.util.Properties import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger} import java.{lang, util} +import scala.collection.Seq import scala.collection.mutable.ArrayBuffer import scala.jdk.CollectionConverters._ @@ -178,8 +179,7 @@ class CustomQuotaCallbackTest extends IntegrationTestHarness with SaslSetup { } private def createTopic(topic: String, numPartitions: Int, leader: Int): Unit = { - val assignment = (0 until numPartitions).map { i => i -> Seq(leader) }.toMap - TestUtils.createTopic(null, topic, assignment, servers) + // TODO createTopic } private def createAdminClient(): Admin = { diff --git a/core/src/test/scala/integration/kafka/api/EndToEndClusterIdTest.scala b/core/src/test/scala/integration/kafka/api/EndToEndClusterIdTest.scala index 0cc927bd11ddc..c423c95ae9d89 100644 --- a/core/src/test/scala/integration/kafka/api/EndToEndClusterIdTest.scala +++ b/core/src/test/scala/integration/kafka/api/EndToEndClusterIdTest.scala @@ -102,7 +102,7 @@ class EndToEndClusterIdTest extends KafkaServerTestHarness { this.serverConfig.setProperty(MetricConfigs.METRIC_REPORTER_CLASSES_CONFIG, classOf[MockBrokerMetricsReporter].getName) override def generateConfigs = { - val cfgs = TestUtils.createBrokerConfigs(serverCount, null, interBrokerSecurityProtocol = Some(securityProtocol), + val cfgs = TestUtils.createBrokerConfigs(serverCount, interBrokerSecurityProtocol = Some(securityProtocol), trustStoreFile = trustStoreFile, saslProperties = serverSaslProperties) cfgs.foreach(_ ++= serverConfig) cfgs.map(KafkaConfig.fromProps) diff --git a/core/src/test/scala/integration/kafka/api/FixedPortTestUtils.scala b/core/src/test/scala/integration/kafka/api/FixedPortTestUtils.scala index bf5f8c1e6a3ec..cb0d5d0ee0f55 100644 --- a/core/src/test/scala/integration/kafka/api/FixedPortTestUtils.scala +++ b/core/src/test/scala/integration/kafka/api/FixedPortTestUtils.scala @@ -39,12 +39,11 @@ object FixedPortTestUtils { } def createBrokerConfigs(numConfigs: Int, - zkConnect: String, enableControlledShutdown: Boolean = true, enableDeleteTopic: Boolean = false): Seq[Properties] = { val ports = FixedPortTestUtils.choosePorts(numConfigs) (0 until numConfigs).map { node => - TestUtils.createBrokerConfig(node, zkConnect, enableControlledShutdown, enableDeleteTopic, ports(node)) + TestUtils.createBrokerConfig(node, enableControlledShutdown, enableDeleteTopic, ports(node)) } } diff --git a/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala b/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala index b660ff3521936..ed5611eb74b84 100644 --- a/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala +++ b/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala @@ -66,7 +66,7 @@ abstract class IntegrationTestHarness extends KafkaServerTestHarness { } override def generateConfigs: Seq[KafkaConfig] = { - val cfgs = TestUtils.createBrokerConfigs(brokerCount, null, interBrokerSecurityProtocol = Some(securityProtocol), + val cfgs = TestUtils.createBrokerConfigs(brokerCount, interBrokerSecurityProtocol = Some(securityProtocol), trustStoreFile = trustStoreFile, saslProperties = serverSaslProperties, logDirCount = logDirCount) configureListeners(cfgs) modifyConfigs(cfgs) diff --git a/core/src/test/scala/integration/kafka/api/ProducerCompressionTest.scala b/core/src/test/scala/integration/kafka/api/ProducerCompressionTest.scala index 3137575289233..2782a46f18abf 100755 --- a/core/src/test/scala/integration/kafka/api/ProducerCompressionTest.scala +++ b/core/src/test/scala/integration/kafka/api/ProducerCompressionTest.scala @@ -48,7 +48,7 @@ class ProducerCompressionTest extends QuorumTestHarness { @BeforeEach override def setUp(testInfo: TestInfo): Unit = { super.setUp(testInfo) - val props = TestUtils.createBrokerConfig(brokerId, null) + val props = TestUtils.createBrokerConfig(brokerId) broker = createBroker(new KafkaConfig(props)) } diff --git a/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala b/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala index 92eac4f1230aa..1826df1c6dc7b 100644 --- a/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala +++ b/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala @@ -52,7 +52,7 @@ class ProducerFailureHandlingTest extends KafkaServerTestHarness { overridingProps.put(GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, 1.toString) def generateConfigs = - TestUtils.createBrokerConfigs(numServers, null, enableControlledShutdown = false).map(KafkaConfig.fromProps(_, overridingProps)) + TestUtils.createBrokerConfigs(numServers, enableControlledShutdown = false).map(KafkaConfig.fromProps(_, overridingProps)) private var producer1: KafkaProducer[Array[Byte], Array[Byte]] = _ private var producer2: KafkaProducer[Array[Byte], Array[Byte]] = _ diff --git a/core/src/test/scala/integration/kafka/api/ProducerIdExpirationTest.scala b/core/src/test/scala/integration/kafka/api/ProducerIdExpirationTest.scala index 2dee826dc9f37..6f50b60aa15d9 100644 --- a/core/src/test/scala/integration/kafka/api/ProducerIdExpirationTest.scala +++ b/core/src/test/scala/integration/kafka/api/ProducerIdExpirationTest.scala @@ -52,7 +52,7 @@ class ProducerIdExpirationTest extends KafkaServerTestHarness { var admin: Admin = _ override def generateConfigs: Seq[KafkaConfig] = { - TestUtils.createBrokerConfigs(3, null).map(KafkaConfig.fromProps(_, serverProps())) + TestUtils.createBrokerConfigs(3).map(KafkaConfig.fromProps(_, serverProps())) } @BeforeEach diff --git a/core/src/test/scala/integration/kafka/api/RackAwareAutoTopicCreationTest.scala b/core/src/test/scala/integration/kafka/api/RackAwareAutoTopicCreationTest.scala index 9bd23de7137b0..03a312d5f077a 100644 --- a/core/src/test/scala/integration/kafka/api/RackAwareAutoTopicCreationTest.scala +++ b/core/src/test/scala/integration/kafka/api/RackAwareAutoTopicCreationTest.scala @@ -45,7 +45,7 @@ class RackAwareAutoTopicCreationTest extends KafkaServerTestHarness with RackAwa def generateConfigs = (0 until numServers) map { node => - TestUtils.createBrokerConfig(node, null, enableControlledShutdown = false, rack = Some((node / 2).toString)) + TestUtils.createBrokerConfig(node, enableControlledShutdown = false, rack = Some((node / 2).toString)) } map (KafkaConfig.fromProps(_, overridingProps)) private val topic = "topic" diff --git a/core/src/test/scala/integration/kafka/api/RebootstrapTest.scala b/core/src/test/scala/integration/kafka/api/RebootstrapTest.scala index 5d3134a0870d4..68982405370b4 100644 --- a/core/src/test/scala/integration/kafka/api/RebootstrapTest.scala +++ b/core/src/test/scala/integration/kafka/api/RebootstrapTest.scala @@ -36,7 +36,7 @@ abstract class RebootstrapTest extends AbstractConsumerTest { // In this test, fixed ports are necessary, because brokers must have the // same port after the restart. - FixedPortTestUtils.createBrokerConfigs(brokerCount, null, enableControlledShutdown = false) + FixedPortTestUtils.createBrokerConfigs(brokerCount, enableControlledShutdown = false) .map(KafkaConfig.fromProps(_, overridingProps)) } diff --git a/core/src/test/scala/integration/kafka/api/TransactionsBounceTest.scala b/core/src/test/scala/integration/kafka/api/TransactionsBounceTest.scala index 4f6fb7e483ce3..a6cd0d905decd 100644 --- a/core/src/test/scala/integration/kafka/api/TransactionsBounceTest.scala +++ b/core/src/test/scala/integration/kafka/api/TransactionsBounceTest.scala @@ -69,7 +69,7 @@ class TransactionsBounceTest extends IntegrationTestHarness { // Since such quick rotation of servers is incredibly unrealistic, we allow this one test to preallocate ports, leaving // a small risk of hitting errors due to port conflicts. Hopefully this is infrequent enough to not cause problems. override def generateConfigs = { - FixedPortTestUtils.createBrokerConfigs(brokerCount, null) + FixedPortTestUtils.createBrokerConfigs(brokerCount) .map(KafkaConfig.fromProps(_, overridingProps)) } diff --git a/core/src/test/scala/integration/kafka/api/TransactionsExpirationTest.scala b/core/src/test/scala/integration/kafka/api/TransactionsExpirationTest.scala index 26ed880aa1af6..2449bcc986bc2 100644 --- a/core/src/test/scala/integration/kafka/api/TransactionsExpirationTest.scala +++ b/core/src/test/scala/integration/kafka/api/TransactionsExpirationTest.scala @@ -51,7 +51,7 @@ class TransactionsExpirationTest extends KafkaServerTestHarness { var admin: Admin = _ override def generateConfigs: Seq[KafkaConfig] = { - TestUtils.createBrokerConfigs(3, null).map(KafkaConfig.fromProps(_, serverProps())) + TestUtils.createBrokerConfigs(3).map(KafkaConfig.fromProps(_, serverProps())) } @BeforeEach diff --git a/core/src/test/scala/integration/kafka/api/TransactionsWithMaxInFlightOneTest.scala b/core/src/test/scala/integration/kafka/api/TransactionsWithMaxInFlightOneTest.scala index fe1ea323162c5..c59997bd37c8a 100644 --- a/core/src/test/scala/integration/kafka/api/TransactionsWithMaxInFlightOneTest.scala +++ b/core/src/test/scala/integration/kafka/api/TransactionsWithMaxInFlightOneTest.scala @@ -50,7 +50,7 @@ class TransactionsWithMaxInFlightOneTest extends KafkaServerTestHarness { val transactionalConsumers = mutable.Buffer[Consumer[Array[Byte], Array[Byte]]]() override def generateConfigs: Seq[KafkaConfig] = { - TestUtils.createBrokerConfigs(numBrokers, null).map(KafkaConfig.fromProps(_, serverProps())) + TestUtils.createBrokerConfigs(numBrokers).map(KafkaConfig.fromProps(_, serverProps())) } @BeforeEach diff --git a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala index 5700375309f02..545e4359fdc4d 100644 --- a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala +++ b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala @@ -118,7 +118,7 @@ class DynamicBrokerReconfigurationTest extends QuorumTestHarness with SaslSetup (0 until numServers).foreach { brokerId => - val props = TestUtils.createBrokerConfig(brokerId, null) + val props = TestUtils.createBrokerConfig(brokerId) props.put(SocketServerConfigs.ADVERTISED_LISTENERS_CONFIG, s"$SecureInternal://localhost:0, $SecureExternal://localhost:0") props ++= securityProps(sslProperties1, TRUSTSTORE_PROPS) // Ensure that we can support multiple listeners per security protocol and multiple security protocols diff --git a/core/src/test/scala/integration/kafka/server/FetchFromFollowerIntegrationTest.scala b/core/src/test/scala/integration/kafka/server/FetchFromFollowerIntegrationTest.scala index 31f3663359ecd..a9961c7c48225 100644 --- a/core/src/test/scala/integration/kafka/server/FetchFromFollowerIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/server/FetchFromFollowerIntegrationTest.scala @@ -52,7 +52,7 @@ class FetchFromFollowerIntegrationTest extends BaseFetchRequestTest { } override def generateConfigs: collection.Seq[KafkaConfig] = { - TestUtils.createBrokerConfigs(numNodes, null, enableControlledShutdown = false, enableFetchFromFollower = true) + TestUtils.createBrokerConfigs(numNodes, enableControlledShutdown = false, enableFetchFromFollower = true) .map(KafkaConfig.fromProps(_, overridingProps)) } diff --git a/core/src/test/scala/integration/kafka/server/MultipleListenersWithSameSecurityProtocolBaseTest.scala b/core/src/test/scala/integration/kafka/server/MultipleListenersWithSameSecurityProtocolBaseTest.scala index ec44af7b2364d..db2d570b761b7 100644 --- a/core/src/test/scala/integration/kafka/server/MultipleListenersWithSameSecurityProtocolBaseTest.scala +++ b/core/src/test/scala/integration/kafka/server/MultipleListenersWithSameSecurityProtocolBaseTest.scala @@ -80,7 +80,7 @@ abstract class MultipleListenersWithSameSecurityProtocolBaseTest extends QuorumT (0 until numServers).foreach { brokerId => - val props = TestUtils.createBrokerConfig(brokerId, null, trustStoreFile = Some(trustStoreFile)) + val props = TestUtils.createBrokerConfig(brokerId, trustStoreFile = Some(trustStoreFile)) // Ensure that we can support multiple listeners per security protocol and multiple security protocols props.put(SocketServerConfigs.LISTENERS_CONFIG, s"$SecureInternal://localhost:0, $Internal://localhost:0, " + s"$SecureExternal://localhost:0, $External://localhost:0") diff --git a/core/src/test/scala/kafka/server/LocalLeaderEndPointTest.scala b/core/src/test/scala/kafka/server/LocalLeaderEndPointTest.scala index 3b71ca01ba2db..95545a38a62dd 100644 --- a/core/src/test/scala/kafka/server/LocalLeaderEndPointTest.scala +++ b/core/src/test/scala/kafka/server/LocalLeaderEndPointTest.scala @@ -56,7 +56,7 @@ class LocalLeaderEndPointTest extends Logging { @BeforeEach def setUp(): Unit = { - val props = TestUtils.createBrokerConfig(sourceBroker.id, null, port = sourceBroker.port) + val props = TestUtils.createBrokerConfig(sourceBroker.id, port = sourceBroker.port) val config = KafkaConfig.fromProps(props) val mockLogMgr = TestUtils.createLogManager(config.logDirs.map(new File(_))) val alterPartitionManager = mock(classOf[AlterPartitionManager]) diff --git a/core/src/test/scala/kafka/server/NodeToControllerRequestThreadTest.scala b/core/src/test/scala/kafka/server/NodeToControllerRequestThreadTest.scala index 7a58f4148266d..8c820e41da439 100644 --- a/core/src/test/scala/kafka/server/NodeToControllerRequestThreadTest.scala +++ b/core/src/test/scala/kafka/server/NodeToControllerRequestThreadTest.scala @@ -50,7 +50,7 @@ class NodeToControllerRequestThreadTest { @Test def testRetryTimeoutWhileControllerNotAvailable(): Unit = { val time = new MockTime() - val config = new KafkaConfig(TestUtils.createBrokerConfig(1, "localhost:2181")) + val config = new KafkaConfig(TestUtils.createBrokerConfig(1)) val metadata = mock(classOf[Metadata]) val mockClient = new MockClient(time, metadata) val controllerNodeProvider = mock(classOf[ControllerNodeProvider]) @@ -84,7 +84,7 @@ class NodeToControllerRequestThreadTest { def testRequestsSent(): Unit = { // just a simple test that tests whether the request from 1 -> 2 is sent and the response callback is called val time = new MockTime() - val config = new KafkaConfig(TestUtils.createBrokerConfig(1, "localhost:2181")) + val config = new KafkaConfig(TestUtils.createBrokerConfig(1)) val controllerId = 2 val metadata = mock(classOf[Metadata]) @@ -125,7 +125,7 @@ class NodeToControllerRequestThreadTest { def testControllerChanged(): Unit = { // in this test the current broker is 1, and the controller changes from 2 -> 3 then back: 3 -> 2 val time = new MockTime() - val config = new KafkaConfig(TestUtils.createBrokerConfig(1, "localhost:2181")) + val config = new KafkaConfig(TestUtils.createBrokerConfig(1)) val oldControllerId = 1 val newControllerId = 2 @@ -173,7 +173,7 @@ class NodeToControllerRequestThreadTest { @Test def testNotController(): Unit = { val time = new MockTime() - val config = new KafkaConfig(TestUtils.createBrokerConfig(1, "localhost:2181")) + val config = new KafkaConfig(TestUtils.createBrokerConfig(1)) val oldControllerId = 1 val newControllerId = 2 @@ -233,7 +233,7 @@ class NodeToControllerRequestThreadTest { @Test def testEnvelopeResponseWithNotControllerError(): Unit = { val time = new MockTime() - val config = new KafkaConfig(TestUtils.createBrokerConfig(1, "localhost:2181")) + val config = new KafkaConfig(TestUtils.createBrokerConfig(1)) val oldControllerId = 1 val newControllerId = 2 @@ -307,7 +307,7 @@ class NodeToControllerRequestThreadTest { @Test def testRetryTimeout(): Unit = { val time = new MockTime() - val config = new KafkaConfig(TestUtils.createBrokerConfig(1, "localhost:2181")) + val config = new KafkaConfig(TestUtils.createBrokerConfig(1)) val controllerId = 1 val metadata = mock(classOf[Metadata]) @@ -356,7 +356,7 @@ class NodeToControllerRequestThreadTest { @Test def testUnsupportedVersionHandling(): Unit = { val time = new MockTime() - val config = new KafkaConfig(TestUtils.createBrokerConfig(1, "localhost:2181")) + val config = new KafkaConfig(TestUtils.createBrokerConfig(1)) val controllerId = 2 val metadata = mock(classOf[Metadata]) @@ -394,7 +394,7 @@ class NodeToControllerRequestThreadTest { @Test def testAuthenticationExceptionHandling(): Unit = { val time = new MockTime() - val config = new KafkaConfig(TestUtils.createBrokerConfig(1, "localhost:2181")) + val config = new KafkaConfig(TestUtils.createBrokerConfig(1)) val controllerId = 2 val metadata = mock(classOf[Metadata]) @@ -434,7 +434,7 @@ class NodeToControllerRequestThreadTest { def testThreadNotStarted(): Unit = { // Make sure we throw if we enqueue anything while the thread is not running val time = new MockTime() - val config = new KafkaConfig(TestUtils.createBrokerConfig(1, "localhost:2181")) + val config = new KafkaConfig(TestUtils.createBrokerConfig(1)) val metadata = mock(classOf[Metadata]) val mockClient = new MockClient(time, metadata) diff --git a/core/src/test/scala/kafka/server/RemoteLeaderEndPointTest.scala b/core/src/test/scala/kafka/server/RemoteLeaderEndPointTest.scala index 514ca0a1b28bf..0b1198a64c321 100644 --- a/core/src/test/scala/kafka/server/RemoteLeaderEndPointTest.scala +++ b/core/src/test/scala/kafka/server/RemoteLeaderEndPointTest.scala @@ -59,7 +59,7 @@ class RemoteLeaderEndPointTest { val time = new MockTime val logPrefix = "remote-leader-endpoint" val sourceBroker: BrokerEndPoint = new BrokerEndPoint(0, "localhost", 9092) - val props = TestUtils.createBrokerConfig(sourceBroker.id, TestUtils.MockZkConnect, port = sourceBroker.port) + val props = TestUtils.createBrokerConfig(sourceBroker.id, port = sourceBroker.port) val fetchSessionHandler = new FetchSessionHandler(new LogContext(logPrefix), sourceBroker.id) val config = KafkaConfig.fromProps(props) blockingSend = new MockBlockingSender(offsets = new util.HashMap[TopicPartition, EpochEndOffset](), diff --git a/core/src/test/scala/unit/kafka/coordinator/AbstractCoordinatorConcurrencyTest.scala b/core/src/test/scala/unit/kafka/coordinator/AbstractCoordinatorConcurrencyTest.scala index b567295809ff1..242a781c2113e 100644 --- a/core/src/test/scala/unit/kafka/coordinator/AbstractCoordinatorConcurrencyTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/AbstractCoordinatorConcurrencyTest.scala @@ -47,7 +47,7 @@ import scala.jdk.CollectionConverters._ abstract class AbstractCoordinatorConcurrencyTest[M <: CoordinatorMember] extends Logging { val nThreads = 5 - val serverProps = TestUtils.createBrokerConfig(nodeId = 0, zkConnect = "") + val serverProps = TestUtils.createBrokerConfig(0) val random = new Random var replicaManager: TestReplicaManager = _ var zkClient: KafkaZkClient = _ diff --git a/core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala b/core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala index 086f191de0b6d..025cfdac528a0 100644 --- a/core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala @@ -100,7 +100,7 @@ class GroupCoordinatorTest { @BeforeEach def setUp(): Unit = { - val props = TestUtils.createBrokerConfig(nodeId = 0, zkConnect = "") + val props = TestUtils.createBrokerConfig(0) props.setProperty(GroupCoordinatorConfig.GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG, GroupMinSessionTimeout.toString) props.setProperty(GroupCoordinatorConfig.GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG, GroupMaxSessionTimeout.toString) props.setProperty(GroupCoordinatorConfig.GROUP_MAX_SIZE_CONFIG, GroupMaxSize.toString) @@ -201,7 +201,7 @@ class GroupCoordinatorTest { @Test def testOffsetsRetentionMsIntegerOverflow(): Unit = { - val props = TestUtils.createBrokerConfig(nodeId = 0, zkConnect = "") + val props = TestUtils.createBrokerConfig(0) props.setProperty(GroupCoordinatorConfig.OFFSETS_RETENTION_MINUTES_CONFIG, Integer.MAX_VALUE.toString) val config = KafkaConfig.fromProps(props) val offsetConfig = GroupCoordinator.offsetConfig(config) diff --git a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala index 6033b312b6087..dbd0a9b05ceab 100644 --- a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala @@ -82,7 +82,7 @@ class GroupMetadataManagerTest { val noExpiration = OptionalLong.empty() private val offsetConfig = { - val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(nodeId = 0, zkConnect = "")) + val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(0)) new OffsetConfig(config.groupCoordinatorConfig.offsetMetadataMaxSize, config.groupCoordinatorConfig.offsetsLoadBufferSize, config.groupCoordinatorConfig.offsetsRetentionMs, diff --git a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerChannelManagerTest.scala b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerChannelManagerTest.scala index 852a076de5cf4..c7cf3d804e69c 100644 --- a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerChannelManagerTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerChannelManagerTest.scala @@ -74,7 +74,7 @@ class TransactionMarkerChannelManagerTest { private val time = new MockTime private val channelManager = new TransactionMarkerChannelManager( - KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:2181")), + KafkaConfig.fromProps(TestUtils.createBrokerConfig(1)), metadataCache, networkClient, txnStateManager, @@ -98,7 +98,7 @@ class TransactionMarkerChannelManagerTest { val mockMetricsGroupCtor = mockConstruction(classOf[KafkaMetricsGroup]) try { val transactionMarkerChannelManager = new TransactionMarkerChannelManager( - KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:2181")), + KafkaConfig.fromProps(TestUtils.createBrokerConfig(1)), metadataCache, networkClient, txnStateManager, diff --git a/core/src/test/scala/unit/kafka/integration/MetricsDuringTopicCreationDeletionTest.scala b/core/src/test/scala/unit/kafka/integration/MetricsDuringTopicCreationDeletionTest.scala index 10a32c4a3c1d8..ada5683405bbb 100644 --- a/core/src/test/scala/unit/kafka/integration/MetricsDuringTopicCreationDeletionTest.scala +++ b/core/src/test/scala/unit/kafka/integration/MetricsDuringTopicCreationDeletionTest.scala @@ -51,7 +51,7 @@ class MetricsDuringTopicCreationDeletionTest extends KafkaServerTestHarness with @volatile private var running = true - override def generateConfigs = TestUtils.createBrokerConfigs(nodesNum, null) + override def generateConfigs = TestUtils.createBrokerConfigs(nodesNum) .map(KafkaConfig.fromProps(_, overridingProps)) @BeforeEach diff --git a/core/src/test/scala/unit/kafka/integration/MinIsrConfigTest.scala b/core/src/test/scala/unit/kafka/integration/MinIsrConfigTest.scala index 4af4a49069f68..fb981369e6b66 100644 --- a/core/src/test/scala/unit/kafka/integration/MinIsrConfigTest.scala +++ b/core/src/test/scala/unit/kafka/integration/MinIsrConfigTest.scala @@ -29,7 +29,7 @@ import org.junit.jupiter.params.provider.ValueSource class MinIsrConfigTest extends KafkaServerTestHarness { val overridingProps = new Properties() overridingProps.put(ServerLogConfigs.MIN_IN_SYNC_REPLICAS_CONFIG, "5") - def generateConfigs: Seq[KafkaConfig] = TestUtils.createBrokerConfigs(1, null).map(KafkaConfig.fromProps(_, overridingProps)) + def generateConfigs: Seq[KafkaConfig] = TestUtils.createBrokerConfigs(1).map(KafkaConfig.fromProps(_, overridingProps)) @ParameterizedTest @ValueSource(strings = Array("kraft")) diff --git a/core/src/test/scala/unit/kafka/integration/UncleanLeaderElectionTest.scala b/core/src/test/scala/unit/kafka/integration/UncleanLeaderElectionTest.scala index 29714d1856b7c..f46bfd8386887 100755 --- a/core/src/test/scala/unit/kafka/integration/UncleanLeaderElectionTest.scala +++ b/core/src/test/scala/unit/kafka/integration/UncleanLeaderElectionTest.scala @@ -71,8 +71,8 @@ class UncleanLeaderElectionTest extends QuorumTestHarness { override def setUp(testInfo: TestInfo): Unit = { super.setUp(testInfo) - configProps1 = createBrokerConfig(brokerId1, null) - configProps2 = createBrokerConfig(brokerId2, null) + configProps1 = createBrokerConfig(brokerId1) + configProps2 = createBrokerConfig(brokerId2) for (configProps <- List(configProps1, configProps2)) { configProps.put("controlled.shutdown.enable", enableControlledShutdown.toString) diff --git a/core/src/test/scala/unit/kafka/log/LogCleanerParameterizedIntegrationTest.scala b/core/src/test/scala/unit/kafka/log/LogCleanerParameterizedIntegrationTest.scala index 0b53960dd04c4..d0a7624ed79bc 100755 --- a/core/src/test/scala/unit/kafka/log/LogCleanerParameterizedIntegrationTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogCleanerParameterizedIntegrationTest.scala @@ -158,7 +158,7 @@ class LogCleanerParameterizedIntegrationTest extends AbstractLogCleanerIntegrati assertTrue(cleaner.cleanerManager.allCleanerCheckpoints.isEmpty, "Should not have cleaned") def kafkaConfigWithCleanerConfig(cleanerConfig: CleanerConfig): KafkaConfig = { - val props = TestUtils.createBrokerConfig(0, "localhost:2181") + val props = TestUtils.createBrokerConfig(0) props.put(CleanerConfig.LOG_CLEANER_THREADS_PROP, cleanerConfig.numThreads.toString) props.put(CleanerConfig.LOG_CLEANER_DEDUPE_BUFFER_SIZE_PROP, cleanerConfig.dedupeBufferSize.toString) props.put(CleanerConfig.LOG_CLEANER_DEDUPE_BUFFER_LOAD_FACTOR_PROP, cleanerConfig.dedupeBufferLoadFactor.toString) diff --git a/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala b/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala index be9e11813018c..e4ebcb2d5da36 100644 --- a/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala @@ -133,8 +133,8 @@ class LogCleanerTest extends Logging { var nonexistent = LogCleaner.MetricNames.diff(KafkaYammerMetrics.defaultRegistry.allMetrics().keySet().asScala.map(_.getName)) assertEquals(0, nonexistent.size, s"$nonexistent should be existent") - logCleaner.reconfigure(new KafkaConfig(TestUtils.createBrokerConfig(1, "localhost:2181")), - new KafkaConfig(TestUtils.createBrokerConfig(1, "localhost:2181"))) + logCleaner.reconfigure(new KafkaConfig(TestUtils.createBrokerConfig(1)), + new KafkaConfig(TestUtils.createBrokerConfig(1))) nonexistent = LogCleaner.MetricNames.diff(KafkaYammerMetrics.defaultRegistry.allMetrics().keySet().asScala.map(_.getName)) assertEquals(0, nonexistent.size, s"$nonexistent should be existent") @@ -2017,7 +2017,7 @@ class LogCleanerTest extends Logging { @Test def testReconfigureLogCleanerIoMaxBytesPerSecond(): Unit = { - val oldKafkaProps = TestUtils.createBrokerConfig(1, "localhost:2181") + val oldKafkaProps = TestUtils.createBrokerConfig(1) oldKafkaProps.setProperty(CleanerConfig.LOG_CLEANER_IO_MAX_BYTES_PER_SECOND_PROP, "10000000") val logCleaner = new LogCleaner(LogCleaner.cleanerConfig(new KafkaConfig(oldKafkaProps)), @@ -2034,7 +2034,7 @@ class LogCleanerTest extends Logging { try { assertEquals(10000000, logCleaner.throttler.desiredRatePerSec, s"Throttler.desiredRatePerSec should be initialized from initial `${CleanerConfig.LOG_CLEANER_IO_MAX_BYTES_PER_SECOND_PROP}` config.") - val newKafkaProps = TestUtils.createBrokerConfig(1, "localhost:2181") + val newKafkaProps = TestUtils.createBrokerConfig(1) newKafkaProps.setProperty(CleanerConfig.LOG_CLEANER_IO_MAX_BYTES_PER_SECOND_PROP, "20000000") logCleaner.reconfigure(new KafkaConfig(oldKafkaProps), new KafkaConfig(newKafkaProps)) diff --git a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala index c1e97d74ef042..3a0a450f05a6c 100644 --- a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala @@ -62,7 +62,7 @@ class LogConfigTest { val millisInHour = 60L * 60L * 1000L val millisInDay = 24L * millisInHour val bytesInGB: Long = 1024 * 1024 * 1024 - val kafkaProps = TestUtils.createBrokerConfig(nodeId = 0, zkConnect = "") + val kafkaProps = TestUtils.createBrokerConfig(nodeId = 0) kafkaProps.put(ServerLogConfigs.LOG_ROLL_TIME_HOURS_CONFIG, "2") kafkaProps.put(ServerLogConfigs.LOG_ROLL_TIME_JITTER_HOURS_CONFIG, "2") kafkaProps.put(ServerLogConfigs.LOG_RETENTION_TIME_HOURS_CONFIG, "960") // 40 days @@ -181,7 +181,7 @@ class LogConfigTest { @Test def testOverriddenConfigsAsLoggableString(): Unit = { - val kafkaProps = TestUtils.createBrokerConfig(nodeId = 0, zkConnect = "") + val kafkaProps = TestUtils.createBrokerConfig(nodeId = 0) kafkaProps.put("unknown.broker.password.config", "aaaaa") kafkaProps.put(ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG, "50") kafkaProps.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, "somekeypassword") @@ -466,7 +466,7 @@ class LogConfigTest { def testValidateWithMetadataVersionJbodSupport(): Unit = { def validate(metadataVersion: MetadataVersion, jbodConfig: Boolean): Unit = KafkaConfig.fromProps( - TestUtils.createBrokerConfig(nodeId = 0, zkConnect = null, logDirCount = if (jbodConfig) 2 else 1) + TestUtils.createBrokerConfig(nodeId = 0, logDirCount = if (jbodConfig) 2 else 1) ).validateWithMetadataVersion(metadataVersion) validate(MetadataVersion.IBP_3_6_IV2, jbodConfig = false) diff --git a/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala b/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala index ec82099f730e9..d6324d95c3abf 100644 --- a/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala @@ -67,7 +67,7 @@ class LogLoaderTest { @BeforeEach def setUp(): Unit = { - val props = TestUtils.createBrokerConfig(0, "127.0.0.1:1", port = -1) + val props = TestUtils.createBrokerConfig(0, port = -1) config = KafkaConfig.fromProps(props) } diff --git a/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala b/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala index 399828ddedc43..43e01ac7e8944 100755 --- a/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala +++ b/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala @@ -74,7 +74,7 @@ class UnifiedLogTest { @BeforeEach def setUp(): Unit = { - val props = TestUtils.createBrokerConfig(0, "127.0.0.1:1", port = -1) + val props = TestUtils.createBrokerConfig(0, port = -1) config = KafkaConfig.fromProps(props) } diff --git a/core/src/test/scala/unit/kafka/metrics/MetricsTest.scala b/core/src/test/scala/unit/kafka/metrics/MetricsTest.scala index 451b8cf2f87e1..b1bc03b6ff479 100644 --- a/core/src/test/scala/unit/kafka/metrics/MetricsTest.scala +++ b/core/src/test/scala/unit/kafka/metrics/MetricsTest.scala @@ -51,7 +51,7 @@ class MetricsTest extends KafkaServerTestHarness with Logging { overridingProps.put(JmxReporter.EXCLUDE_CONFIG, s"$requiredKafkaServerPrefix=ClusterId") def generateConfigs: Seq[KafkaConfig] = - TestUtils.createBrokerConfigs(numNodes, null, enableControlledShutdown = false). + TestUtils.createBrokerConfigs(numNodes, enableControlledShutdown = false). map(KafkaConfig.fromProps(_, overridingProps)) val nMessages = 2 diff --git a/core/src/test/scala/unit/kafka/network/ConnectionQuotasTest.scala b/core/src/test/scala/unit/kafka/network/ConnectionQuotasTest.scala index 20a48bb5099d4..3906011a20380 100644 --- a/core/src/test/scala/unit/kafka/network/ConnectionQuotasTest.scala +++ b/core/src/test/scala/unit/kafka/network/ConnectionQuotasTest.scala @@ -66,7 +66,7 @@ class ConnectionQuotasTest { } def brokerPropsWithDefaultConnectionLimits: Properties = { - val props = TestUtils.createBrokerConfig(0, null, port = 0) + val props = TestUtils.createBrokerConfig(0, port = 0) props.put(SocketServerConfigs.LISTENERS_CONFIG, "EXTERNAL://localhost:0,REPLICATION://localhost:1,ADMIN://localhost:2") // ConnectionQuotas does not limit inter-broker listener even when broker-wide connection limit is reached props.put(ReplicationConfigs.INTER_BROKER_LISTENER_NAME_CONFIG, "REPLICATION") diff --git a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala index d1b5395591c4b..a690a70365e32 100644 --- a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala +++ b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala @@ -66,7 +66,7 @@ import scala.jdk.CollectionConverters._ import scala.util.control.ControlThrowable class SocketServerTest { - val props = TestUtils.createBrokerConfig(0, null, port = 0) + val props = TestUtils.createBrokerConfig(0, port = 0) props.put("listeners", "PLAINTEXT://localhost:0") props.put("num.network.threads", "1") props.put("socket.send.buffer.bytes", "300000") @@ -767,7 +767,7 @@ class SocketServerTest { @Test def testZeroMaxConnectionsPerIp(): Unit = { - val newProps = TestUtils.createBrokerConfig(0, null, port = 0) + val newProps = TestUtils.createBrokerConfig(0, port = 0) newProps.setProperty(SocketServerConfigs.MAX_CONNECTIONS_PER_IP_CONFIG, "0") newProps.setProperty(SocketServerConfigs.MAX_CONNECTIONS_PER_IP_OVERRIDES_CONFIG, "%s:%s".format("127.0.0.1", "5")) val server = new SocketServer(KafkaConfig.fromProps(newProps), new Metrics(), @@ -806,7 +806,7 @@ class SocketServerTest { @Test def testMaxConnectionsPerIpOverrides(): Unit = { val overrideNum = server.config.maxConnectionsPerIp + 1 - val overrideProps = TestUtils.createBrokerConfig(0, null, port = 0) + val overrideProps = TestUtils.createBrokerConfig(0, port = 0) overrideProps.put(SocketServerConfigs.MAX_CONNECTIONS_PER_IP_OVERRIDES_CONFIG, s"localhost:$overrideNum") val serverMetrics = new Metrics() val overrideServer = new SocketServer(KafkaConfig.fromProps(overrideProps), serverMetrics, @@ -865,7 +865,7 @@ class SocketServerTest { @Test def testConnectionRatePerIp(): Unit = { val defaultTimeoutMs = 2000 - val overrideProps = TestUtils.createBrokerConfig(0, null, port = 0) + val overrideProps = TestUtils.createBrokerConfig(0, port = 0) overrideProps.remove(SocketServerConfigs.MAX_CONNECTIONS_PER_IP_CONFIG) overrideProps.put(QuotaConfig.NUM_QUOTA_SAMPLES_CONFIG, String.valueOf(2)) val connectionRate = 5 @@ -916,7 +916,7 @@ class SocketServerTest { @Test def testThrottledSocketsClosedOnShutdown(): Unit = { - val overrideProps = TestUtils.createBrokerConfig(0, null, port = 0) + val overrideProps = TestUtils.createBrokerConfig(0, port = 0) overrideProps.remove("max.connections.per.ip") overrideProps.put(QuotaConfig.NUM_QUOTA_SAMPLES_CONFIG, String.valueOf(2)) val connectionRate = 5 @@ -1004,7 +1004,7 @@ class SocketServerTest { props.setProperty("connections.max.reauth.ms", reauthMs.toString) props.setProperty("listener.security.protocol.map", "SASL_PLAINTEXT:SASL_PLAINTEXT,CONTROLLER:PLAINTEXT") - val overrideProps = TestUtils.createBrokerConfig(0, null, saslProperties = Some(props), enableSaslPlaintext = true) + val overrideProps = TestUtils.createBrokerConfig(0, saslProperties = Some(props), enableSaslPlaintext = true) val time = new MockTime() val overrideServer = new TestableSocketServer(KafkaConfig.fromProps(overrideProps), time = time) try { @@ -1084,7 +1084,7 @@ class SocketServerTest { } private def checkClientDisconnectionUpdatesRequestMetrics(responseBufferSize: Int): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 0) + val props = TestUtils.createBrokerConfig(0, port = 0) val overrideServer = new TestableSocketServer(KafkaConfig.fromProps(props)) try { @@ -1117,7 +1117,7 @@ class SocketServerTest { def testServerShutdownWithoutEnable(): Unit = { // The harness server has already been enabled, so it's invalid for this test. shutdownServerAndMetrics(server) - val props = TestUtils.createBrokerConfig(0, null, port = 0) + val props = TestUtils.createBrokerConfig(0, port = 0) val overrideServer = new TestableSocketServer(KafkaConfig.fromProps(props)) overrideServer.shutdown() assertFalse(overrideServer.testableAcceptor.isOpen) @@ -1926,7 +1926,7 @@ class SocketServerTest { private def sslServerProps: Properties = { val trustStoreFile = TestUtils.tempFile("truststore", ".jks") - val sslProps = TestUtils.createBrokerConfig(0, null, interBrokerSecurityProtocol = Some(SecurityProtocol.SSL), + val sslProps = TestUtils.createBrokerConfig(0, interBrokerSecurityProtocol = Some(SecurityProtocol.SSL), trustStoreFile = Some(trustStoreFile)) sslProps.put(SocketServerConfigs.LISTENERS_CONFIG, "SSL://localhost:0") sslProps.put(SocketServerConfigs.ADVERTISED_LISTENERS_CONFIG, "SSL://localhost:0") diff --git a/core/src/test/scala/unit/kafka/security/authorizer/AuthorizerTest.scala b/core/src/test/scala/unit/kafka/security/authorizer/AuthorizerTest.scala index a614c92b780e4..833cae0672d67 100644 --- a/core/src/test/scala/unit/kafka/security/authorizer/AuthorizerTest.scala +++ b/core/src/test/scala/unit/kafka/security/authorizer/AuthorizerTest.scala @@ -89,7 +89,7 @@ class AuthorizerTest extends QuorumTestHarness with BaseAuthorizerTest { } def properties: Properties = { - val props = TestUtils.createBrokerConfig(0, null) + val props = TestUtils.createBrokerConfig(0) props.put(StandardAuthorizer.SUPER_USERS_CONFIG, superUsers) props } diff --git a/core/src/test/scala/unit/kafka/server/AddPartitionsToTxnManagerTest.scala b/core/src/test/scala/unit/kafka/server/AddPartitionsToTxnManagerTest.scala index b6507a2a0dba5..701ff8f079ce2 100644 --- a/core/src/test/scala/unit/kafka/server/AddPartitionsToTxnManagerTest.scala +++ b/core/src/test/scala/unit/kafka/server/AddPartitionsToTxnManagerTest.scala @@ -74,7 +74,7 @@ class AddPartitionsToTxnManagerTest { private val disconnectedResponse = clientResponse(null, disconnected = true) private val transactionSupportedOperation = genericErrorSupported - private val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:2181")) + private val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1)) @BeforeEach def setup(): Unit = { diff --git a/core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala b/core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala index 3a053d784c3e6..56886c5d742f8 100644 --- a/core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala +++ b/core/src/test/scala/unit/kafka/server/AutoTopicCreationManagerTest.scala @@ -67,7 +67,7 @@ class AutoTopicCreationManagerTest { @BeforeEach def setup(): Unit = { - val props = TestUtils.createBrokerConfig(1, "localhost") + val props = TestUtils.createBrokerConfig(1) props.setProperty(ServerConfigs.REQUEST_TIMEOUT_MS_CONFIG, requestTimeout.toString) props.setProperty(GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, internalTopicPartitions.toString) diff --git a/core/src/test/scala/unit/kafka/server/ConfigAdminManagerTest.scala b/core/src/test/scala/unit/kafka/server/ConfigAdminManagerTest.scala index 34753acabbdc3..a4494c5f1e776 100644 --- a/core/src/test/scala/unit/kafka/server/ConfigAdminManagerTest.scala +++ b/core/src/test/scala/unit/kafka/server/ConfigAdminManagerTest.scala @@ -47,7 +47,7 @@ class ConfigAdminManagerTest { val logger = LoggerFactory.getLogger(classOf[ConfigAdminManagerTest]) def newConfigAdminManager(brokerId: Integer): ConfigAdminManager = { - val config = TestUtils.createBrokerConfig(nodeId = brokerId, zkConnect = null) + val config = TestUtils.createBrokerConfig(nodeId = brokerId) new ConfigAdminManager(brokerId, new KafkaConfig(config), new MockConfigRepository()) } diff --git a/core/src/test/scala/unit/kafka/server/DeleteTopicsRequestWithDeletionDisabledTest.scala b/core/src/test/scala/unit/kafka/server/DeleteTopicsRequestWithDeletionDisabledTest.scala index 81875ceaac22b..4232030634cb8 100644 --- a/core/src/test/scala/unit/kafka/server/DeleteTopicsRequestWithDeletionDisabledTest.scala +++ b/core/src/test/scala/unit/kafka/server/DeleteTopicsRequestWithDeletionDisabledTest.scala @@ -40,7 +40,7 @@ class DeleteTopicsRequestWithDeletionDisabledTest extends BaseRequestTest { } override def generateConfigs = { - val props = TestUtils.createBrokerConfigs(brokerCount, null, + val props = TestUtils.createBrokerConfigs(brokerCount, enableControlledShutdown = false, enableDeleteTopic = false, interBrokerSecurityProtocol = Some(securityProtocol), trustStoreFile = trustStoreFile, saslProperties = serverSaslProperties, logDirCount = logDirCount) diff --git a/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala b/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala index 262b0abdc194f..09555351c6799 100755 --- a/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala +++ b/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala @@ -52,7 +52,7 @@ class DynamicBrokerConfigTest { @Test def testConfigUpdate(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) val oldKeystore = "oldKs.jks" props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, oldKeystore) val config = KafkaConfig(props) @@ -97,7 +97,7 @@ class DynamicBrokerConfigTest { @Test def testUpdateDynamicThreadPool(): Unit = { - val origProps = TestUtils.createBrokerConfig(0, null, port = 8181) + val origProps = TestUtils.createBrokerConfig(0, port = 8181) origProps.put(ServerConfigs.NUM_IO_THREADS_CONFIG, "4") origProps.put(SocketServerConfigs.NUM_NETWORK_THREADS_CONFIG, "2") origProps.put(ReplicationConfigs.NUM_REPLICA_FETCHERS_CONFIG, "1") @@ -168,7 +168,7 @@ class DynamicBrokerConfigTest { @Test def testUpdateRemoteLogManagerDynamicThreadPool(): Unit = { - val origProps = TestUtils.createBrokerConfig(0, null, port = 8181) + val origProps = TestUtils.createBrokerConfig(0, port = 8181) val config = KafkaConfig(origProps) assertEquals(RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE, config.remoteLogManagerConfig.remoteLogManagerCopierThreadPoolSize()) assertEquals(RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE, config.remoteLogManagerConfig.remoteLogManagerExpirationThreadPoolSize()) @@ -207,7 +207,7 @@ class DynamicBrokerConfigTest { @Test def testRemoteLogDynamicThreadPoolWithInvalidValues(): Unit = { - val origProps = TestUtils.createBrokerConfig(0, null, port = 8181) + val origProps = TestUtils.createBrokerConfig(0, port = 8181) val config = KafkaConfig(origProps) val serverMock = mock(classOf[KafkaBroker]) @@ -243,7 +243,7 @@ class DynamicBrokerConfigTest { @Test def testConfigUpdateWithSomeInvalidConfigs(): Unit = { - val origProps = TestUtils.createBrokerConfig(0, null, port = 8181) + val origProps = TestUtils.createBrokerConfig(0, port = 8181) origProps.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, "JKS") val config = KafkaConfig(origProps) config.dynamicConfig.initialize(None, None) @@ -262,7 +262,7 @@ class DynamicBrokerConfigTest { @Test def testConfigUpdateWithReconfigurableValidationFailure(): Unit = { - val origProps = TestUtils.createBrokerConfig(0, null, port = 8181) + val origProps = TestUtils.createBrokerConfig(0, port = 8181) origProps.put(CleanerConfig.LOG_CLEANER_DEDUPE_BUFFER_SIZE_PROP, "100000000") val config = KafkaConfig(origProps) config.dynamicConfig.initialize(None, None) @@ -296,7 +296,7 @@ class DynamicBrokerConfigTest { @Test def testReconfigurableValidation(): Unit = { - val origProps = TestUtils.createBrokerConfig(0, null, port = 8181) + val origProps = TestUtils.createBrokerConfig(0, port = 8181) val config = KafkaConfig(origProps) val invalidReconfigurableProps = Set(CleanerConfig.LOG_CLEANER_THREADS_PROP, ServerConfigs.BROKER_ID_CONFIG, "some.prop") val validReconfigurableProps = Set(CleanerConfig.LOG_CLEANER_THREADS_PROP, CleanerConfig.LOG_CLEANER_DEDUPE_BUFFER_SIZE_PROP, "some.prop") @@ -366,7 +366,7 @@ class DynamicBrokerConfigTest { } private def verifyConfigUpdate(name: String, value: Object, perBrokerConfig: Boolean, expectFailure: Boolean): Unit = { - val configProps = TestUtils.createBrokerConfig(0, null, port = 8181) + val configProps = TestUtils.createBrokerConfig(0, port = 8181) val config = KafkaConfig(configProps) config.dynamicConfig.initialize(None, None) @@ -415,7 +415,7 @@ class DynamicBrokerConfigTest { @Test def testDynamicListenerConfig(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 9092) + val props = TestUtils.createBrokerConfig(0, port = 9092) val oldConfig = KafkaConfig.fromProps(props) val kafkaServer: KafkaBroker = mock(classOf[kafka.server.KafkaBroker]) when(kafkaServer.config).thenReturn(oldConfig) @@ -457,7 +457,7 @@ class DynamicBrokerConfigTest { @Test def testAuthorizerConfig(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 9092) + val props = TestUtils.createBrokerConfig(0, port = 9092) val oldConfig = KafkaConfig.fromProps(props) oldConfig.dynamicConfig.initialize(None, None) @@ -491,7 +491,6 @@ class DynamicBrokerConfigTest { port: Int ): Properties = { val retval = TestUtils.createBrokerConfig(nodeId, - zkConnect = null, enableControlledShutdown = true, enableDeleteTopic = true, port) @@ -534,7 +533,6 @@ class DynamicBrokerConfigTest { port: Int ): Properties = { val retval = TestUtils.createBrokerConfig(nodeId, - zkConnect = null, enableControlledShutdown = true, enableDeleteTopic = true, port @@ -589,7 +587,7 @@ class DynamicBrokerConfigTest { @Test def testImproperConfigsAreRemoved(): Unit = { - val props = TestUtils.createBrokerConfig(0, null) + val props = TestUtils.createBrokerConfig(0) val config = KafkaConfig(props) config.dynamicConfig.initialize(None, None) @@ -618,7 +616,7 @@ class DynamicBrokerConfigTest { @Test def testUpdateMetricReporters(): Unit = { val brokerId = 0 - val origProps = TestUtils.createBrokerConfig(brokerId, null, port = 8181) + val origProps = TestUtils.createBrokerConfig(brokerId, port = 8181) val config = KafkaConfig(origProps) val serverMock = Mockito.mock(classOf[KafkaBroker]) @@ -642,7 +640,7 @@ class DynamicBrokerConfigTest { @Test def testUpdateMetricReportersNoJmxReporter(): Unit = { val brokerId = 0 - val origProps = TestUtils.createBrokerConfig(brokerId, null, port = 8181) + val origProps = TestUtils.createBrokerConfig(brokerId, port = 8181) origProps.put(MetricConfigs.METRIC_REPORTER_CLASSES_CONFIG, "") val config = KafkaConfig(origProps) @@ -669,7 +667,7 @@ class DynamicBrokerConfigTest { @Test def testNonInternalValuesDoesNotExposeInternalConfigs(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.put(KRaftConfigs.METADATA_LOG_SEGMENT_MIN_BYTES_CONFIG, "1024") val config = new KafkaConfig(props) assertFalse(config.nonInternalValues.containsKey(KRaftConfigs.METADATA_LOG_SEGMENT_MIN_BYTES_CONFIG)) @@ -679,7 +677,7 @@ class DynamicBrokerConfigTest { @Test def testDynamicLogLocalRetentionMsConfig(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.put(ServerLogConfigs.LOG_RETENTION_TIME_MILLIS_CONFIG, "2592000000") val config = KafkaConfig(props) val dynamicLogConfig = new DynamicLogConfig(mock(classOf[LogManager]), mock(classOf[KafkaBroker])) @@ -702,7 +700,7 @@ class DynamicBrokerConfigTest { @Test def testDynamicLogLocalRetentionSizeConfig(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.put(ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG, "4294967296") val config = KafkaConfig(props) val dynamicLogConfig = new DynamicLogConfig(mock(classOf[LogManager]), mock(classOf[KafkaBroker])) @@ -725,7 +723,7 @@ class DynamicBrokerConfigTest { @Test def testDynamicLogLocalRetentionSkipsOnInvalidConfig(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP, "1000") props.put(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP, "1024") val config = KafkaConfig(props) @@ -751,7 +749,7 @@ class DynamicBrokerConfigTest { @Test def testDynamicRemoteFetchMaxWaitMsConfig(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) val config = KafkaConfig(props) val kafkaBroker = mock(classOf[KafkaBroker]) when(kafkaBroker.config).thenReturn(config) @@ -785,7 +783,7 @@ class DynamicBrokerConfigTest { @Test def testDynamicRemoteListOffsetsRequestTimeoutMsConfig(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) val config = KafkaConfig(props) val kafkaBroker = mock(classOf[KafkaBroker]) when(kafkaBroker.config).thenReturn(config) @@ -820,7 +818,7 @@ class DynamicBrokerConfigTest { @Test def testUpdateDynamicRemoteLogManagerConfig(): Unit = { - val origProps = TestUtils.createBrokerConfig(0, null, port = 8181) + val origProps = TestUtils.createBrokerConfig(0, port = 8181) origProps.put(RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP, "2") val config = KafkaConfig(origProps) @@ -845,7 +843,7 @@ class DynamicBrokerConfigTest { @Test def testRemoteLogManagerCopyQuotaUpdates(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 9092) + val props = TestUtils.createBrokerConfig(0, port = 9092) val config = KafkaConfig.fromProps(props) val serverMock: KafkaBroker = mock(classOf[KafkaBroker]) val remoteLogManager = mock(classOf[RemoteLogManager]) @@ -876,7 +874,7 @@ class DynamicBrokerConfigTest { @Test def testRemoteLogManagerFetchQuotaUpdates(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 9092) + val props = TestUtils.createBrokerConfig(0, port = 9092) val config = KafkaConfig.fromProps(props) val serverMock: KafkaBroker = mock(classOf[KafkaBroker]) val remoteLogManager = mock(classOf[RemoteLogManager]) @@ -911,7 +909,7 @@ class DynamicBrokerConfigTest { val copyQuotaProp = RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP val fetchQuotaProp = RemoteLogManagerConfig.REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP - val props = TestUtils.createBrokerConfig(0, null, port = 9092) + val props = TestUtils.createBrokerConfig(0, port = 9092) val config = KafkaConfig.fromProps(props) val serverMock: KafkaBroker = mock(classOf[KafkaBroker]) val remoteLogManager = Mockito.mock(classOf[RemoteLogManager]) @@ -961,7 +959,7 @@ class DynamicBrokerConfigTest { retentionMs: Long, logLocalRetentionBytes: Long, retentionBytes: Long): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.put(ServerLogConfigs.LOG_RETENTION_TIME_MILLIS_CONFIG, retentionMs.toString) props.put(ServerLogConfigs.LOG_RETENTION_BYTES_CONFIG, retentionBytes.toString) val config = KafkaConfig(props) @@ -998,7 +996,7 @@ class DynamicBrokerConfigTest { @Test def testDynamicLogConfigHandlesSynonymsCorrectly(): Unit = { - val origProps = TestUtils.createBrokerConfig(0, null, port = 8181) + val origProps = TestUtils.createBrokerConfig(0, port = 8181) origProps.put(ServerLogConfigs.LOG_RETENTION_TIME_MINUTES_CONFIG, "1") val ctx = new DynamicLogConfigContext(origProps) assertEquals(TimeUnit.MINUTES.toMillis(1), ctx.config.logRetentionTimeMillis) @@ -1011,7 +1009,7 @@ class DynamicBrokerConfigTest { @Test def testLogRetentionTimeMinutesIsNotDynamicallyReconfigurable(): Unit = { - val origProps = TestUtils.createBrokerConfig(0, null, port = 8181) + val origProps = TestUtils.createBrokerConfig(0, port = 8181) origProps.put(ServerLogConfigs.LOG_RETENTION_TIME_HOURS_CONFIG, "1") val ctx = new DynamicLogConfigContext(origProps) assertEquals(TimeUnit.HOURS.toMillis(1), ctx.config.logRetentionTimeMillis) diff --git a/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala b/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala index eaca2cc2c955d..fcb2ffcbb7d54 100644 --- a/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala +++ b/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala @@ -56,7 +56,7 @@ import scala.jdk.CollectionConverters._ @Timeout(100) class DynamicConfigChangeTest extends KafkaServerTestHarness { override def generateConfigs: Seq[KafkaConfig] = { - val cfg = TestUtils.createBrokerConfig(0, null) + val cfg = TestUtils.createBrokerConfig(0) List(KafkaConfig.fromProps(cfg)) } diff --git a/core/src/test/scala/unit/kafka/server/EdgeCaseRequestTest.scala b/core/src/test/scala/unit/kafka/server/EdgeCaseRequestTest.scala index bf1a934fe02f3..85ae9121843a4 100755 --- a/core/src/test/scala/unit/kafka/server/EdgeCaseRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/EdgeCaseRequestTest.scala @@ -44,7 +44,7 @@ import scala.jdk.CollectionConverters._ class EdgeCaseRequestTest extends KafkaServerTestHarness { def generateConfigs = { - val props = TestUtils.createBrokerConfig(1, null) + val props = TestUtils.createBrokerConfig(1) props.setProperty(ServerLogConfigs.AUTO_CREATE_TOPICS_ENABLE_CONFIG, "false") List(KafkaConfig.fromProps(props)) } diff --git a/core/src/test/scala/unit/kafka/server/HighwatermarkPersistenceTest.scala b/core/src/test/scala/unit/kafka/server/HighwatermarkPersistenceTest.scala index a5f294f378aa0..579487bc47786 100755 --- a/core/src/test/scala/unit/kafka/server/HighwatermarkPersistenceTest.scala +++ b/core/src/test/scala/unit/kafka/server/HighwatermarkPersistenceTest.scala @@ -32,7 +32,7 @@ import org.apache.kafka.storage.internals.log.{CleanerConfig, LogDirFailureChann class HighwatermarkPersistenceTest { - val configs = TestUtils.createBrokerConfigs(2, null).map(KafkaConfig.fromProps) + val configs = TestUtils.createBrokerConfigs(2).map(KafkaConfig.fromProps) val topic = "foo" val configRepository = new MockConfigRepository() val logManagers = configs map { config => diff --git a/core/src/test/scala/unit/kafka/server/IsrExpirationTest.scala b/core/src/test/scala/unit/kafka/server/IsrExpirationTest.scala index f82e14c6e0f22..429b6869013f9 100644 --- a/core/src/test/scala/unit/kafka/server/IsrExpirationTest.scala +++ b/core/src/test/scala/unit/kafka/server/IsrExpirationTest.scala @@ -47,7 +47,7 @@ class IsrExpirationTest { val overridingProps = new Properties() overridingProps.put(ReplicationConfigs.REPLICA_LAG_TIME_MAX_MS_CONFIG, replicaLagTimeMaxMs.toString) overridingProps.put(ReplicationConfigs.REPLICA_FETCH_WAIT_MAX_MS_CONFIG, replicaFetchWaitMaxMs.toString) - val configs = TestUtils.createBrokerConfigs(2, null).map(KafkaConfig.fromProps(_, overridingProps)) + val configs = TestUtils.createBrokerConfigs(2).map(KafkaConfig.fromProps(_, overridingProps)) val topic = "foo" val time = new MockTime diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala index 02e793b4dc051..229b1a18a19a7 100644 --- a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala +++ b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala @@ -85,7 +85,6 @@ import org.apache.kafka.raft.QuorumConfig import org.apache.kafka.security.authorizer.AclEntry import org.apache.kafka.server.{BrokerFeatures, ClientMetricsManager} import org.apache.kafka.server.authorizer.{Action, AuthorizationResult, Authorizer} -import org.apache.kafka.server.common.MetadataVersion.{IBP_0_10_2_IV0, IBP_2_2_IV1} import org.apache.kafka.server.common.{FeatureVersion, FinalizedFeatures, GroupVersion, KRaftVersion, MetadataVersion, RequestLocal, TransactionVersion} import org.apache.kafka.server.config.{ConfigType, KRaftConfigs, ReplicationConfigs, ServerConfigs, ServerLogConfigs} import org.apache.kafka.server.metrics.ClientMetricsTestUtils @@ -139,7 +138,7 @@ class KafkaApisTest extends Logging { private val brokerId = 1 // KRaft tests should override this with a KRaftMetadataCache private var metadataCache: MetadataCache = MetadataCache.zkMetadataCache(brokerId, MetadataVersion.latestTesting()) - private var brokerEpochManager: ZkBrokerEpochManager = new ZkBrokerEpochManager(metadataCache, controller, None) + private val brokerEpochManager: ZkBrokerEpochManager = new ZkBrokerEpochManager(metadataCache, controller, None) private val clientQuotaManager: ClientQuotaManager = mock(classOf[ClientQuotaManager]) private val clientRequestQuotaManager: ClientRequestQuotaManager = mock(classOf[ClientRequestQuotaManager]) private val clientControllerQuotaManager: ControllerMutationQuotaManager = mock(classOf[ControllerMutationQuotaManager]) @@ -173,14 +172,14 @@ class KafkaApisTest extends Logging { overrideProperties: Map[String, String] = Map.empty, featureVersions: Seq[FeatureVersion] = Seq.empty): KafkaApis = { val properties = if (raftSupport) { - val properties = TestUtils.createBrokerConfig(brokerId, null) + val properties = TestUtils.createBrokerConfig(brokerId) properties.put(KRaftConfigs.NODE_ID_CONFIG, brokerId.toString) properties.put(KRaftConfigs.PROCESS_ROLES_CONFIG, "broker") val voterId = brokerId + 1 properties.put(QuorumConfig.QUORUM_VOTERS_CONFIG, s"$voterId@localhost:9093") properties } else { - TestUtils.createBrokerConfig(brokerId, "zk") + TestUtils.createBrokerConfig(brokerId) } overrideProperties.foreach( p => properties.put(p._1, p._2)) TestUtils.setIbpVersion(properties, interBrokerProtocolVersion) @@ -2952,52 +2951,7 @@ class KafkaApisTest extends Logging { checkInvalidPartition(-1) checkInvalidPartition(1) // topic has only one partition } - - @Test - def shouldThrowUnsupportedVersionExceptionOnHandleAddOffsetToTxnRequestWhenInterBrokerProtocolNotSupported(): Unit = { - metadataCache = MetadataCache.zkMetadataCache(brokerId, IBP_0_10_2_IV0) - brokerEpochManager = new ZkBrokerEpochManager(metadataCache, controller, None) - kafkaApis = createKafkaApis(IBP_0_10_2_IV0) - assertThrows(classOf[UnsupportedVersionException], - () => kafkaApis.handleAddOffsetsToTxnRequest(null, RequestLocal.withThreadConfinedCaching)) - } - - @Test - def shouldThrowUnsupportedVersionExceptionOnHandleAddPartitionsToTxnRequestWhenInterBrokerProtocolNotSupported(): Unit = { - metadataCache = MetadataCache.zkMetadataCache(brokerId, IBP_0_10_2_IV0) - brokerEpochManager = new ZkBrokerEpochManager(metadataCache, controller, None) - kafkaApis = createKafkaApis(IBP_0_10_2_IV0) - assertThrows(classOf[UnsupportedVersionException], - () => kafkaApis.handleAddPartitionsToTxnRequest(null, RequestLocal.withThreadConfinedCaching)) - } - - @Test - def shouldThrowUnsupportedVersionExceptionOnHandleTxnOffsetCommitRequestWhenInterBrokerProtocolNotSupported(): Unit = { - metadataCache = MetadataCache.zkMetadataCache(brokerId, IBP_0_10_2_IV0) - brokerEpochManager = new ZkBrokerEpochManager(metadataCache, controller, None) - kafkaApis = createKafkaApis(IBP_0_10_2_IV0) - assertThrows(classOf[UnsupportedVersionException], - () => kafkaApis.handleAddPartitionsToTxnRequest(null, RequestLocal.withThreadConfinedCaching)) - } - - @Test - def shouldThrowUnsupportedVersionExceptionOnHandleEndTxnRequestWhenInterBrokerProtocolNotSupported(): Unit = { - metadataCache = MetadataCache.zkMetadataCache(brokerId, IBP_0_10_2_IV0) - brokerEpochManager = new ZkBrokerEpochManager(metadataCache, controller, None) - kafkaApis = createKafkaApis(IBP_0_10_2_IV0) - assertThrows(classOf[UnsupportedVersionException], - () => kafkaApis.handleEndTxnRequest(null, RequestLocal.withThreadConfinedCaching)) - } - - @Test - def shouldThrowUnsupportedVersionExceptionOnHandleWriteTxnMarkersRequestWhenInterBrokerProtocolNotSupported(): Unit = { - metadataCache = MetadataCache.zkMetadataCache(brokerId, IBP_0_10_2_IV0) - brokerEpochManager = new ZkBrokerEpochManager(metadataCache, controller, None) - kafkaApis = createKafkaApis(IBP_0_10_2_IV0) - assertThrows(classOf[UnsupportedVersionException], - () => kafkaApis.handleWriteTxnMarkersRequest(null, RequestLocal.withThreadConfinedCaching)) - } - + @Test def requiredAclsNotPresentWriteTxnMarkersThrowsAuthorizationException(): Unit = { val topicPartition = new TopicPartition("t", 0) @@ -8882,107 +8836,6 @@ class KafkaApisTest extends Logging { assertEquals(Errors.GROUP_AUTHORIZATION_FAILED, response.error) } - @Test - def rejectJoinGroupRequestWhenStaticMembershipNotSupported(): Unit = { - val joinGroupRequest = new JoinGroupRequest.Builder( - new JoinGroupRequestData() - .setGroupId("test") - .setMemberId("test") - .setGroupInstanceId("instanceId") - .setProtocolType("consumer") - .setProtocols(new JoinGroupRequestData.JoinGroupRequestProtocolCollection) - ).build() - - val requestChannelRequest = buildRequest(joinGroupRequest) - metadataCache = MetadataCache.zkMetadataCache(brokerId, IBP_2_2_IV1) - brokerEpochManager = new ZkBrokerEpochManager(metadataCache, controller, None) - kafkaApis = createKafkaApis(IBP_2_2_IV1) - kafkaApis.handleJoinGroupRequest(requestChannelRequest, RequestLocal.withThreadConfinedCaching) - - val response = verifyNoThrottling[JoinGroupResponse](requestChannelRequest) - assertEquals(Errors.UNSUPPORTED_VERSION, response.error()) - } - - @Test - def rejectSyncGroupRequestWhenStaticMembershipNotSupported(): Unit = { - val syncGroupRequest = new SyncGroupRequest.Builder( - new SyncGroupRequestData() - .setGroupId("test") - .setMemberId("test") - .setGroupInstanceId("instanceId") - .setGenerationId(1) - ).build() - - val requestChannelRequest = buildRequest(syncGroupRequest) - metadataCache = MetadataCache.zkMetadataCache(brokerId, IBP_2_2_IV1) - brokerEpochManager = new ZkBrokerEpochManager(metadataCache, controller, None) - kafkaApis = createKafkaApis(IBP_2_2_IV1) - kafkaApis.handleSyncGroupRequest(requestChannelRequest, RequestLocal.withThreadConfinedCaching) - - val response = verifyNoThrottling[SyncGroupResponse](requestChannelRequest) - assertEquals(Errors.UNSUPPORTED_VERSION, response.error) - } - - @Test - def rejectHeartbeatRequestWhenStaticMembershipNotSupported(): Unit = { - val heartbeatRequest = new HeartbeatRequest.Builder( - new HeartbeatRequestData() - .setGroupId("test") - .setMemberId("test") - .setGroupInstanceId("instanceId") - .setGenerationId(1) - ).build() - val requestChannelRequest = buildRequest(heartbeatRequest) - metadataCache = MetadataCache.zkMetadataCache(brokerId, IBP_2_2_IV1) - brokerEpochManager = new ZkBrokerEpochManager(metadataCache, controller, None) - kafkaApis = createKafkaApis(IBP_2_2_IV1) - kafkaApis.handleHeartbeatRequest(requestChannelRequest) - - val response = verifyNoThrottling[HeartbeatResponse](requestChannelRequest) - assertEquals(Errors.UNSUPPORTED_VERSION, response.error()) - } - - @Test - def rejectOffsetCommitRequestWhenStaticMembershipNotSupported(): Unit = { - val offsetCommitRequest = new OffsetCommitRequest.Builder( - new OffsetCommitRequestData() - .setGroupId("test") - .setMemberId("test") - .setGroupInstanceId("instanceId") - .setGenerationIdOrMemberEpoch(100) - .setTopics(Collections.singletonList( - new OffsetCommitRequestData.OffsetCommitRequestTopic() - .setName("test") - .setPartitions(Collections.singletonList( - new OffsetCommitRequestData.OffsetCommitRequestPartition() - .setPartitionIndex(0) - .setCommittedOffset(100) - .setCommittedLeaderEpoch(RecordBatch.NO_PARTITION_LEADER_EPOCH) - .setCommittedMetadata("") - )) - )) - ).build() - - val requestChannelRequest = buildRequest(offsetCommitRequest) - - metadataCache = MetadataCache.zkMetadataCache(brokerId, IBP_2_2_IV1) - brokerEpochManager = new ZkBrokerEpochManager(metadataCache, controller, None) - kafkaApis = createKafkaApis(IBP_2_2_IV1) - kafkaApis.handleOffsetCommitRequest(requestChannelRequest, RequestLocal.withThreadConfinedCaching) - - val expectedTopicErrors = Collections.singletonList( - new OffsetCommitResponseData.OffsetCommitResponseTopic() - .setName("test") - .setPartitions(Collections.singletonList( - new OffsetCommitResponseData.OffsetCommitResponsePartition() - .setPartitionIndex(0) - .setErrorCode(Errors.UNSUPPORTED_VERSION.code) - )) - ) - val response = verifyNoThrottling[OffsetCommitResponse](requestChannelRequest) - assertEquals(expectedTopicErrors, response.data.topics()) - } - @ParameterizedTest @ApiKeyVersionsSource(apiKey = ApiKeys.LEAVE_GROUP) def testHandleLeaveGroupWithMultipleMembers(version: Short): Unit = { @@ -9737,41 +9590,6 @@ class KafkaApisTest extends Logging { assertEquals(records.sizeInBytes(), brokerTopicStats.allTopicsStats.replicationBytesOutRate.get.count()) } - @Test - def rejectInitProducerIdWhenIdButNotEpochProvided(): Unit = { - val initProducerIdRequest = new InitProducerIdRequest.Builder( - new InitProducerIdRequestData() - .setTransactionalId("known") - .setTransactionTimeoutMs(TimeUnit.MINUTES.toMillis(15).toInt) - .setProducerId(10) - .setProducerEpoch(RecordBatch.NO_PRODUCER_EPOCH) - ).build() - - val requestChannelRequest = buildRequest(initProducerIdRequest) - kafkaApis = createKafkaApis(IBP_2_2_IV1) - kafkaApis.handleInitProducerIdRequest(requestChannelRequest, RequestLocal.withThreadConfinedCaching) - - val response = verifyNoThrottling[InitProducerIdResponse](requestChannelRequest) - assertEquals(Errors.INVALID_REQUEST, response.error) - } - - @Test - def rejectInitProducerIdWhenEpochButNotIdProvided(): Unit = { - val initProducerIdRequest = new InitProducerIdRequest.Builder( - new InitProducerIdRequestData() - .setTransactionalId("known") - .setTransactionTimeoutMs(TimeUnit.MINUTES.toMillis(15).toInt) - .setProducerId(RecordBatch.NO_PRODUCER_ID) - .setProducerEpoch(2) - ).build() - val requestChannelRequest = buildRequest(initProducerIdRequest) - kafkaApis = createKafkaApis(IBP_2_2_IV1) - kafkaApis.handleInitProducerIdRequest(requestChannelRequest, RequestLocal.withThreadConfinedCaching) - - val response = verifyNoThrottling[InitProducerIdResponse](requestChannelRequest) - assertEquals(Errors.INVALID_REQUEST, response.error) - } - @Test def testUpdateMetadataRequestWithCurrentBrokerEpoch(): Unit = { val currentBrokerEpoch = 1239875L diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala index 32810e078ec50..c4544544bc2d3 100755 --- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala +++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala @@ -53,7 +53,7 @@ class KafkaConfigTest { @Test def testLogRetentionTimeHoursProvided(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.setProperty(ServerLogConfigs.LOG_RETENTION_TIME_HOURS_CONFIG, "1") val cfg = KafkaConfig.fromProps(props) @@ -62,7 +62,7 @@ class KafkaConfigTest { @Test def testLogRetentionTimeMinutesProvided(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.setProperty(ServerLogConfigs.LOG_RETENTION_TIME_MINUTES_CONFIG, "30") val cfg = KafkaConfig.fromProps(props) @@ -71,7 +71,7 @@ class KafkaConfigTest { @Test def testLogRetentionTimeMsProvided(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.setProperty(ServerLogConfigs.LOG_RETENTION_TIME_MILLIS_CONFIG, "1800000") val cfg = KafkaConfig.fromProps(props) @@ -80,7 +80,7 @@ class KafkaConfigTest { @Test def testLogRetentionTimeNoConfigProvided(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) val cfg = KafkaConfig.fromProps(props) assertEquals(24 * 7 * 60L * 60L * 1000L, cfg.logRetentionTimeMillis) @@ -88,7 +88,7 @@ class KafkaConfigTest { @Test def testLogRetentionTimeBothMinutesAndHoursProvided(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.setProperty(ServerLogConfigs.LOG_RETENTION_TIME_MINUTES_CONFIG, "30") props.setProperty(ServerLogConfigs.LOG_RETENTION_TIME_HOURS_CONFIG, "1") @@ -98,7 +98,7 @@ class KafkaConfigTest { @Test def testLogRetentionTimeBothMinutesAndMsProvided(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.setProperty(ServerLogConfigs.LOG_RETENTION_TIME_MILLIS_CONFIG, "1800000") props.setProperty(ServerLogConfigs.LOG_RETENTION_TIME_MINUTES_CONFIG, "10") @@ -108,11 +108,11 @@ class KafkaConfigTest { @Test def testLogRetentionUnlimited(): Unit = { - val props1 = TestUtils.createBrokerConfig(0, null, port = 8181) - val props2 = TestUtils.createBrokerConfig(0, null, port = 8181) - val props3 = TestUtils.createBrokerConfig(0, null, port = 8181) - val props4 = TestUtils.createBrokerConfig(0, null, port = 8181) - val props5 = TestUtils.createBrokerConfig(0, null, port = 8181) + val props1 = TestUtils.createBrokerConfig(0, port = 8181) + val props2 = TestUtils.createBrokerConfig(0, port = 8181) + val props3 = TestUtils.createBrokerConfig(0, port = 8181) + val props4 = TestUtils.createBrokerConfig(0, port = 8181) + val props5 = TestUtils.createBrokerConfig(0, port = 8181) props1.setProperty("log.retention.ms", "-1") props2.setProperty("log.retention.minutes", "-1") @@ -138,9 +138,9 @@ class KafkaConfigTest { @Test def testLogRetentionValid(): Unit = { - val props1 = TestUtils.createBrokerConfig(0, null, port = 8181) - val props2 = TestUtils.createBrokerConfig(0, null, port = 8181) - val props3 = TestUtils.createBrokerConfig(0, null, port = 8181) + val props1 = TestUtils.createBrokerConfig(0, port = 8181) + val props2 = TestUtils.createBrokerConfig(0, port = 8181) + val props3 = TestUtils.createBrokerConfig(0, port = 8181) props1.setProperty("log.retention.ms", "0") props2.setProperty("log.retention.minutes", "0") @@ -174,7 +174,7 @@ class KafkaConfigTest { val advertisedHostName = "routable-host" val advertisedPort = 1234 - val props = TestUtils.createBrokerConfig(0, null) + val props = TestUtils.createBrokerConfig(0) props.setProperty(SocketServerConfigs.ADVERTISED_LISTENERS_CONFIG, s"PLAINTEXT://$advertisedHostName:$advertisedPort") val serverConfig = KafkaConfig.fromProps(props) @@ -654,7 +654,7 @@ class KafkaConfigTest { @Test def testUncleanLeaderElectionDefault(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) val serverConfig = KafkaConfig.fromProps(props) assertEquals(serverConfig.uncleanLeaderElectionEnable, false) @@ -662,7 +662,7 @@ class KafkaConfigTest { @Test def testUncleanElectionDisabled(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.setProperty(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, String.valueOf(false)) val serverConfig = KafkaConfig.fromProps(props) @@ -671,7 +671,7 @@ class KafkaConfigTest { @Test def testUncleanElectionEnabled(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.setProperty(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, String.valueOf(true)) val serverConfig = KafkaConfig.fromProps(props) @@ -680,7 +680,7 @@ class KafkaConfigTest { @Test def testUncleanElectionInvalid(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.setProperty(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, "invalid") assertThrows(classOf[ConfigException], () => KafkaConfig.fromProps(props)) @@ -688,7 +688,7 @@ class KafkaConfigTest { @Test def testLogRollTimeMsProvided(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.setProperty(ServerLogConfigs.LOG_ROLL_TIME_MILLIS_CONFIG, "1800000") val cfg = KafkaConfig.fromProps(props) @@ -697,7 +697,7 @@ class KafkaConfigTest { @Test def testLogRollTimeBothMsAndHoursProvided(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.setProperty(ServerLogConfigs.LOG_ROLL_TIME_MILLIS_CONFIG, "1800000") props.setProperty(ServerLogConfigs.LOG_ROLL_TIME_HOURS_CONFIG, "1") @@ -707,7 +707,7 @@ class KafkaConfigTest { @Test def testLogRollTimeNoConfigProvided(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) val cfg = KafkaConfig.fromProps(props) assertEquals(24 * 7 * 60L * 60L * 1000L, cfg.logRollTimeMillis ) @@ -715,14 +715,14 @@ class KafkaConfigTest { @Test def testDefaultCompressionType(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) val serverConfig = KafkaConfig.fromProps(props) assertEquals(serverConfig.compressionType, "producer") } @Test def testValidCompressionType(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.setProperty("compression.type", "gzip") val serverConfig = KafkaConfig.fromProps(props) assertEquals(serverConfig.compressionType, "gzip") @@ -730,14 +730,14 @@ class KafkaConfigTest { @Test def testInvalidCompressionType(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.setProperty(ServerConfigs.COMPRESSION_TYPE_CONFIG, "abc") assertThrows(classOf[ConfigException], () => KafkaConfig.fromProps(props)) } @Test def testInvalidGzipCompressionLevel(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.setProperty(ServerConfigs.COMPRESSION_TYPE_CONFIG, "gzip") props.setProperty(ServerConfigs.COMPRESSION_GZIP_LEVEL_CONFIG, (CompressionType.GZIP.maxLevel() + 1).toString) assertThrows(classOf[ConfigException], () => KafkaConfig.fromProps(props)) @@ -745,7 +745,7 @@ class KafkaConfigTest { @Test def testInvalidLz4CompressionLevel(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.setProperty(ServerConfigs.COMPRESSION_TYPE_CONFIG, "lz4") props.setProperty(ServerConfigs.COMPRESSION_LZ4_LEVEL_CONFIG, (CompressionType.LZ4.maxLevel() + 1).toString) assertThrows(classOf[ConfigException], () => KafkaConfig.fromProps(props)) @@ -753,7 +753,7 @@ class KafkaConfigTest { @Test def testInvalidZstdCompressionLevel(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.setProperty(ServerConfigs.COMPRESSION_TYPE_CONFIG, "zstd") props.setProperty(ServerConfigs.COMPRESSION_ZSTD_LEVEL_CONFIG, (CompressionType.ZSTD.maxLevel() + 1).toString) assertThrows(classOf[ConfigException], () => KafkaConfig.fromProps(props)) @@ -761,7 +761,7 @@ class KafkaConfigTest { @Test def testInvalidInterBrokerSecurityProtocol(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.setProperty(SocketServerConfigs.LISTENERS_CONFIG, "SSL://localhost:0") props.setProperty(ReplicationConfigs.INTER_BROKER_SECURITY_PROTOCOL_CONFIG, SecurityProtocol.PLAINTEXT.toString) assertThrows(classOf[IllegalArgumentException], () => KafkaConfig.fromProps(props)) @@ -769,7 +769,7 @@ class KafkaConfigTest { @Test def testEqualAdvertisedListenersProtocol(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.setProperty(SocketServerConfigs.LISTENERS_CONFIG, "PLAINTEXT://localhost:9092,SSL://localhost:9093") props.setProperty(SocketServerConfigs.ADVERTISED_LISTENERS_CONFIG, "PLAINTEXT://localhost:9092,SSL://localhost:9093") props.setProperty(SocketServerConfigs.LISTENER_SECURITY_PROTOCOL_MAP_CONFIG, "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL") @@ -778,7 +778,7 @@ class KafkaConfigTest { @Test def testInvalidAdvertisedListenersProtocol(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.setProperty(SocketServerConfigs.LISTENERS_CONFIG, "TRACE://localhost:9091,SSL://localhost:9093") props.setProperty(SocketServerConfigs.ADVERTISED_LISTENERS_CONFIG, "PLAINTEXT://localhost:9092") assertBadConfigContainingMessage(props, "No security protocol defined for listener TRACE") @@ -1234,7 +1234,7 @@ class KafkaConfigTest { @Test def testMaxConnectionsPerIpProp(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.setProperty(SocketServerConfigs.MAX_CONNECTIONS_PER_IP_CONFIG, "0") assertFalse(isValidKafkaConfig(props)) props.setProperty(SocketServerConfigs.MAX_CONNECTIONS_PER_IP_OVERRIDES_CONFIG, "127.0.0.1:100") @@ -1382,7 +1382,7 @@ class KafkaConfigTest { } private def assertInvalidQuorumVoters(value: String): Unit = { - val props = TestUtils.createBrokerConfig(0, null) + val props = TestUtils.createBrokerConfig(0) props.setProperty(QuorumConfig.QUORUM_VOTERS_CONFIG, value) assertThrows(classOf[ConfigException], () => KafkaConfig.fromProps(props)) } @@ -1404,7 +1404,7 @@ class KafkaConfigTest { } private def assertValidQuorumVoters(expectedVoters: util.Map[Integer, InetSocketAddress], value: String): Unit = { - val props = TestUtils.createBrokerConfig(0, null) + val props = TestUtils.createBrokerConfig(0) props.setProperty(QuorumConfig.QUORUM_VOTERS_CONFIG, value) val addresses = QuorumConfig.parseVoterConnections(KafkaConfig.fromProps(props).quorumConfig.voters) assertEquals(expectedVoters, addresses) @@ -1417,7 +1417,7 @@ class KafkaConfigTest { InetSocketAddress.createUnresolved("kafka2", 9092) ) - val props = TestUtils.createBrokerConfig(0, null) + val props = TestUtils.createBrokerConfig(0) props.setProperty(QuorumConfig.QUORUM_BOOTSTRAP_SERVERS_CONFIG, "kafka1:9092,kafka2:9092") val addresses = QuorumConfig.parseBootstrapServers( @@ -1596,7 +1596,7 @@ class KafkaConfigTest { @Test def testInvalidAuthorizerClassName(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) val configs = new util.HashMap[Object, Object](props) configs.put(ServerConfigs.AUTHORIZER_CLASS_NAME_CONFIG, null) val ce = assertThrows(classOf[ConfigException], () => KafkaConfig.apply(configs)) @@ -1605,7 +1605,7 @@ class KafkaConfigTest { @Test def testInvalidSecurityInterBrokerProtocol(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.setProperty(ReplicationConfigs.INTER_BROKER_SECURITY_PROTOCOL_CONFIG, "abc") val ce = assertThrows(classOf[ConfigException], () => KafkaConfig.fromProps(props)) assertTrue(ce.getMessage.contains(ReplicationConfigs.INTER_BROKER_SECURITY_PROTOCOL_CONFIG)) @@ -1788,7 +1788,7 @@ class KafkaConfigTest { @Test def testSingleLogDirectoryWithRemoteLogStorage(): Unit = { - val props = TestUtils.createBrokerConfig(0, null, port = 8181) + val props = TestUtils.createBrokerConfig(0, port = 8181) props.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, String.valueOf(true)) props.put(ServerLogConfigs.LOG_DIRS_CONFIG, "/tmp/a") assertDoesNotThrow(() => KafkaConfig.fromProps(props)) diff --git a/core/src/test/scala/unit/kafka/server/KafkaMetricsReporterTest.scala b/core/src/test/scala/unit/kafka/server/KafkaMetricsReporterTest.scala index d319a18bc824a..e07ae3032ca6b 100644 --- a/core/src/test/scala/unit/kafka/server/KafkaMetricsReporterTest.scala +++ b/core/src/test/scala/unit/kafka/server/KafkaMetricsReporterTest.scala @@ -71,7 +71,7 @@ class KafkaMetricsReporterTest extends QuorumTestHarness { @BeforeEach override def setUp(testInfo: TestInfo): Unit = { super.setUp(testInfo) - val props = TestUtils.createBrokerConfig(1, null) + val props = TestUtils.createBrokerConfig(1) props.setProperty(MetricConfigs.METRIC_REPORTER_CLASSES_CONFIG, "kafka.server.KafkaMetricsReporterTest$MockMetricsReporter") props.setProperty(ServerConfigs.BROKER_ID_GENERATION_ENABLE_CONFIG, "true") props.setProperty(ServerConfigs.BROKER_ID_CONFIG, "1") diff --git a/core/src/test/scala/unit/kafka/server/LogRecoveryTest.scala b/core/src/test/scala/unit/kafka/server/LogRecoveryTest.scala index baf51347cb27f..f9970d2967afa 100755 --- a/core/src/test/scala/unit/kafka/server/LogRecoveryTest.scala +++ b/core/src/test/scala/unit/kafka/server/LogRecoveryTest.scala @@ -82,7 +82,7 @@ class LogRecoveryTest extends QuorumTestHarness { override def setUp(testInfo: TestInfo): Unit = { super.setUp(testInfo) - configs = TestUtils.createBrokerConfigs(2, null, enableControlledShutdown = false).map(KafkaConfig.fromProps(_, overridingProps)) + configs = TestUtils.createBrokerConfigs(2, enableControlledShutdown = false).map(KafkaConfig.fromProps(_, overridingProps)) // start both servers server1 = createBroker(configProps1) diff --git a/core/src/test/scala/unit/kafka/server/ReplicaAlterLogDirsThreadTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaAlterLogDirsThreadTest.scala index d61ca335b47e4..c18544002a3e8 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaAlterLogDirsThreadTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicaAlterLogDirsThreadTest.scala @@ -77,7 +77,7 @@ class ReplicaAlterLogDirsThreadTest { @Test def shouldNotAddPartitionIfFutureLogIsNotDefined(): Unit = { val brokerId = 1 - val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(brokerId, "localhost:1234")) + val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(brokerId)) val replicaManager = Mockito.mock(classOf[ReplicaManager]) val quotaManager = Mockito.mock(classOf[ReplicationQuotaManager]) @@ -105,7 +105,7 @@ class ReplicaAlterLogDirsThreadTest { def shouldUpdateLeaderEpochAfterFencedEpochError(): Unit = { val brokerId = 1 val partitionId = 0 - val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(brokerId, "localhost:1234")) + val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(brokerId)) val partition = Mockito.mock(classOf[Partition]) val replicaManager = Mockito.mock(classOf[ReplicaManager]) @@ -205,7 +205,7 @@ class ReplicaAlterLogDirsThreadTest { def shouldReplaceCurrentLogDirWhenCaughtUp(): Unit = { val brokerId = 1 val partitionId = 0 - val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(brokerId, "localhost:1234")) + val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(brokerId)) val partition = Mockito.mock(classOf[Partition]) val replicaManager = Mockito.mock(classOf[ReplicaManager]) @@ -284,7 +284,7 @@ class ReplicaAlterLogDirsThreadTest { def shouldReplaceCurrentLogDirWhenCaughtUpWithAfterAssignmentRequestHasBeenCompleted(): Unit = { val brokerId = 1 val partitionId = 0 - val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(brokerId, "localhost:1234")) + val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(brokerId)) val partition = Mockito.mock(classOf[Partition]) val replicaManager = Mockito.mock(classOf[ReplicaManager]) @@ -381,7 +381,7 @@ class ReplicaAlterLogDirsThreadTest { def shouldRevertAnyScheduledAssignmentRequestIfAssignmentIsCancelled(): Unit = { val brokerId = 1 val partitionId = 0 - val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(brokerId, "localhost:1234")) + val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(brokerId)) val partition = Mockito.mock(classOf[Partition]) val replicaManager = Mockito.mock(classOf[ReplicaManager]) @@ -478,7 +478,7 @@ class ReplicaAlterLogDirsThreadTest { val replicaManager = Mockito.mock(classOf[ReplicaManager]) val directoryEventHandler = mock(classOf[DirectoryEventHandler]) val quotaManager = Mockito.mock(classOf[ReplicationQuotaManager]) - val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:1234")) + val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1)) val endPoint = new BrokerEndPoint(0, "localhost", 1000) val leader = new LocalLeaderEndPoint(endPoint, config, replicaManager, quotaManager) val thread = new ReplicaAlterLogDirsThread( @@ -540,7 +540,7 @@ class ReplicaAlterLogDirsThreadTest { @Test def issuesEpochRequestFromLocalReplica(): Unit = { - val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:1234")) + val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1)) //Setup all dependencies @@ -614,7 +614,7 @@ class ReplicaAlterLogDirsThreadTest { @Test def fetchEpochsFromLeaderShouldHandleExceptionFromGetLocalReplica(): Unit = { - val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:1234")) + val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1)) //Setup all dependencies val partitionT1p0: Partition = mock(classOf[Partition]) @@ -680,7 +680,7 @@ class ReplicaAlterLogDirsThreadTest { val truncateCaptureT1p1: ArgumentCaptor[Long] = ArgumentCaptor.forClass(classOf[Long]) // Setup all the dependencies - val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:1234")) + val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1)) val quotaManager: ReplicationQuotaManager = mock(classOf[ReplicationQuotaManager]) val logManager: LogManager = mock(classOf[LogManager]) val logT1p0: UnifiedLog = mock(classOf[UnifiedLog]) @@ -772,7 +772,7 @@ class ReplicaAlterLogDirsThreadTest { val truncateToCapture: ArgumentCaptor[Long] = ArgumentCaptor.forClass(classOf[Long]) // Setup all the dependencies - val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:1234")) + val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1)) val quotaManager: ReplicationQuotaManager = mock(classOf[ReplicationQuotaManager]) val logManager: LogManager = mock(classOf[LogManager]) val log: UnifiedLog = mock(classOf[UnifiedLog]) @@ -858,7 +858,7 @@ class ReplicaAlterLogDirsThreadTest { val truncated: ArgumentCaptor[Long] = ArgumentCaptor.forClass(classOf[Long]) // Setup all the dependencies - val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:1234")) + val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1)) val quotaManager: ReplicationQuotaManager = mock(classOf[ReplicationQuotaManager]) val logManager: LogManager = mock(classOf[LogManager]) val log: UnifiedLog = mock(classOf[UnifiedLog]) @@ -913,7 +913,7 @@ class ReplicaAlterLogDirsThreadTest { val truncated: ArgumentCaptor[Long] = ArgumentCaptor.forClass(classOf[Long]) // Setup all the dependencies - val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:1234")) + val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1)) val quotaManager: ReplicationQuotaManager = mock(classOf[ReplicationQuotaManager]) val logManager: LogManager = mock(classOf[LogManager]) val log: UnifiedLog = mock(classOf[UnifiedLog]) @@ -1002,7 +1002,7 @@ class ReplicaAlterLogDirsThreadTest { def shouldFetchLeaderEpochOnFirstFetchOnly(): Unit = { //Setup all dependencies - val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:1234")) + val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1)) val quotaManager: ReplicationQuotaManager = mock(classOf[ReplicationQuotaManager]) val logManager: LogManager = mock(classOf[LogManager]) val log: UnifiedLog = mock(classOf[UnifiedLog]) @@ -1065,7 +1065,7 @@ class ReplicaAlterLogDirsThreadTest { def shouldFetchOneReplicaAtATime(): Unit = { //Setup all dependencies - val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:1234")) + val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1)) val quotaManager: ReplicationQuotaManager = mock(classOf[ReplicationQuotaManager]) val logManager: LogManager = mock(classOf[LogManager]) val log: UnifiedLog = mock(classOf[UnifiedLog]) @@ -1117,7 +1117,7 @@ class ReplicaAlterLogDirsThreadTest { def shouldFetchNonDelayedAndNonTruncatingReplicas(): Unit = { //Setup all dependencies - val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:1234")) + val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1)) val quotaManager: ReplicationQuotaManager = mock(classOf[ReplicationQuotaManager]) val logManager: LogManager = mock(classOf[LogManager]) val log: UnifiedLog = mock(classOf[UnifiedLog]) diff --git a/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala index a398fc68cf029..f98846908a8f0 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala @@ -43,7 +43,7 @@ import org.apache.kafka.server.network.BrokerEndPoint import org.apache.kafka.storage.internals.log.LogAppendInfo import org.apache.kafka.storage.log.metrics.BrokerTopicStats import org.junit.jupiter.api.Assertions._ -import org.junit.jupiter.api.{AfterEach, Test} +import org.junit.jupiter.api.{AfterEach, Disabled, Test} import org.junit.jupiter.params.ParameterizedTest import org.junit.jupiter.params.provider.ValueSource import org.mockito.ArgumentCaptor @@ -140,6 +140,7 @@ class ReplicaFetcherThreadTest { ) } + @Disabled("KAFKA-18370") @Test def testFetchLeaderEpochRequestIfLastEpochDefinedForSomePartitions(): Unit = { val config = kafkaConfigNoTruncateOnFetch @@ -244,7 +245,7 @@ class ReplicaFetcherThreadTest { @Test def shouldHandleExceptionFromBlockingSend(): Unit = { - val props = TestUtils.createBrokerConfig(1, "localhost:1234") + val props = TestUtils.createBrokerConfig(1) val config = KafkaConfig.fromProps(props) val mockBlockingSend: BlockingSend = mock(classOf[BlockingSend]) when(mockBlockingSend.brokerEndPoint()).thenReturn(brokerEndPoint) @@ -280,6 +281,7 @@ class ReplicaFetcherThreadTest { verify(mockBlockingSend).sendRequest(any()) } + @Disabled("KAFKA-18370") @Test def shouldFetchLeaderEpochOnFirstFetchOnlyIfLeaderEpochKnownToBothIbp26(): Unit = { verifyFetchLeaderEpochOnFirstFetch(IBP_2_6_IV0) @@ -291,7 +293,7 @@ class ReplicaFetcherThreadTest { } private def verifyFetchLeaderEpochOnFirstFetch(ibp: MetadataVersion, epochFetchCount: Int = 1): Unit = { - val props = TestUtils.createBrokerConfig(1, "localhost:1234") + val props = TestUtils.createBrokerConfig(1) props.setProperty(ReplicationConfigs.INTER_BROKER_PROTOCOL_VERSION_CONFIG, ibp.version) val config = KafkaConfig.fromProps(props) @@ -353,6 +355,7 @@ class ReplicaFetcherThreadTest { assertEquals(3, mockNetwork.fetchCount) } + @Disabled("KAFKA-18370") @Test def shouldTruncateToOffsetSpecifiedInEpochOffsetResponse(): Unit = { @@ -414,6 +417,7 @@ class ReplicaFetcherThreadTest { "Expected " + t2p1 + " to truncate to offset 172 (truncation offsets: " + truncateToCapture.getAllValues + ")") } + @Disabled("KAFKA-18370") @Test def shouldTruncateToOffsetSpecifiedInEpochOffsetResponseIfFollowerHasNoMoreEpochs(): Unit = { // Create a capture to track what partitions/offsets are truncated @@ -475,6 +479,7 @@ class ReplicaFetcherThreadTest { " (truncation offsets: " + truncateToCapture.getAllValues + ")") } + @Disabled("KAFKA-18370") @Test def shouldFetchLeaderEpochSecondTimeIfLeaderRepliesWithEpochNotKnownToFollower(): Unit = { // Create a capture to track what partitions/offsets are truncated @@ -563,7 +568,7 @@ class ReplicaFetcherThreadTest { // Create a capture to track what partitions/offsets are truncated val truncateToCapture: ArgumentCaptor[Long] = ArgumentCaptor.forClass(classOf[Long]) - val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:1234")) + val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1)) // Setup all dependencies val quota: ReplicationQuotaManager = mock(classOf[ReplicationQuotaManager]) @@ -671,7 +676,7 @@ class ReplicaFetcherThreadTest { @Test def testTruncateOnFetchDoesNotUpdateHighWatermark(): Unit = { - val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:1234")) + val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1)) val quota: ReplicationQuotaManager = mock(classOf[ReplicationQuotaManager]) val logManager: LogManager = mock(classOf[LogManager]) val log: UnifiedLog = mock(classOf[UnifiedLog]) @@ -755,7 +760,7 @@ class ReplicaFetcherThreadTest { @Test def testLagIsUpdatedWhenNoRecords(): Unit = { - val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:1234")) + val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1)) val quota: ReplicationQuotaManager = mock(classOf[ReplicationQuotaManager]) val logManager: LogManager = mock(classOf[LogManager]) val log: UnifiedLog = mock(classOf[UnifiedLog]) @@ -848,13 +853,14 @@ class ReplicaFetcherThreadTest { assertEquals(Some(lastFetchedEpoch), thread.fetchState(t1p0).flatMap(_.lastFetchedEpoch)) } + @Disabled("KAFKA-18370") @Test def shouldUseLeaderEndOffsetIfInterBrokerVersionBelow20(): Unit = { // Create a capture to track what partitions/offsets are truncated val truncateToCapture: ArgumentCaptor[Long] = ArgumentCaptor.forClass(classOf[Long]) - val props = TestUtils.createBrokerConfig(1, "localhost:1234") + val props = TestUtils.createBrokerConfig(1) props.put(ReplicationConfigs.INTER_BROKER_PROTOCOL_VERSION_CONFIG, "0.11.0") val config = KafkaConfig.fromProps(props) @@ -923,6 +929,7 @@ class ReplicaFetcherThreadTest { "Expected " + t1p1 + " to truncate to offset 143 (truncation offsets: " + truncateToCapture.getAllValues + ")") } + @Disabled("KAFKA-18370") @Test def shouldTruncateToInitialFetchOffsetIfLeaderReturnsUndefinedOffset(): Unit = { @@ -975,6 +982,7 @@ class ReplicaFetcherThreadTest { assertEquals(initialFetchOffset, truncated.getValue) } + @Disabled("KAFKA-18370") @Test def shouldPollIndefinitelyIfLeaderReturnsAnyException(): Unit = { @@ -1046,6 +1054,7 @@ class ReplicaFetcherThreadTest { assertEquals(156, truncated.getValue) } + @Disabled("KAFKA-18370") @Test def shouldMovePartitionsOutOfTruncatingLogState(): Unit = { val config = kafkaConfigNoTruncateOnFetch @@ -1105,6 +1114,7 @@ class ReplicaFetcherThreadTest { verify(partition, times(2)).truncateTo(0L, false) } + @Disabled("KAFKA-18370") @Test def shouldFilterPartitionsMadeLeaderDuringLeaderEpochRequest(): Unit ={ val config = kafkaConfigNoTruncateOnFetch @@ -1168,7 +1178,7 @@ class ReplicaFetcherThreadTest { @Test def shouldCatchExceptionFromBlockingSendWhenShuttingDownReplicaFetcherThread(): Unit = { - val props = TestUtils.createBrokerConfig(1, "localhost:1234") + val props = TestUtils.createBrokerConfig(1) val config = KafkaConfig.fromProps(props) val mockBlockingSend: BlockingSend = mock(classOf[BlockingSend]) @@ -1216,7 +1226,7 @@ class ReplicaFetcherThreadTest { val tid1p1 = new TopicIdPartition(topicId1, t1p1) val tid2p1 = new TopicIdPartition(topicId2, t2p1) - val props = TestUtils.createBrokerConfig(1, "localhost:1234") + val props = TestUtils.createBrokerConfig(1) val config = KafkaConfig.fromProps(props) val replicaManager: ReplicaManager = mock(classOf[ReplicaManager]) val mockBlockingSend: BlockingSend = mock(classOf[BlockingSend]) @@ -1294,7 +1304,7 @@ class ReplicaFetcherThreadTest { @ParameterizedTest @ValueSource(booleans = Array(true, false)) def testLocalFetchCompletionIfHighWatermarkUpdated(highWatermarkUpdated: Boolean): Unit = { - val props = TestUtils.createBrokerConfig(1, "localhost:1234") + val props = TestUtils.createBrokerConfig(1) val config = KafkaConfig.fromProps(props) val highWatermarkReceivedFromLeader = 100L @@ -1381,7 +1391,7 @@ class ReplicaFetcherThreadTest { } private def assertProcessPartitionDataWhen(isReassigning: Boolean): Unit = { - val props = TestUtils.createBrokerConfig(1, "localhost:1234") + val props = TestUtils.createBrokerConfig(1) val config = KafkaConfig.fromProps(props) val mockBlockingSend: BlockingSend = mock(classOf[BlockingSend]) @@ -1440,7 +1450,7 @@ class ReplicaFetcherThreadTest { } private def kafkaConfigNoTruncateOnFetch: KafkaConfig = { - val props = TestUtils.createBrokerConfig(1, "localhost:1234") + val props = TestUtils.createBrokerConfig(1) props.setProperty(ReplicationConfigs.INTER_BROKER_PROTOCOL_VERSION_CONFIG, IBP_2_6_IV0.version) KafkaConfig.fromProps(props) } diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerQuotasTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerQuotasTest.scala index 2690bcfdc65fb..160fa849d9bd5 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaManagerQuotasTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerQuotasTest.scala @@ -42,7 +42,7 @@ import org.mockito.{AdditionalMatchers, ArgumentMatchers} import scala.jdk.CollectionConverters._ class ReplicaManagerQuotasTest { - val configs = TestUtils.createBrokerConfigs(2, null).map(KafkaConfig.fromProps(_, new Properties())) + val configs = TestUtils.createBrokerConfigs(2).map(KafkaConfig.fromProps(_, new Properties())) val time = new MockTime val metrics = new Metrics val record = new SimpleRecord("some-data-in-a-message".getBytes()) diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala index 950975f9faf97..aa9dff906147d 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala @@ -132,7 +132,7 @@ class ReplicaManagerTest { @BeforeEach def setUp(): Unit = { - val props = TestUtils.createBrokerConfig(1, null) + val props = TestUtils.createBrokerConfig(1) config = KafkaConfig.fromProps(props) alterPartitionManager = mock(classOf[AlterPartitionManager]) quotaManager = QuotaFactory.instantiate(config, metrics, time, "") @@ -191,7 +191,7 @@ class ReplicaManagerTest { @Test def testHighwaterMarkRelativeDirectoryMapping(): Unit = { - val props = TestUtils.createBrokerConfig(1, null) + val props = TestUtils.createBrokerConfig(1) props.put("log.dir", TestUtils.tempRelativeDir("data").getAbsolutePath) val config = KafkaConfig.fromProps(props) val mockLogMgr = TestUtils.createLogManager(config.logDirs.map(new File(_))) @@ -268,7 +268,7 @@ class ReplicaManagerTest { def testMaybeAddLogDirFetchersWithoutEpochCache(): Unit = { val dir1 = TestUtils.tempDir() val dir2 = TestUtils.tempDir() - val props = TestUtils.createBrokerConfig(0, null) + val props = TestUtils.createBrokerConfig(0) props.put("log.dirs", dir1.getAbsolutePath + "," + dir2.getAbsolutePath) val config = KafkaConfig.fromProps(props) val logManager = TestUtils.createLogManager(config.logDirs.map(new File(_)), new LogConfig(new Properties())) @@ -331,7 +331,7 @@ class ReplicaManagerTest { def testMaybeAddLogDirFetchersPausingCleaning(futureLogCreated: Boolean): Unit = { val dir1 = TestUtils.tempDir() val dir2 = TestUtils.tempDir() - val props = TestUtils.createBrokerConfig(0, null) + val props = TestUtils.createBrokerConfig(0) props.put("log.dirs", dir1.getAbsolutePath + "," + dir2.getAbsolutePath) val config = KafkaConfig.fromProps(props) val logManager = TestUtils.createLogManager(config.logDirs.map(new File(_)), new LogConfig(new Properties())) @@ -403,7 +403,7 @@ class ReplicaManagerTest { @Test def testClearPurgatoryOnBecomingFollower(): Unit = { - val props = TestUtils.createBrokerConfig(0, null) + val props = TestUtils.createBrokerConfig(0) props.put("log.dir", TestUtils.tempRelativeDir("data").getAbsolutePath) val config = KafkaConfig.fromProps(props) val logProps = new Properties() @@ -2511,7 +2511,7 @@ class ReplicaManagerTest { @Test def testDisabledTransactionVerification(): Unit = { - val props = TestUtils.createBrokerConfig(0, null) + val props = TestUtils.createBrokerConfig(0) props.put("transaction.partition.verification.enable", "false") val config = KafkaConfig.fromProps(props) @@ -2887,7 +2887,7 @@ class ReplicaManagerTest { leaderEpochFromLeader: Int = 3, extraProps: Properties = new Properties(), topicId: Option[Uuid] = None): (ReplicaManager, LogManager) = { - val props = TestUtils.createBrokerConfig(0, null) + val props = TestUtils.createBrokerConfig(0) props.put("log.dir", TestUtils.tempRelativeDir("data").getAbsolutePath) props.asScala ++= extraProps.asScala val config = KafkaConfig.fromProps(props) @@ -3358,7 +3358,7 @@ class ReplicaManagerTest { buildRemoteLogAuxState: Boolean = false, remoteFetchQuotaExceeded: Option[Boolean] = None ): ReplicaManager = { - val props = TestUtils.createBrokerConfig(brokerId, null) + val props = TestUtils.createBrokerConfig(brokerId) val path1 = TestUtils.tempRelativeDir("data").getAbsolutePath val path2 = TestUtils.tempRelativeDir("data2").getAbsolutePath props.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, enableRemoteStorage.toString) @@ -3719,8 +3719,8 @@ class ReplicaManagerTest { private def prepareDifferentReplicaManagers(brokerTopicStats1: BrokerTopicStats, brokerTopicStats2: BrokerTopicStats): (ReplicaManager, ReplicaManager) = { - val props0 = TestUtils.createBrokerConfig(0, null) - val props1 = TestUtils.createBrokerConfig(1, null) + val props0 = TestUtils.createBrokerConfig(0) + val props1 = TestUtils.createBrokerConfig(1) props0.put("log0.dir", TestUtils.tempRelativeDir("data").getAbsolutePath) props1.put("log1.dir", TestUtils.tempRelativeDir("data").getAbsolutePath) @@ -4607,7 +4607,7 @@ class ReplicaManagerTest { def testReplicaNotAvailable(): Unit = { def createReplicaManager(): ReplicaManager = { - val props = TestUtils.createBrokerConfig(1, null) + val props = TestUtils.createBrokerConfig(1) val config = KafkaConfig.fromProps(props) val mockLogMgr = TestUtils.createLogManager(config.logDirs.map(new File(_))) new ReplicaManager( diff --git a/core/src/test/scala/unit/kafka/server/ReplicationQuotasTest.scala b/core/src/test/scala/unit/kafka/server/ReplicationQuotasTest.scala index fbd6f75a6a99d..ae1e0a1f5e871 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicationQuotasTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicationQuotasTest.scala @@ -88,7 +88,7 @@ class ReplicationQuotasTest extends QuorumTestHarness { * regular replication works as expected. */ - brokers = (100 to 105).map { id => createBroker(fromProps(createBrokerConfig(id, null))) } + brokers = (100 to 105).map { id => createBroker(fromProps(createBrokerConfig(id))) } //Given six partitions, led on nodes 0,1,2,3,4,5 but with followers on node 6,7 (not started yet) //And two extra partitions 6,7, which we don't intend on throttling. @@ -202,7 +202,7 @@ class ReplicationQuotasTest extends QuorumTestHarness { */ //2 brokers with 1MB Segment Size & 1 partition - val config: Properties = createBrokerConfig(100, null) + val config: Properties = createBrokerConfig(100) config.put("log.segment.bytes", (1024 * 1024).toString) brokers = Seq(createBroker(fromProps(config))) @@ -231,7 +231,7 @@ class ReplicationQuotasTest extends QuorumTestHarness { //Start the new broker (and hence start replicating) debug("Starting new broker") - brokers = brokers :+ createBroker(fromProps(createBrokerConfig(101, null))) + brokers = brokers :+ createBroker(fromProps(createBrokerConfig(101))) val start = System.currentTimeMillis() waitForOffsetsToMatch(msgCount, 0, 101) @@ -261,7 +261,7 @@ class ReplicationQuotasTest extends QuorumTestHarness { def createBrokers(brokerIds: Seq[Int]): Unit = { brokerIds.foreach { id => - brokers = brokers :+ createBroker(fromProps(createBrokerConfig(id, null))) + brokers = brokers :+ createBroker(fromProps(createBrokerConfig(id))) } } diff --git a/core/src/test/scala/unit/kafka/server/ServerMetricsTest.scala b/core/src/test/scala/unit/kafka/server/ServerMetricsTest.scala index 9f788ae31231f..e15ec5814f8b3 100755 --- a/core/src/test/scala/unit/kafka/server/ServerMetricsTest.scala +++ b/core/src/test/scala/unit/kafka/server/ServerMetricsTest.scala @@ -29,7 +29,7 @@ class ServerMetricsTest { def testMetricsConfig(): Unit = { val recordingLevels = List(Sensor.RecordingLevel.DEBUG, Sensor.RecordingLevel.INFO) val illegalNames = List("IllegalName", "") - val props = TestUtils.createBrokerConfig(0, "localhost:2818") + val props = TestUtils.createBrokerConfig(0) for (recordingLevel <- recordingLevels) { props.put(MetricConfigs.METRIC_RECORDING_LEVEL_CONFIG, recordingLevel.name) diff --git a/core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala b/core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala index 3a1043198fac8..cb3ebc93785b7 100644 --- a/core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala +++ b/core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala @@ -61,7 +61,7 @@ class ServerShutdownTest extends KafkaServerTestHarness { propsToChangeUponRestart.put(ServerLogConfigs.LOG_DIR_CONFIG, originals.get(ServerLogConfigs.LOG_DIR_CONFIG)) } } - priorConfig = Some(KafkaConfig.fromProps(TestUtils.createBrokerConfigs(1, null).head, propsToChangeUponRestart)) + priorConfig = Some(KafkaConfig.fromProps(TestUtils.createBrokerConfigs(1).head, propsToChangeUponRestart)) Seq(priorConfig.get) } diff --git a/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala b/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala index 7c7e7fd6e0943..591e76c77d75f 100644 --- a/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala +++ b/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala @@ -68,7 +68,7 @@ class LeaderEpochIntegrationTest extends QuorumTestHarness with Logging { @ParameterizedTest @ValueSource(strings = Array("kraft")) def shouldAddCurrentLeaderEpochToMessagesAsTheyAreWrittenToLeader(quorum: String): Unit = { - brokers ++= (0 to 1).map { id => createBroker(fromProps(createBrokerConfig(id, null))) } + brokers ++= (0 to 1).map { id => createBroker(fromProps(createBrokerConfig(id))) } // Given two topics with replication of a single partition for (topic <- List(topic1, topic2)) { @@ -103,7 +103,7 @@ class LeaderEpochIntegrationTest extends QuorumTestHarness with Logging { def shouldSendLeaderEpochRequestAndGetAResponse(quorum: String): Unit = { //3 brokers, put partition on 100/101 and then pretend to be 102 - brokers ++= (100 to 102).map { id => createBroker(fromProps(createBrokerConfig(id, null))) } + brokers ++= (100 to 102).map { id => createBroker(fromProps(createBrokerConfig(id))) } val assignment1 = Map(0 -> Seq(100), 1 -> Seq(101)) createTopic(topic1, assignment1) @@ -150,10 +150,10 @@ class LeaderEpochIntegrationTest extends QuorumTestHarness with Logging { @ValueSource(strings = Array("kraft")) def shouldIncreaseLeaderEpochBetweenLeaderRestarts(quorum: String): Unit = { //Setup: we are only interested in the single partition on broker 101 - brokers += createBroker(fromProps(createBrokerConfig(100, null))) + brokers += createBroker(fromProps(createBrokerConfig(100))) assertEquals(controllerServer.config.nodeId, waitUntilQuorumLeaderElected(controllerServer)) - brokers += createBroker(fromProps(createBrokerConfig(101, null))) + brokers += createBroker(fromProps(createBrokerConfig(101))) def leo() = brokers(1).replicaManager.localLog(tp).get.logEndOffset diff --git a/core/src/test/scala/unit/kafka/server/epoch/OffsetsForLeaderEpochTest.scala b/core/src/test/scala/unit/kafka/server/epoch/OffsetsForLeaderEpochTest.scala index 13c018cf3452c..faab5ab4a6175 100644 --- a/core/src/test/scala/unit/kafka/server/epoch/OffsetsForLeaderEpochTest.scala +++ b/core/src/test/scala/unit/kafka/server/epoch/OffsetsForLeaderEpochTest.scala @@ -38,7 +38,7 @@ import org.mockito.Mockito.{mock, when} import scala.jdk.CollectionConverters._ class OffsetsForLeaderEpochTest { - private val config = TestUtils.createBrokerConfigs(1, null).map(KafkaConfig.fromProps).head + private val config = TestUtils.createBrokerConfigs(1).map(KafkaConfig.fromProps).head private val time = new MockTime private val metrics = new Metrics private val alterIsrManager = TestUtils.createAlterIsrManager() diff --git a/core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataPublisherTest.scala b/core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataPublisherTest.scala index d0e93a56307ea..08b6bbe7f21d2 100644 --- a/core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataPublisherTest.scala +++ b/core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataPublisherTest.scala @@ -181,7 +181,7 @@ class BrokerMetadataPublisherTest { @Test def testNewImagePushedToGroupCoordinator(): Unit = { - val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(0, "")) + val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(0)) val metadataCache = new KRaftMetadataCache(0, () => KRaftVersion.KRAFT_VERSION_1) val logManager = mock(classOf[LogManager]) val replicaManager = mock(classOf[ReplicaManager]) diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala index 45027078ca38e..11de42ea1a633 100755 --- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala +++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala @@ -23,7 +23,6 @@ import kafka.security.JaasTestUtils import kafka.server._ import kafka.server.metadata.{ConfigRepository, MockConfigRepository} import kafka.utils.Implicits._ -import kafka.zk._ import org.apache.kafka.clients.admin.AlterConfigOp.OpType import org.apache.kafka.clients.admin._ import org.apache.kafka.clients.consumer._ @@ -55,14 +54,13 @@ import org.apache.kafka.network.metrics.RequestChannelMetrics import org.apache.kafka.raft.QuorumConfig import org.apache.kafka.server.authorizer.{AuthorizableRequestContext, Authorizer => JAuthorizer} import org.apache.kafka.server.common.{ControllerRequestCompletionHandler, MetadataVersion} -import org.apache.kafka.server.config.{DelegationTokenManagerConfigs, KRaftConfigs, ReplicationConfigs, ServerConfigs, ServerLogConfigs, ZkConfigs} +import org.apache.kafka.server.config.{DelegationTokenManagerConfigs, KRaftConfigs, ReplicationConfigs, ServerConfigs, ServerLogConfigs} import org.apache.kafka.server.metrics.KafkaYammerMetrics import org.apache.kafka.server.util.MockTime import org.apache.kafka.storage.internals.checkpoint.OffsetCheckpointFile import org.apache.kafka.storage.internals.log.{CleanerConfig, LogConfig, LogDirFailureChannel, ProducerStateManagerConfig} import org.apache.kafka.storage.log.metrics.BrokerTopicStats import org.apache.kafka.test.{TestUtils => JTestUtils} -import org.apache.zookeeper.KeeperException.SessionExpiredException import org.junit.jupiter.api.Assertions._ import org.mockito.ArgumentMatchers.{any, anyBoolean} import org.mockito.Mockito @@ -100,14 +98,6 @@ object TestUtils extends Logging { by any other service and hence we use a reserved port. */ val IncorrectBrokerPort = 225 - /** Port to use for unit tests that mock/don't require a real ZK server. */ - val MockZkPort = 1 - /** ZooKeeper connection string to use for unit tests that mock/don't require a real ZK server. */ - val MockZkConnect = "127.0.0.1:" + MockZkPort - - val MockKraftPort = 1 - - val MockKraftConnect = "127.0.0.1:" + MockKraftPort // CN in SSL certificates - this is used for endpoint validation when enabled val SslCertificateCn = "localhost" @@ -157,7 +147,6 @@ object TestUtils extends Logging { */ def createBrokerConfigs( numConfigs: Int, - zkConnect: String, enableControlledShutdown: Boolean = true, enableDeleteTopic: Boolean = true, interBrokerSecurityProtocol: Option[SecurityProtocol] = None, @@ -176,7 +165,7 @@ object TestUtils extends Logging { enableFetchFromFollower: Boolean = false): Seq[Properties] = { val endingIdNumber = startingIdNumber + numConfigs - 1 (startingIdNumber to endingIdNumber).map { node => - createBrokerConfig(node, zkConnect, enableControlledShutdown, enableDeleteTopic, RandomPort, + createBrokerConfig(node, enableControlledShutdown, enableDeleteTopic, RandomPort, interBrokerSecurityProtocol, trustStoreFile, saslProperties, enablePlaintext = enablePlaintext, enableSsl = enableSsl, enableSaslPlaintext = enableSaslPlaintext, enableSaslSsl = enableSaslSsl, rack = rackInfo.get(node), logDirCount = logDirCount, enableToken = enableToken, numPartitions = numPartitions, defaultReplicationFactor = defaultReplicationFactor, enableFetchFromFollower = enableFetchFromFollower) @@ -215,7 +204,7 @@ object TestUtils extends Logging { } def createDummyBrokerConfig(): Properties = { - createBrokerConfig(0, "") + createBrokerConfig(0) } /** @@ -224,7 +213,6 @@ object TestUtils extends Logging { * Note that if `interBrokerSecurityProtocol` is defined, the listener for the `SecurityProtocol` will be enabled. */ def createBrokerConfig(nodeId: Int, - zkConnect: String, enableControlledShutdown: Boolean = true, enableDeleteTopic: Boolean = true, port: Int = RandomPort, @@ -263,19 +251,15 @@ object TestUtils extends Logging { val props = new Properties props.put(ServerConfigs.UNSTABLE_FEATURE_VERSIONS_ENABLE_CONFIG, "true") props.put(ServerConfigs.UNSTABLE_API_VERSIONS_ENABLE_CONFIG, "true") - if (zkConnect == null) { - props.setProperty(KRaftConfigs.SERVER_MAX_STARTUP_TIME_MS_CONFIG, TimeUnit.MINUTES.toMillis(10).toString) - props.put(KRaftConfigs.NODE_ID_CONFIG, nodeId.toString) - props.put(ServerConfigs.BROKER_ID_CONFIG, nodeId.toString) - props.put(SocketServerConfigs.ADVERTISED_LISTENERS_CONFIG, listeners) - props.put(SocketServerConfigs.LISTENERS_CONFIG, listeners) - props.put(KRaftConfigs.CONTROLLER_LISTENER_NAMES_CONFIG, "CONTROLLER") - props.put(SocketServerConfigs.LISTENER_SECURITY_PROTOCOL_MAP_CONFIG, protocolAndPorts. - map(p => "%s:%s".format(p._1, p._1)).mkString(",") + ",CONTROLLER:PLAINTEXT") - } else { - if (nodeId >= 0) props.put(ServerConfigs.BROKER_ID_CONFIG, nodeId.toString) - props.put(SocketServerConfigs.LISTENERS_CONFIG, listeners) - } + props.setProperty(KRaftConfigs.SERVER_MAX_STARTUP_TIME_MS_CONFIG, TimeUnit.MINUTES.toMillis(10).toString) + props.put(KRaftConfigs.NODE_ID_CONFIG, nodeId.toString) + props.put(ServerConfigs.BROKER_ID_CONFIG, nodeId.toString) + props.put(SocketServerConfigs.ADVERTISED_LISTENERS_CONFIG, listeners) + props.put(SocketServerConfigs.LISTENERS_CONFIG, listeners) + props.put(KRaftConfigs.CONTROLLER_LISTENER_NAMES_CONFIG, "CONTROLLER") + props.put(SocketServerConfigs.LISTENER_SECURITY_PROTOCOL_MAP_CONFIG, protocolAndPorts. + map(p => "%s:%s".format(p._1, p._1)).mkString(",") + ",CONTROLLER:PLAINTEXT") + if (logDirCount > 1) { val logDirs = (1 to logDirCount).toList.map(i => // We would like to allow user to specify both relative path and absolute path as log directory for backward-compatibility reason @@ -286,17 +270,12 @@ object TestUtils extends Logging { } else { props.put(ServerLogConfigs.LOG_DIR_CONFIG, tempDir().getAbsolutePath) } - if (zkConnect == null) { - props.put(KRaftConfigs.PROCESS_ROLES_CONFIG, "broker") - // Note: this is just a placeholder value for controller.quorum.voters. JUnit - // tests use random port assignment, so the controller ports are not known ahead of - // time. Therefore, we ignore controller.quorum.voters and use - // controllerQuorumVotersFuture instead. - props.put(QuorumConfig.QUORUM_VOTERS_CONFIG, "1000@localhost:0") - } else { - props.put(ZkConfigs.ZK_CONNECT_CONFIG, zkConnect) - props.put(ZkConfigs.ZK_CONNECTION_TIMEOUT_MS_CONFIG, "10000") - } + props.put(KRaftConfigs.PROCESS_ROLES_CONFIG, "broker") + // Note: this is just a placeholder value for controller.quorum.voters. JUnit + // tests use random port assignment, so the controller ports are not known ahead of + // time. Therefore, we ignore controller.quorum.voters and use + // controllerQuorumVotersFuture instead. + props.put(QuorumConfig.QUORUM_VOTERS_CONFIG, "1000@localhost:0") props.put(ReplicationConfigs.REPLICA_SOCKET_TIMEOUT_MS_CONFIG, "1500") props.put(ReplicationConfigs.CONTROLLER_SOCKET_TIMEOUT_MS_CONFIG, "1500") props.put(ServerConfigs.CONTROLLED_SHUTDOWN_ENABLE_CONFIG, enableControlledShutdown.toString) @@ -504,50 +483,6 @@ object TestUtils extends Logging { controllers.foreach(controller => ensureConsistentKRaftMetadata(brokers, controller)) } - /** - * Create a topic in ZooKeeper using a customized replica assignment. - * Wait until the leader is elected and the metadata is propagated to all brokers. - * Return the leader for each partition. - */ - def createTopic[B <: KafkaBroker](zkClient: KafkaZkClient, - topic: String, - partitionReplicaAssignment: collection.Map[Int, Seq[Int]], - servers: Seq[B]): scala.collection.immutable.Map[Int, Int] = { - createTopic(zkClient, topic, partitionReplicaAssignment, servers, new Properties()) - } - - /** - * Create a topic in ZooKeeper using a customized replica assignment. - * Wait until the leader is elected and the metadata is propagated to all brokers. - * Return the leader for each partition. - */ - def createTopic(zkClient: KafkaZkClient, - topic: String, - partitionReplicaAssignment: collection.Map[Int, Seq[Int]], - servers: Seq[KafkaBroker], - topicConfig: Properties): scala.collection.immutable.Map[Int, Int] = { - val adminZkClient = new AdminZkClient(zkClient) - // create topic - waitUntilTrue( () => { - var hasSessionExpirationException = false - try { - adminZkClient.createTopicWithAssignment(topic, topicConfig, partitionReplicaAssignment) - } catch { - case _: SessionExpiredException => hasSessionExpirationException = true - case e: Throwable => throw e // let other exceptions propagate - } - !hasSessionExpirationException}, - s"Can't create topic $topic") - - // wait until we've propagated all partitions metadata to all servers - val allPartitionsMetadata = waitForAllPartitionsMetadata(servers, topic, partitionReplicaAssignment.size) - - partitionReplicaAssignment.keySet.map { i => - i -> allPartitionsMetadata.get(new TopicPartition(topic, i)).map(_.leader()).getOrElse( - throw new IllegalStateException(s"Cannot get the partition leader for topic: $topic, partition: $i in server metadata cache")) - }.toMap - } - /** * Wrap a single record log buffer. */ diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/CheckpointBench.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/CheckpointBench.java index 07b1ecedd08f6..ddea968215ebd 100644 --- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/CheckpointBench.java +++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/CheckpointBench.java @@ -102,7 +102,7 @@ public class CheckpointBench { public void setup() { this.scheduler = new KafkaScheduler(1, true, "scheduler-thread"); this.brokerProperties = KafkaConfig.fromProps(TestUtils.createBrokerConfig( - 0, null, true, true, 9092, Option.empty(), Option.empty(), + 0, true, true, 9092, Option.empty(), Option.empty(), Option.empty(), true, false, 0, false, 0, false, 0, Option.empty(), 1, true, 1, (short) 1, false)); this.metrics = new Metrics(); diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/PartitionCreationBench.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/PartitionCreationBench.java index c04106040a128..4a50e05433c28 100644 --- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/PartitionCreationBench.java +++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/server/PartitionCreationBench.java @@ -109,7 +109,7 @@ public void setup() { this.scheduler = new KafkaScheduler(1, true, "scheduler-thread"); this.brokerProperties = KafkaConfig.fromProps(TestUtils.createBrokerConfig( - 0, null, true, true, 9092, Option.empty(), Option.empty(), + 0, true, true, 9092, Option.empty(), Option.empty(), Option.empty(), true, false, 0, false, 0, false, 0, Option.empty(), 1, true, 1, (short) 1, false)); this.metrics = new Metrics(); diff --git a/tools/src/test/java/org/apache/kafka/tools/ToolsTestUtils.java b/tools/src/test/java/org/apache/kafka/tools/ToolsTestUtils.java index 64cb72a795550..f279c3e99f8a6 100644 --- a/tools/src/test/java/org/apache/kafka/tools/ToolsTestUtils.java +++ b/tools/src/test/java/org/apache/kafka/tools/ToolsTestUtils.java @@ -16,9 +16,6 @@ */ package org.apache.kafka.tools; -import kafka.utils.TestInfoUtils; -import kafka.utils.TestUtils; - import org.apache.kafka.clients.admin.Admin; import org.apache.kafka.clients.admin.AlterConfigOp; import org.apache.kafka.clients.admin.ConfigEntry; @@ -39,16 +36,11 @@ import java.util.List; import java.util.Map; import java.util.Map.Entry; -import java.util.Properties; import java.util.Set; import java.util.concurrent.ExecutionException; import java.util.stream.Collectors; public class ToolsTestUtils { - /** @see TestInfoUtils#TestWithParameterizedQuorumAndGroupProtocolNames() */ - public static final String TEST_WITH_PARAMETERIZED_QUORUM_AND_GROUP_PROTOCOL_NAMES = "{displayName}.quorum={0}.groupProtocol={1}"; - - private static final int RANDOM_PORT = 0; public static String captureStandardOut(Runnable runnable) { return captureStandardStream(false, runnable); @@ -79,38 +71,6 @@ private static String captureStandardStream(boolean isErr, Runnable runnable) { } } - public static List createBrokerProperties(int numConfigs, String zkConnect, - Map rackInfo, - int numPartitions, - short defaultReplicationFactor) { - - return createBrokerProperties(numConfigs, zkConnect, rackInfo, 1, false, numPartitions, - defaultReplicationFactor, 0); - } - - /** - * Create a test config for the provided parameters. - * - * Note that if `interBrokerSecurityProtocol` is defined, the listener for the `SecurityProtocol` will be enabled. - */ - public static List createBrokerProperties(int numConfigs, String zkConnect, - Map rackInfo, int logDirCount, - boolean enableToken, int numPartitions, short defaultReplicationFactor, - int startingIdNumber) { - List result = new ArrayList<>(); - int endingIdNumber = startingIdNumber + numConfigs - 1; - for (int node = startingIdNumber; node <= endingIdNumber; node++) { - result.add(TestUtils.createBrokerConfig(node, zkConnect, true, true, RANDOM_PORT, - scala.Option.empty(), - scala.Option.empty(), - scala.Option.empty(), - true, false, RANDOM_PORT, false, RANDOM_PORT, false, RANDOM_PORT, - scala.Option.apply(rackInfo.get(node)), - logDirCount, enableToken, numPartitions, defaultReplicationFactor, false)); - } - return result; - } - /** * Set broker replication quotas and enable throttling for a set of partitions. This * will override any previous replication quotas, but will leave the throttling status