This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 0ccf386caf7 MINOR: Use Map for internal topic configs (#22474)
0ccf386caf7 is described below
commit 0ccf386caf76ede690d7f37dd9fd06ada6d9dd6a
Author: Kuan-Po Tseng <[email protected]>
AuthorDate: Mon Jun 15 14:46:57 2026 +0800
MINOR: Use Map for internal topic configs (#22474)
Replace Properties with Map<String, String> for internal topic
configuration APIs and test topic creation helpers.
This updates group, share, and transaction coordinator config accessors,
converts auto-topic creation config handling to typed maps, and adjusts
related tests to pass map-based topic configs.
Reviewers: Chia-Ping Tsai <[email protected]>
---
.../producer/ProducerFailureHandlingTest.java | 8 +--
.../transaction/TransactionCoordinator.scala | 17 +++--
.../kafka/admin/RemoteTopicCrudTest.scala | 11 ++-
.../kafka/api/BaseProducerSendTest.scala | 14 ++--
.../integration/kafka/api/ConsumerBounceTest.scala | 3 +-
.../kafka/api/EndToEndAuthorizationTest.scala | 2 +-
.../kafka/api/PlaintextAdminIntegrationTest.scala | 37 +++++-----
.../kafka/api/PlaintextProducerSendTest.scala | 18 ++---
.../kafka/api/TransactionsBounceTest.scala | 8 +--
.../server/DynamicBrokerReconfigurationTest.scala | 11 ++-
.../kafka/integration/KafkaServerTestHarness.scala | 2 +-
.../scala/unit/kafka/metrics/MetricsTest.scala | 5 +-
.../AddPartitionsToTxnRequestServerTest.scala | 3 +-
.../server/AlterReplicaLogDirsRequestTest.scala | 17 ++---
.../unit/kafka/server/BaseFetchRequestTest.scala | 6 +-
.../kafka/server/DynamicConfigChangeTest.scala | 10 ++-
.../kafka/server/FetchRequestMaxBytesTest.scala | 6 +-
.../server/GroupCoordinatorBaseRequestTest.scala | 6 +-
.../scala/unit/kafka/server/KafkaApisTest.scala | 8 +--
.../ListOffsetsRequestWithRemoteStoreTest.scala | 5 +-
.../test/scala/unit/kafka/utils/TestUtils.scala | 11 ++-
.../kafka/coordinator/group/GroupCoordinator.java | 8 +--
.../coordinator/group/GroupCoordinatorService.java | 12 ++--
.../group/GroupCoordinatorServiceTest.java | 15 +++--
.../server/DefaultAutoTopicCreationManager.java | 26 ++++----
.../DefaultAutoTopicCreationManagerTest.java | 78 +++++++++++-----------
.../kafka/coordinator/share/ShareCoordinator.java | 8 +--
.../coordinator/share/ShareCoordinatorService.java | 17 +++--
.../share/ShareCoordinatorServiceTest.java | 7 +-
29 files changed, 180 insertions(+), 199 deletions(-)
diff --git
a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerFailureHandlingTest.java
b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerFailureHandlingTest.java
index e816e00b2a5..4df1d0bb3b6 100644
---
a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerFailureHandlingTest.java
+++
b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerFailureHandlingTest.java
@@ -218,11 +218,9 @@ public class ProducerFailureHandlingTest {
@ClusterTest
public void testCannotSendToInternalTopic(ClusterInstance clusterInstance)
throws InterruptedException {
try (Admin admin = clusterInstance.admin()) {
- Map<String, String> topicConfig = new HashMap<>();
- clusterInstance.brokers().get(0)
- .groupCoordinator()
- .groupMetadataTopicConfigs()
- .forEach((k, v) -> topicConfig.put(k.toString(),
v.toString()));
+ Map<String, String> topicConfig = clusterInstance.brokers().get(0)
+ .groupCoordinator()
+ .groupMetadataTopicConfigs();
admin.createTopics(List.of(new
NewTopic(Topic.GROUP_METADATA_TOPIC_NAME, 1, (short) 1).configs(topicConfig)));
clusterInstance.waitTopicDeletion(Topic.GROUP_METADATA_TOPIC_NAME);
}
diff --git
a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala
b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala
index 70db45b1eba..771c6727072 100644
---
a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala
+++
b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala
@@ -36,7 +36,6 @@ import org.apache.kafka.server.record.BrokerCompressionType
import org.apache.kafka.server.util.Scheduler
import java.util
-import java.util.Properties
import java.util.concurrent.atomic.AtomicBoolean
import scala.jdk.OptionConverters._
@@ -1012,14 +1011,14 @@ class TransactionCoordinator(txnConfig:
TransactionConfig,
*
* @return Properties of the transaction state topic.
*/
- def transactionStateTopicConfigs: Properties = {
- val props = new Properties
- props.put(TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, "false")
- props.put(TopicConfig.COMPRESSION_TYPE_CONFIG,
BrokerCompressionType.UNCOMPRESSED.name)
- props.put(TopicConfig.CLEANUP_POLICY_CONFIG,
TopicConfig.CLEANUP_POLICY_COMPACT)
- props.put(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG,
txnConfig.transactionLogMinInsyncReplicas.toString)
- props.put(TopicConfig.SEGMENT_BYTES_CONFIG,
txnConfig.transactionLogSegmentBytes.toString)
- props
+ def transactionStateTopicConfigs: util.Map[String, String] = {
+ util.Map.of(
+ TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, "false",
+ TopicConfig.COMPRESSION_TYPE_CONFIG,
BrokerCompressionType.UNCOMPRESSED.name,
+ TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT,
+ TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG,
txnConfig.transactionLogMinInsyncReplicas.toString,
+ TopicConfig.SEGMENT_BYTES_CONFIG,
txnConfig.transactionLogSegmentBytes.toString
+ )
}
def partitionFor(transactionalId: String): Int =
txnManager.partitionFor(transactionalId)
diff --git
a/core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala
b/core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala
index 2ac15a29e20..606517d22f3 100644
--- a/core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala
+++ b/core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala
@@ -25,6 +25,7 @@ import org.apache.kafka.server.log.remote.storage._
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{BeforeEach, Tag, Test, TestInfo}
+import java.util
import java.util.Properties
import scala.collection.Seq
import scala.util.Random
@@ -58,11 +59,10 @@ class RemoteTopicCrudTest extends IntegrationTestHarness {
@Test
def testClusterWideDisablementOfTieredStorageWithEnabledTieredTopic(): Unit
= {
- val topicConfig = new Properties()
- topicConfig.setProperty(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG,
"true")
+ val topicConfigs =
util.Map.of(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true")
TestUtils.createTopicWithAdmin(createAdminClient(), testTopicName,
brokers, controllerServers, numPartitions, brokerCount,
- topicConfig = topicConfig)
+ topicConfig = topicConfigs)
val tsDisabledProps = TestUtils.createBrokerConfigs(1).head
instanceConfigs = List(KafkaConfig.fromProps(tsDisabledProps))
@@ -75,11 +75,10 @@ class RemoteTopicCrudTest extends IntegrationTestHarness {
@Test
def
testClusterWithoutTieredStorageStartsSuccessfullyIfTopicWithTieringDisabled():
Unit = {
- val topicConfig = new Properties()
- topicConfig.setProperty(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG,
false.toString)
+ val topicConfigs =
util.Map.of(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, false.toString)
TestUtils.createTopicWithAdmin(createAdminClient(), testTopicName,
brokers, controllerServers, numPartitions, brokerCount,
- topicConfig = topicConfig)
+ topicConfig = topicConfigs)
val tsDisabledProps = TestUtils.createBrokerConfigs(1).head
instanceConfigs = List(KafkaConfig.fromProps(tsDisabledProps))
diff --git
a/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala
b/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala
index 77e8fdf2214..71d0dd82739 100644
--- a/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala
+++ b/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala
@@ -19,6 +19,7 @@ package kafka.api
import java.time.Duration
import java.nio.charset.StandardCharsets
+import java.util
import java.util.Properties
import java.util.concurrent.TimeUnit
import kafka.integration.KafkaServerTestHarness
@@ -265,12 +266,11 @@ abstract class BaseProducerSendTest extends
KafkaServerTestHarness {
try {
// create topic
- val topicProps = new Properties()
- if (timestampType == TimestampType.LOG_APPEND_TIME)
- topicProps.setProperty(TopicConfig.MESSAGE_TIMESTAMP_TYPE_CONFIG,
"LogAppendTime")
- else
- topicProps.setProperty(TopicConfig.MESSAGE_TIMESTAMP_TYPE_CONFIG,
"CreateTime")
- TestUtils.createTopicWithAdmin(admin, topic, brokers, controllerServers,
1, 2, topicConfig = topicProps)
+ val topicConfigs = util.Map.of(
+ TopicConfig.MESSAGE_TIMESTAMP_TYPE_CONFIG,
+ if (timestampType == TimestampType.LOG_APPEND_TIME) "LogAppendTime"
else "CreateTime"
+ )
+ TestUtils.createTopicWithAdmin(admin, topic, brokers, controllerServers,
1, 2, topicConfig = topicConfigs)
val recordAndFutures = for (i <- 1 to numRecords) yield {
val record = new ProducerRecord(topic, partition, baseTimestamp + i,
s"key$i".getBytes(StandardCharsets.UTF_8),
@@ -448,7 +448,7 @@ abstract class BaseProducerSendTest extends
KafkaServerTestHarness {
val e = assertThrows(classOf[ExecutionException], () => producer.send(new
ProducerRecord(topic, partition1, null,
"value".getBytes(StandardCharsets.UTF_8))).get())
assertEquals(classOf[TimeoutException], e.getCause.getClass)
- admin.createPartitions(java.util.Map.of(topic,
NewPartitions.increaseTo(2))).all().get()
+ admin.createPartitions(util.Map.of(topic,
NewPartitions.increaseTo(2))).all().get()
// read metadata from a broker and verify the new topic partitions exist
TestUtils.waitForPartitionMetadata(brokers, topic, 0)
diff --git a/core/src/test/scala/integration/kafka/api/ConsumerBounceTest.scala
b/core/src/test/scala/integration/kafka/api/ConsumerBounceTest.scala
index 2db1fad5dd2..f0089f3de5b 100644
--- a/core/src/test/scala/integration/kafka/api/ConsumerBounceTest.scala
+++ b/core/src/test/scala/integration/kafka/api/ConsumerBounceTest.scala
@@ -13,6 +13,7 @@
package kafka.api
+import java.util
import java.util.concurrent._
import java.util.Properties
import kafka.server.KafkaConfig
@@ -138,7 +139,7 @@ class ConsumerBounceTest extends AbstractConsumerTest with
Logging {
}
private def createTopicPartitions(topic: String, numPartitions: Int,
replicationFactor: Int,
- topicConfig: Properties = new Properties):
Set[TopicPartition] = {
+ topicConfig: util.Map[String, String] =
util.Map.of()): Set[TopicPartition] = {
createTopic(topic, numPartitions = numPartitions, replicationFactor =
replicationFactor, topicConfig = topicConfig)
Range(0, numPartitions).map(part => new TopicPartition(topic, part)).toSet
}
diff --git
a/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala
b/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala
index 40bb4f649cb..65828486e76 100644
--- a/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala
@@ -156,7 +156,7 @@ abstract class EndToEndAuthorizationTest extends
IntegrationTestHarness with Sas
// create the test topic with all the brokers as replicas
val superuserAdminClient = createSuperuserAdminClient()
TestUtils.createTopicWithAdmin(admin = superuserAdminClient, topic =
topic, brokers = brokers, controllers = controllerServers,
- replicationFactor = 3, topicConfig = new Properties)
+ replicationFactor = 3, topicConfig = util.Map.of())
}
/**
diff --git
a/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala
b/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala
index 2e4b7acf26f..913aec618a2 100644
---
a/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala
+++
b/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala
@@ -1018,11 +1018,12 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
// Create topics
val topic1 = "describe-alter-configs-topic-1"
val topicResource1 = new ConfigResource(ConfigResource.Type.TOPIC, topic1)
- val topicConfig1 = new Properties
val maxMessageBytes = "500000"
val retentionMs = "60000000"
- topicConfig1.setProperty(TopicConfig.MAX_MESSAGE_BYTES_CONFIG,
maxMessageBytes)
- topicConfig1.setProperty(TopicConfig.RETENTION_MS_CONFIG, retentionMs)
+ val topicConfig1 = util.Map.of(
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG, maxMessageBytes,
+ TopicConfig.RETENTION_MS_CONFIG, retentionMs
+ )
createTopic(topic1, numPartitions = 1, replicationFactor = 1, topicConfig1)
val topic2 = "describe-alter-configs-topic-2"
@@ -1621,9 +1622,8 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
@ParameterizedTest(name =
TestInfoUtils.TestWithParameterizedGroupProtocolNames)
@MethodSource(Array("getTestGroupProtocolParametersAll"))
def testDeleteRecordsAfterCorruptRecords(groupProtocol: String): Unit = {
- val config = new Properties()
- config.put(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, "200")
- createTopic(topic, numPartitions = 1, replicationFactor = 1, config)
+ val configs = util.Map.of(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, "200")
+ createTopic(topic, numPartitions = 1, replicationFactor = 1, configs)
client = createAdminClient
@@ -3521,9 +3521,10 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
// Create topics
val topic1 = "incremental-alter-configs-topic-1"
val topic1Resource = new ConfigResource(ConfigResource.Type.TOPIC, topic1)
- val topic1CreateConfigs = new Properties
- topic1CreateConfigs.setProperty(TopicConfig.RETENTION_MS_CONFIG,
"60000000")
- topic1CreateConfigs.setProperty(TopicConfig.CLEANUP_POLICY_CONFIG,
TopicConfig.CLEANUP_POLICY_COMPACT)
+ val topic1CreateConfigs = util.Map.of(
+ TopicConfig.RETENTION_MS_CONFIG, "60000000",
+ TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT
+ )
createTopic(topic1, numPartitions = 1, replicationFactor = 1,
topic1CreateConfigs)
val topic2 = "incremental-alter-configs-topic-2"
@@ -3640,8 +3641,7 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
val subtractValues = brokers.tail.map(broker =>
s"0:${broker.config.brokerId}").mkString(",")
assertNotEquals("", subtractValues)
- val topicCreateConfigs = new Properties
-
topicCreateConfigs.setProperty(QuotaConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG,
appendValues)
+ val topicCreateConfigs =
util.Map.of(QuotaConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG,
appendValues)
createTopic(topic, numPartitions = 1, replicationFactor = 1,
topicCreateConfigs)
// Append value that is already present
@@ -3898,8 +3898,8 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
def validateLogConfig(compressionType: String): Unit = {
ensureConsistentKRaftMetadata()
- val topicProps = brokers.head.metadataCache.topicConfig(topic)
- val logConfig = LogConfig.fromProps(util.Map.of[String, AnyRef],
topicProps)
+ val topicConfigs = brokers.head.metadataCache.topicConfig(topic)
+ val logConfig = LogConfig.fromProps(util.Map.of[String, AnyRef],
topicConfigs)
assertEquals(compressionType,
logConfig.originals.get(TopicConfig.COMPRESSION_TYPE_CONFIG))
assertNull(logConfig.originals.get(TopicConfig.RETENTION_BYTES_CONFIG))
@@ -4130,14 +4130,13 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
@Test
def testAppendConfigToEmptyDefaultValue(): Unit = {
- testAppendConfig(new Properties(), "0:0", "0:0")
+ testAppendConfig(util.Map.of(), "0:0", "0:0")
}
@Test
def testAppendConfigToExistentValue(): Unit = {
- val props = new Properties()
-
props.setProperty(QuotaConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG,
"1:1")
- testAppendConfig(props, "0:0", "1:1,0:0")
+ val configs =
util.Map.of(QuotaConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, "1:1")
+ testAppendConfig(configs, "0:0", "1:1,0:0")
}
private def disableEligibleLeaderReplicas(admin: Admin): Unit = {
@@ -4147,9 +4146,9 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
}
}
- private def testAppendConfig(props: Properties, append: String, expected:
String): Unit = {
+ private def testAppendConfig(configs: util.Map[String, String], append:
String, expected: String): Unit = {
client = createAdminClient
- createTopic(topic, topicConfig = props)
+ createTopic(topic, topicConfig = configs)
val topicResource = new ConfigResource(ConfigResource.Type.TOPIC, topic)
val topicAlterConfigs = util.List.of(
new AlterConfigOp(new
ConfigEntry(QuotaConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, append),
AlterConfigOp.OpType.APPEND),
diff --git
a/core/src/test/scala/integration/kafka/api/PlaintextProducerSendTest.scala
b/core/src/test/scala/integration/kafka/api/PlaintextProducerSendTest.scala
index d3a76ce4bda..960a362dab7 100644
--- a/core/src/test/scala/integration/kafka/api/PlaintextProducerSendTest.scala
+++ b/core/src/test/scala/integration/kafka/api/PlaintextProducerSendTest.scala
@@ -17,6 +17,7 @@
package kafka.api
+import java.util
import java.util.{Locale, Properties}
import java.util.concurrent.{ExecutionException, Future, TimeUnit}
import kafka.utils.{TestInfoUtils, TestUtils}
@@ -187,11 +188,10 @@ class PlaintextProducerSendTest extends
BaseProducerSendTest {
@ParameterizedTest(name =
TestInfoUtils.TestWithParameterizedGroupProtocolNames)
@MethodSource(Array("timestampConfigProvider"))
def testSendWithInvalidBeforeAndAfterTimestamp(groupProtocol: String,
messageTimeStampConfig: String, recordTimestamp: Long): Unit = {
- val topicProps = new Properties()
// set the TopicConfig for timestamp validation to have 1 minute
threshold. Note that recordTimestamp has 5 minutes diff
val oneMinuteInMs: Long = 1 * 60 * 60 * 1000L
- topicProps.setProperty(messageTimeStampConfig, oneMinuteInMs.toString)
- TestUtils.createTopicWithAdmin(admin, topic, brokers, controllerServers,
1, 2, topicConfig = topicProps)
+ val topicConfigs = util.Map.of(messageTimeStampConfig,
oneMinuteInMs.toString)
+ TestUtils.createTopicWithAdmin(admin, topic, brokers, controllerServers,
1, 2, topicConfig = topicConfigs)
val producer = createProducer()
try {
@@ -216,11 +216,9 @@ class PlaintextProducerSendTest extends
BaseProducerSendTest {
@ParameterizedTest(name =
TestInfoUtils.TestWithParameterizedGroupProtocolNames)
@MethodSource(Array("timestampConfigProvider"))
def testValidBeforeAndAfterTimestampsAtThreshold(groupProtocol: String,
messageTimeStampConfig: String, recordTimestamp: Long): Unit = {
- val topicProps = new Properties()
-
// set the TopicConfig for timestamp validation to be the same as the
record timestamp
- topicProps.setProperty(messageTimeStampConfig, recordTimestamp.toString)
- TestUtils.createTopicWithAdmin(admin, topic, brokers, controllerServers,
1, 2, topicConfig = topicProps)
+ val topicConfigs = util.Map.of(messageTimeStampConfig,
recordTimestamp.toString)
+ TestUtils.createTopicWithAdmin(admin, topic, brokers, controllerServers,
1, 2, topicConfig = topicConfigs)
val producer = createProducer()
@@ -236,12 +234,10 @@ class PlaintextProducerSendTest extends
BaseProducerSendTest {
@ParameterizedTest(name =
TestInfoUtils.TestWithParameterizedGroupProtocolNames)
@MethodSource(Array("timestampConfigProvider"))
def testValidBeforeAndAfterTimestampsWithinThreshold(groupProtocol: String,
messageTimeStampConfig: String, recordTimestamp: Long): Unit = {
- val topicProps = new Properties()
-
// set the TopicConfig for timestamp validation to have 10 minute
threshold. Note that recordTimestamp has 5 minutes diff
val tenMinutesInMs: Long = 10 * 60 * 60 * 1000L
- topicProps.setProperty(messageTimeStampConfig, tenMinutesInMs.toString)
- TestUtils.createTopicWithAdmin(admin, topic, brokers, controllerServers,
1, 2, topicConfig = topicProps)
+ val topicConfigs = util.Map.of(messageTimeStampConfig,
tenMinutesInMs.toString)
+ TestUtils.createTopicWithAdmin(admin, topic, brokers, controllerServers,
1, 2, topicConfig = topicConfigs)
val producer = createProducer()
diff --git
a/core/src/test/scala/integration/kafka/api/TransactionsBounceTest.scala
b/core/src/test/scala/integration/kafka/api/TransactionsBounceTest.scala
index 8c95aaf49bc..d4d065d9a37 100644
--- a/core/src/test/scala/integration/kafka/api/TransactionsBounceTest.scala
+++ b/core/src/test/scala/integration/kafka/api/TransactionsBounceTest.scala
@@ -17,6 +17,7 @@
package kafka.api
+import java.util
import java.util.Properties
import kafka.server.KafkaConfig
import kafka.utils.{TestInfoUtils, TestUtils}
@@ -177,10 +178,9 @@ class TransactionsBounceTest extends
IntegrationTestHarness {
}
private def createTopics() = {
- val topicConfig = new Properties()
- topicConfig.put(ServerLogConfigs.MIN_IN_SYNC_REPLICAS_CONFIG, 2.toString)
- createTopic(inputTopic, numPartitions, 3, topicConfig)
- createTopic(outputTopic, numPartitions, 3, topicConfig)
+ val topicConfigs =
util.Map.of(ServerLogConfigs.MIN_IN_SYNC_REPLICAS_CONFIG, 2.toString)
+ createTopic(inputTopic, numPartitions, 3, topicConfigs)
+ createTopic(outputTopic, numPartitions, 3, topicConfigs)
}
private class BounceScheduler extends
ShutdownableThread("daemon-broker-bouncer", false) {
diff --git
a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
index c3706c7bb0c..ce51f80632e 100644
---
a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
+++
b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
@@ -568,9 +568,8 @@ class DynamicBrokerReconfigurationTest extends
QuorumTestHarness with SaslSetup
@MethodSource(Array("getTestGroupProtocolParametersAll"))
def testConsecutiveConfigChange(groupProtocol: String): Unit = {
val topic2 = "testtopic2"
- val topicProps = new Properties
- topicProps.put(ServerLogConfigs.MIN_IN_SYNC_REPLICAS_CONFIG, "2")
- TestUtils.createTopicWithAdmin(adminClients.head, topic2, servers,
controllerServers, numPartitions = 1, replicationFactor = numServers,
topicConfig = topicProps)
+ val topicConfigs =
util.Map.of(ServerLogConfigs.MIN_IN_SYNC_REPLICAS_CONFIG, "2")
+ TestUtils.createTopicWithAdmin(adminClients.head, topic2, servers,
controllerServers, numPartitions = 1, replicationFactor = numServers,
topicConfig = topicConfigs)
def getLogOrThrow(tp: TopicPartition): UnifiedLog = {
var (logOpt, found) = TestUtils.computeUntilTrue {
@@ -1255,7 +1254,7 @@ class DynamicBrokerReconfigurationTest extends
QuorumTestHarness with SaslSetup
private def alterConfigsOnServer(server: KafkaBroker, props: Properties):
Unit = {
val configEntries = props.asScala.map { case (k, v) => new
AlterConfigOp(new ConfigEntry(k, v), OpType.SET) }.toList.asJava
- val alterConfigs = new java.util.HashMap[ConfigResource,
java.util.Collection[AlterConfigOp]]()
+ val alterConfigs = new util.HashMap[ConfigResource,
util.Collection[AlterConfigOp]]()
alterConfigs.put(new ConfigResource(ConfigResource.Type.BROKER,
server.config.brokerId.toString), configEntries)
adminClients.head.incrementalAlterConfigs(alterConfigs)
props.asScala.foreach { case (k, v) => waitForConfigOnServer(server, k, v)
}
@@ -1276,11 +1275,11 @@ class DynamicBrokerReconfigurationTest extends
QuorumTestHarness with SaslSetup
perBrokerConfig: Boolean): AlterConfigsResult
= {
val configEntries = props.asScala.map { case (k, v) => new
AlterConfigOp(new ConfigEntry(k, v), OpType.SET) }.toList.asJava
val configs = if (perBrokerConfig) {
- val alterConfigs = new java.util.HashMap[ConfigResource,
java.util.Collection[AlterConfigOp]]()
+ val alterConfigs = new util.HashMap[ConfigResource,
util.Collection[AlterConfigOp]]()
servers.foreach(server => alterConfigs.put(new
ConfigResource(ConfigResource.Type.BROKER, server.config.brokerId.toString),
configEntries))
alterConfigs
} else {
- val alterConfigs = new java.util.HashMap[ConfigResource,
java.util.Collection[AlterConfigOp]]()
+ val alterConfigs = new util.HashMap[ConfigResource,
util.Collection[AlterConfigOp]]()
alterConfigs.put(new ConfigResource(ConfigResource.Type.BROKER, ""),
configEntries)
alterConfigs
}
diff --git
a/core/src/test/scala/unit/kafka/integration/KafkaServerTestHarness.scala
b/core/src/test/scala/unit/kafka/integration/KafkaServerTestHarness.scala
index 4b6690a4fdc..c6a0414a85f 100755
--- a/core/src/test/scala/unit/kafka/integration/KafkaServerTestHarness.scala
+++ b/core/src/test/scala/unit/kafka/integration/KafkaServerTestHarness.scala
@@ -165,7 +165,7 @@ abstract class KafkaServerTestHarness extends
QuorumTestHarness {
topic: String,
numPartitions: Int = 1,
replicationFactor: Int = 1,
- topicConfig: Properties = new Properties,
+ topicConfig: util.Map[String, String] = util.Map.of(),
listenerName: ListenerName = listenerName,
adminClientConfig: Properties = new Properties
): scala.collection.immutable.Map[Int, Int] = {
diff --git a/core/src/test/scala/unit/kafka/metrics/MetricsTest.scala
b/core/src/test/scala/unit/kafka/metrics/MetricsTest.scala
index e30531779c2..64b506e0065 100644
--- a/core/src/test/scala/unit/kafka/metrics/MetricsTest.scala
+++ b/core/src/test/scala/unit/kafka/metrics/MetricsTest.scala
@@ -175,9 +175,8 @@ class MetricsTest extends KafkaServerTestHarness with
Logging {
val bytesIn = s"${BrokerTopicMetrics.BYTES_IN_PER_SEC},topic=$topic"
val bytesOut = s"${BrokerTopicMetrics.BYTES_OUT_PER_SEC},topic=$topic"
- val topicConfig = new Properties
- topicConfig.setProperty(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2")
- createTopic(topic, 1, numNodes, topicConfig)
+ val topicConfigs = util.Map.of(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG,
"2")
+ createTopic(topic, 1, numNodes, topicConfigs)
// Produce a few messages to create the metrics
TestUtils.generateAndProduceMessages(brokers, topic, nMessages)
diff --git
a/core/src/test/scala/unit/kafka/server/AddPartitionsToTxnRequestServerTest.scala
b/core/src/test/scala/unit/kafka/server/AddPartitionsToTxnRequestServerTest.scala
index 406609239a0..efcb8104efc 100644
---
a/core/src/test/scala/unit/kafka/server/AddPartitionsToTxnRequestServerTest.scala
+++
b/core/src/test/scala/unit/kafka/server/AddPartitionsToTxnRequestServerTest.scala
@@ -19,6 +19,7 @@ package kafka.server
import kafka.utils.TestUtils
+import java.util
import java.util.{Collections, Properties}
import java.util.stream.{Stream => JStream}
import org.apache.kafka.common.TopicPartition
@@ -50,7 +51,7 @@ class AddPartitionsToTxnRequestServerTest extends
BaseRequestTest {
@BeforeEach
override def setUp(testInfo: TestInfo): Unit = {
super.setUp(testInfo)
- createTopic(topic1, numPartitions, brokers.size, new Properties())
+ createTopic(topic1, numPartitions, brokers.size, util.Map.of())
}
@ParameterizedTest
diff --git
a/core/src/test/scala/unit/kafka/server/AlterReplicaLogDirsRequestTest.scala
b/core/src/test/scala/unit/kafka/server/AlterReplicaLogDirsRequestTest.scala
index 16cce3ed81a..a27925e4e34 100644
--- a/core/src/test/scala/unit/kafka/server/AlterReplicaLogDirsRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/AlterReplicaLogDirsRequestTest.scala
@@ -17,6 +17,7 @@
package kafka.server
+import java.util
import java.io.File
import kafka.utils._
import org.apache.kafka.common.TopicPartition
@@ -138,15 +139,15 @@ class AlterReplicaLogDirsRequestTest extends
BaseRequestTest {
assertEquals(Errors.UNKNOWN_TOPIC_OR_PARTITION,
findErrorForPartition(alterReplicaLogDirsResponse1, tp))
assertTrue(brokers.head.logManager.getLog(tp).isEmpty)
- val topicProperties = new Properties()
- topicProperties.put(TopicConfig.RETENTION_BYTES_CONFIG, "1024")
- // This test needs enough time to wait for dir movement happened.
- // We don't want files with `.deleted` suffix are removed too fast,
- // so we can validate there will be orphan files and orphan files will be
removed eventually.
- topicProperties.put(TopicConfig.FILE_DELETE_DELAY_MS_CONFIG, "10000")
- topicProperties.put(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, "1024")
+ val topicConfigs = util.Map.of(
+ TopicConfig.RETENTION_BYTES_CONFIG, "1024",
+ // This test needs enough time to wait for dir movement happened.
+ // We don't want files with `.deleted` suffix are removed too fast,
+ // so we can validate there will be orphan files and orphan files will
be removed eventually.
+ TopicConfig.FILE_DELETE_DELAY_MS_CONFIG, "10000",
+ LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, "1024")
- createTopic(topic, partitionNum, 1, topicProperties)
+ createTopic(topic, partitionNum, 1, topicConfigs)
assertEquals(logDir1, brokers.head.logManager.getLog(tp).get.dir.getParent)
// send enough records to trigger log rolling
diff --git a/core/src/test/scala/unit/kafka/server/BaseFetchRequestTest.scala
b/core/src/test/scala/unit/kafka/server/BaseFetchRequestTest.scala
index 30cd37023ef..978b0eb9c21 100644
--- a/core/src/test/scala/unit/kafka/server/BaseFetchRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/BaseFetchRequestTest.scala
@@ -85,9 +85,9 @@ class BaseFetchRequestTest extends BaseRequestTest {
protected def createTopics(numTopics: Int, numPartitions: Int, configs:
Map[String, String] = Map.empty): Map[TopicPartition, Int] = {
val topics = (0 until numTopics).map(t => s"topic$t")
- val topicConfig = new Properties
- topicConfig.setProperty(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG,
2.toString)
- configs.foreach { case (k, v) => topicConfig.setProperty(k, v) }
+ val topicConfig = new util.HashMap[String, String]()
+ topicConfig.put(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, 2.toString)
+ configs.foreach { case (k, v) => topicConfig.put(k, v) }
topics.flatMap { topic =>
val partitionToLeader = createTopic(topic, numPartitions =
numPartitions, replicationFactor = 2,
topicConfig = topicConfig)
diff --git
a/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala
b/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala
index 0f510d1a06a..a282dc9846f 100644
--- a/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala
+++ b/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala
@@ -67,9 +67,8 @@ class DynamicConfigChangeTest extends KafkaServerTestHarness {
val oldVal: java.lang.Long = 100000L
val newVal: java.lang.Long = 200000L
val tp = new TopicPartition("test", 0)
- val logProps = new Properties()
- logProps.put(TopicConfig.FLUSH_MESSAGES_INTERVAL_CONFIG, oldVal.toString)
- createTopic(tp.topic, 1, 1, logProps)
+ val logConfigs = util.Map.of(TopicConfig.FLUSH_MESSAGES_INTERVAL_CONFIG,
oldVal.toString)
+ createTopic(tp.topic, 1, 1, logConfigs)
TestUtils.retry(10000) {
val logOpt = this.brokers.head.logManager.getLog(tp)
assertTrue(logOpt.isPresent)
@@ -99,9 +98,8 @@ class DynamicConfigChangeTest extends KafkaServerTestHarness {
def testDynamicTopicConfigChange(): Unit = {
val tp = new TopicPartition("test", 0)
val oldSegmentSize = 2 * 1024 * 1024
- val logProps = new Properties()
- logProps.put(TopicConfig.SEGMENT_BYTES_CONFIG, oldSegmentSize.toString)
- createTopic(tp.topic, 1, 1, logProps)
+ val logConfigs = util.Map.of(TopicConfig.SEGMENT_BYTES_CONFIG,
oldSegmentSize.toString)
+ createTopic(tp.topic, 1, 1, logConfigs)
TestUtils.retry(10000) {
val logOpt = this.brokers.head.logManager.getLog(tp)
assertTrue(logOpt.isPresent)
diff --git
a/core/src/test/scala/unit/kafka/server/FetchRequestMaxBytesTest.scala
b/core/src/test/scala/unit/kafka/server/FetchRequestMaxBytesTest.scala
index 63215defd8f..b3de9ac5e62 100644
--- a/core/src/test/scala/unit/kafka/server/FetchRequestMaxBytesTest.scala
+++ b/core/src/test/scala/unit/kafka/server/FetchRequestMaxBytesTest.scala
@@ -27,6 +27,7 @@ import org.apache.kafka.server.config.ServerConfigs
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test, TestInfo}
+import java.util
import java.util.{Optional, Properties}
import scala.jdk.CollectionConverters._
@@ -77,12 +78,11 @@ class FetchRequestMaxBytesTest extends BaseRequestTest {
}
private def createTopics(): Unit = {
- val topicConfig = new Properties
- topicConfig.setProperty(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG,
1.toString)
+ val topicConfigs = util.Map.of(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG,
1.toString)
createTopic(testTopic,
numPartitions = 1,
replicationFactor = 1,
- topicConfig = topicConfig)
+ topicConfig = topicConfigs)
// Produce several messages as single batches.
messages.indices.foreach(i => {
val record = new ProducerRecord(testTopic, 0, oneByteArray(i.toByte),
messages(i))
diff --git
a/core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala
b/core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala
index fdc653e032d..2abfdfa6327 100644
---
a/core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala
+++
b/core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala
@@ -39,7 +39,7 @@ import org.junit.jupiter.api.Assertions.{assertEquals, fail}
import java.net.Socket
import java.util
-import java.util.{Comparator, Properties}
+import java.util.Comparator
import java.util.stream.Collectors
import scala.collection.Seq
import scala.collection.mutable.ListBuffer
@@ -78,7 +78,7 @@ class GroupCoordinatorBaseRequestTest(cluster:
ClusterInstance) {
replicationFactor =
brokers().head.config.getShort(TransactionLogConfig.TRANSACTIONS_TOPIC_REPLICATION_FACTOR_CONFIG).toInt,
brokers = brokers(),
controllers = controllerServers(),
- topicConfig = new Properties()
+ topicConfig = util.Map.of()
)
} finally {
admin.close()
@@ -127,7 +127,7 @@ class GroupCoordinatorBaseRequestTest(cluster:
ClusterInstance) {
topic: String,
numPartitions: Int = 1,
replicationFactor: Int = 1,
- topicConfig: Properties = new Properties
+ topicConfig: util.Map[String, String] = util.Map.of()
): Map[TopicIdPartition, Int] = {
val admin = cluster.admin()
try {
diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
index 2b1d5c775f2..aa3f91a0ee1 100644
--- a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
@@ -759,14 +759,14 @@ class KafkaApisTest extends Logging {
case CoordinatorType.GROUP =>
topicConfigOverride.put(GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG,
numBrokersNeeded.toString)
topicConfigOverride.put(GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG,
numBrokersNeeded.toString)
- when(groupCoordinator.groupMetadataTopicConfigs).thenReturn(new
Properties)
+
when(groupCoordinator.groupMetadataTopicConfigs).thenReturn(util.Map.of())
authorizeResource(authorizer, AclOperation.DESCRIBE,
ResourceType.GROUP,
groupId, AuthorizationResult.ALLOWED)
Topic.GROUP_METADATA_TOPIC_NAME
case CoordinatorType.TRANSACTION =>
topicConfigOverride.put(TransactionLogConfig.TRANSACTIONS_TOPIC_PARTITIONS_CONFIG,
numBrokersNeeded.toString)
topicConfigOverride.put(TransactionLogConfig.TRANSACTIONS_TOPIC_REPLICATION_FACTOR_CONFIG,
numBrokersNeeded.toString)
- when(txnCoordinator.transactionStateTopicConfigs).thenReturn(new
Properties)
+
when(txnCoordinator.transactionStateTopicConfigs).thenReturn(util.Map.of())
authorizeResource(authorizer, AclOperation.DESCRIBE,
ResourceType.TRANSACTIONAL_ID,
groupId, AuthorizationResult.ALLOWED)
Topic.TRANSACTION_STATE_TOPIC_NAME
@@ -932,13 +932,13 @@ class KafkaApisTest extends Logging {
case Topic.GROUP_METADATA_TOPIC_NAME =>
topicConfigOverride.put(GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG,
numBrokersNeeded.toString)
topicConfigOverride.put(GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG,
numBrokersNeeded.toString)
- when(groupCoordinator.groupMetadataTopicConfigs).thenReturn(new
Properties)
+
when(groupCoordinator.groupMetadataTopicConfigs).thenReturn(util.Map.of())
true
case Topic.TRANSACTION_STATE_TOPIC_NAME =>
topicConfigOverride.put(TransactionLogConfig.TRANSACTIONS_TOPIC_PARTITIONS_CONFIG,
numBrokersNeeded.toString)
topicConfigOverride.put(TransactionLogConfig.TRANSACTIONS_TOPIC_REPLICATION_FACTOR_CONFIG,
numBrokersNeeded.toString)
- when(txnCoordinator.transactionStateTopicConfigs).thenReturn(new
Properties)
+
when(txnCoordinator.transactionStateTopicConfigs).thenReturn(util.Map.of())
true
case _ =>
topicConfigOverride.put(ServerLogConfigs.NUM_PARTITIONS_CONFIG,
numBrokersNeeded.toString)
diff --git
a/core/src/test/scala/unit/kafka/server/ListOffsetsRequestWithRemoteStoreTest.scala
b/core/src/test/scala/unit/kafka/server/ListOffsetsRequestWithRemoteStoreTest.scala
index 970f3f7489b..61cff73217d 100644
---
a/core/src/test/scala/unit/kafka/server/ListOffsetsRequestWithRemoteStoreTest.scala
+++
b/core/src/test/scala/unit/kafka/server/ListOffsetsRequestWithRemoteStoreTest.scala
@@ -20,6 +20,7 @@ import org.apache.kafka.common.config.TopicConfig
import
org.apache.kafka.server.log.remote.storage.{NoOpRemoteLogMetadataManager,
NoOpRemoteStorageManager, RemoteLogManagerConfig}
import org.junit.jupiter.api.TestInfo
+import java.util
import java.util.Properties
import scala.collection.Seq
@@ -38,8 +39,6 @@ class ListOffsetsRequestWithRemoteStoreTest extends
ListOffsetsRequestTest {
}
override def createTopic(numPartitions: Int, replicationFactor: Int):
Map[Int, Int] = {
- val props = new Properties()
- props.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true")
- super.createTopic(topic, numPartitions, replicationFactor, props)
+ super.createTopic(topic, numPartitions, replicationFactor,
util.Map.of(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true"))
}
}
diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala
b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
index b55661b95ef..bc41c2ed2ae 100755
--- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala
+++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
@@ -317,14 +317,11 @@ object TestUtils extends Logging {
numPartitions: Int = 1,
replicationFactor: Int = 1,
replicaAssignment: collection.Map[Int, Seq[Int]] = Map.empty,
- topicConfig: Properties = new Properties,
+ topicConfig: util.Map[String, String] = util.Map.of(),
): Uuid = {
- val configsMap = new util.HashMap[String, String]()
- topicConfig.forEach((k, v) => configsMap.put(k.toString, v.toString))
-
val result = if (replicaAssignment.isEmpty) {
admin.createTopics(util.List.of(new NewTopic(
- topic, numPartitions, replicationFactor.toShort).configs(configsMap)))
+ topic, numPartitions, replicationFactor.toShort).configs(topicConfig)))
} else {
val assignment = new util.HashMap[Integer, util.List[Integer]]()
replicaAssignment.foreachEntry { case (k, v) =>
@@ -333,7 +330,7 @@ object TestUtils extends Logging {
assignment.put(k.asInstanceOf[Integer], replicas)
}
admin.createTopics(util.List.of(new NewTopic(
- topic, assignment).configs(configsMap)))
+ topic, assignment).configs(topicConfig)))
}
result.topicId(topic).get()
@@ -347,7 +344,7 @@ object TestUtils extends Logging {
numPartitions: Int = 1,
replicationFactor: Int = 1,
replicaAssignment: collection.Map[Int, Seq[Int]] = Map.empty,
- topicConfig: Properties = new Properties,
+ topicConfig: util.Map[String, String] = util.Map.of(),
): scala.collection.immutable.Map[Int, Int] = {
val effectiveNumPartitions = if (replicaAssignment.isEmpty) {
numPartitions
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java
index da82e775761..c6f67ade887 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java
@@ -59,6 +59,7 @@ import org.apache.kafka.image.MetadataImage;
import org.apache.kafka.server.authorizer.AuthorizableRequestContext;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.OptionalInt;
import java.util.Properties;
@@ -446,12 +447,11 @@ public interface GroupCoordinator {
);
/**
- * Return the configuration properties of the internal group
- * metadata topic.
+ * Returns the configuration of the internal group metadata topic.
*
- * @return Properties of the internal topic.
+ * @return The configuration of the internal group metadata topic.
*/
- Properties groupMetadataTopicConfigs();
+ Map<String, String> groupMetadataTopicConfigs();
/**
* Return the configuration of the provided group.
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
index 2911736a010..e879c7083fe 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
@@ -2325,12 +2325,12 @@ public class GroupCoordinatorService implements
GroupCoordinator {
* See {@link GroupCoordinator#groupMetadataTopicConfigs()}.
*/
@Override
- public Properties groupMetadataTopicConfigs() {
- Properties properties = new Properties();
- properties.put(TopicConfig.CLEANUP_POLICY_CONFIG,
TopicConfig.CLEANUP_POLICY_COMPACT);
- properties.put(TopicConfig.COMPRESSION_TYPE_CONFIG,
BrokerCompressionType.PRODUCER.name);
- properties.put(TopicConfig.SEGMENT_BYTES_CONFIG,
String.valueOf(config.offsetsTopicSegmentBytes()));
- return properties;
+ public Map<String, String> groupMetadataTopicConfigs() {
+ return Map.of(
+ TopicConfig.CLEANUP_POLICY_CONFIG,
TopicConfig.CLEANUP_POLICY_COMPACT,
+ TopicConfig.COMPRESSION_TYPE_CONFIG,
BrokerCompressionType.PRODUCER.name,
+ TopicConfig.SEGMENT_BYTES_CONFIG,
String.valueOf(config.offsetsTopicSegmentBytes())
+ );
}
/**
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java
index 23c12223af0..1ac1baf0553 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java
@@ -134,7 +134,6 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.OptionalInt;
-import java.util.Properties;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
@@ -976,12 +975,14 @@ public class GroupCoordinatorServiceTest {
.setRuntime(runtime)
.build();
- Properties expectedProperties = new Properties();
- expectedProperties.put(TopicConfig.CLEANUP_POLICY_CONFIG,
TopicConfig.CLEANUP_POLICY_COMPACT);
- expectedProperties.put(TopicConfig.COMPRESSION_TYPE_CONFIG,
BrokerCompressionType.PRODUCER.name);
- expectedProperties.put(TopicConfig.SEGMENT_BYTES_CONFIG, "1000");
-
- assertEquals(expectedProperties, service.groupMetadataTopicConfigs());
+ assertEquals(
+ Map.of(
+ TopicConfig.CLEANUP_POLICY_CONFIG,
TopicConfig.CLEANUP_POLICY_COMPACT,
+ TopicConfig.COMPRESSION_TYPE_CONFIG,
BrokerCompressionType.PRODUCER.name,
+ TopicConfig.SEGMENT_BYTES_CONFIG, "1000"
+ ),
+ service.groupMetadataTopicConfigs()
+ );
}
@Test
diff --git
a/server/src/main/java/org/apache/kafka/server/DefaultAutoTopicCreationManager.java
b/server/src/main/java/org/apache/kafka/server/DefaultAutoTopicCreationManager.java
index 04e754a924e..8cbdeb83be2 100644
---
a/server/src/main/java/org/apache/kafka/server/DefaultAutoTopicCreationManager.java
+++
b/server/src/main/java/org/apache/kafka/server/DefaultAutoTopicCreationManager.java
@@ -48,7 +48,6 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
-import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Supplier;
@@ -61,18 +60,18 @@ public class DefaultAutoTopicCreationManager implements
AutoTopicCreationManager
private final AbstractKafkaConfig config;
private final TopicCreator topicCreator;
- private final Supplier<Properties> groupCoordinatorConfigsSupplier;
- private final Supplier<Properties> shareCoordinatorConfigsSupplier;
- private final Supplier<Properties> transactionTopicConfigsSupplier;
+ private final Supplier<Map<String, String>>
groupCoordinatorConfigsSupplier;
+ private final Supplier<Map<String, String>>
shareCoordinatorConfigsSupplier;
+ private final Supplier<Map<String, String>>
transactionTopicConfigsSupplier;
private final Time time;
private final Set<String> inflightTopics = ConcurrentHashMap.newKeySet();
private final ExpiringErrorCache topicCreationErrorCache;
public DefaultAutoTopicCreationManager(
AbstractKafkaConfig config,
- Supplier<Properties> groupCoordinatorConfigsSupplier,
- Supplier<Properties> transactionTopicConfigsSupplier,
- Supplier<Properties> shareCoordinatorConfigsSupplier,
+ Supplier<Map<String, String>> groupCoordinatorConfigsSupplier,
+ Supplier<Map<String, String>> transactionTopicConfigsSupplier,
+ Supplier<Map<String, String>> shareCoordinatorConfigsSupplier,
TopicCreator topicCreator,
Time time
) {
@@ -91,9 +90,9 @@ public class DefaultAutoTopicCreationManager implements
AutoTopicCreationManager
// VisibleForTesting
DefaultAutoTopicCreationManager(
AbstractKafkaConfig config,
- Supplier<Properties> groupCoordinatorConfigsSupplier,
- Supplier<Properties> transactionTopicConfigsSupplier,
- Supplier<Properties> shareCoordinatorConfigsSupplier,
+ Supplier<Map<String, String>> groupCoordinatorConfigsSupplier,
+ Supplier<Map<String, String>> transactionTopicConfigsSupplier,
+ Supplier<Map<String, String>> shareCoordinatorConfigsSupplier,
TopicCreator topicCreator,
Time time,
int topicErrorCacheCapacity
@@ -276,14 +275,13 @@ public class DefaultAutoTopicCreationManager implements
AutoTopicCreationManager
};
}
- private static CreatableTopicConfigCollection
convertToTopicConfigCollections(Properties config) {
+ private static CreatableTopicConfigCollection
convertToTopicConfigCollections(Map<String, String> config) {
return new CreatableTopicConfigCollection(
config.entrySet().stream()
.map(entry -> new CreatableTopicConfig()
- .setName(entry.getKey().toString())
- .setValue(entry.getValue().toString()))
+ .setName(entry.getKey())
+ .setValue(entry.getValue()))
.toList()
- .iterator()
);
}
diff --git
a/server/src/test/java/org/apache/kafka/server/DefaultAutoTopicCreationManagerTest.java
b/server/src/test/java/org/apache/kafka/server/DefaultAutoTopicCreationManagerTest.java
index b897179d576..1954b30780e 100644
---
a/server/src/test/java/org/apache/kafka/server/DefaultAutoTopicCreationManagerTest.java
+++
b/server/src/test/java/org/apache/kafka/server/DefaultAutoTopicCreationManagerTest.java
@@ -132,9 +132,9 @@ public class DefaultAutoTopicCreationManagerTest {
) {
autoTopicCreationManager = new DefaultAutoTopicCreationManager(
config,
- Properties::new,
- Properties::new,
- Properties::new,
+ Map::of,
+ Map::of,
+ Map::of,
topicCreator,
mockTime,
testCacheCapacity);
@@ -179,9 +179,9 @@ public class DefaultAutoTopicCreationManagerTest {
public void testTopicCreationWithMetadataContext() throws
UnknownHostException {
autoTopicCreationManager = new DefaultAutoTopicCreationManager(
config,
- Properties::new,
- Properties::new,
- Properties::new,
+ Map::of,
+ Map::of,
+ Map::of,
topicCreator,
mockTime,
testCacheCapacity);
@@ -239,9 +239,9 @@ public class DefaultAutoTopicCreationManagerTest {
autoTopicCreationManager = new DefaultAutoTopicCreationManager(
config,
- Properties::new,
- Properties::new,
- Properties::new,
+ Map::of,
+ Map::of,
+ Map::of,
topicCreator,
mockTime,
testCacheCapacity);
@@ -280,9 +280,9 @@ public class DefaultAutoTopicCreationManagerTest {
autoTopicCreationManager = new DefaultAutoTopicCreationManager(
config,
- Properties::new,
- Properties::new,
- Properties::new,
+ Map::of,
+ Map::of,
+ Map::of,
topicCreator,
mockTime,
testCacheCapacity);
@@ -306,9 +306,9 @@ public class DefaultAutoTopicCreationManagerTest {
autoTopicCreationManager = new DefaultAutoTopicCreationManager(
config,
- Properties::new,
- Properties::new,
- Properties::new,
+ Map::of,
+ Map::of,
+ Map::of,
topicCreator,
mockTime,
testCacheCapacity);
@@ -333,9 +333,9 @@ public class DefaultAutoTopicCreationManagerTest {
public void testTopicCreationErrorCaching() throws UnknownHostException {
autoTopicCreationManager = new DefaultAutoTopicCreationManager(
config,
- Properties::new,
- Properties::new,
- Properties::new,
+ Map::of,
+ Map::of,
+ Map::of,
topicCreator,
mockTime,
testCacheCapacity);
@@ -367,9 +367,9 @@ public class DefaultAutoTopicCreationManagerTest {
public void testGetTopicCreationErrorsWithMultipleTopics() throws
UnknownHostException {
autoTopicCreationManager = new DefaultAutoTopicCreationManager(
config,
- Properties::new,
- Properties::new,
- Properties::new,
+ Map::of,
+ Map::of,
+ Map::of,
topicCreator,
mockTime,
testCacheCapacity);
@@ -407,9 +407,9 @@ public class DefaultAutoTopicCreationManagerTest {
public void testErrorCacheTTL() throws UnknownHostException {
autoTopicCreationManager = new DefaultAutoTopicCreationManager(
config,
- Properties::new,
- Properties::new,
- Properties::new,
+ Map::of,
+ Map::of,
+ Map::of,
topicCreator,
mockTime,
testCacheCapacity);
@@ -450,9 +450,9 @@ public class DefaultAutoTopicCreationManagerTest {
// Create manager with small cache size for testing
autoTopicCreationManager = new DefaultAutoTopicCreationManager(
config,
- Properties::new,
- Properties::new,
- Properties::new,
+ Map::of,
+ Map::of,
+ Map::of,
topicCreator,
mockTime,
3);
@@ -499,9 +499,9 @@ public class DefaultAutoTopicCreationManagerTest {
public void testTopicsInBackoffAreNotRetried() throws UnknownHostException
{
autoTopicCreationManager = new DefaultAutoTopicCreationManager(
config,
- Properties::new,
- Properties::new,
- Properties::new,
+ Map::of,
+ Map::of,
+ Map::of,
topicCreator,
mockTime,
testCacheCapacity);
@@ -539,9 +539,9 @@ public class DefaultAutoTopicCreationManagerTest {
public void testTopicsOutOfBackoffCanBeRetried() throws
UnknownHostException {
autoTopicCreationManager = new DefaultAutoTopicCreationManager(
config,
- Properties::new,
- Properties::new,
- Properties::new,
+ Map::of,
+ Map::of,
+ Map::of,
topicCreator,
mockTime,
testCacheCapacity);
@@ -586,9 +586,9 @@ public class DefaultAutoTopicCreationManagerTest {
public void testInflightTopicsAreNotRetriedConcurrently() throws
UnknownHostException {
autoTopicCreationManager = new DefaultAutoTopicCreationManager(
config,
- Properties::new,
- Properties::new,
- Properties::new,
+ Map::of,
+ Map::of,
+ Map::of,
topicCreator,
mockTime,
testCacheCapacity);
@@ -620,9 +620,9 @@ public class DefaultAutoTopicCreationManagerTest {
public void testBackoffAndInflightInteraction() throws
UnknownHostException {
autoTopicCreationManager = new DefaultAutoTopicCreationManager(
config,
- Properties::new,
- Properties::new,
- Properties::new,
+ Map::of,
+ Map::of,
+ Map::of,
topicCreator,
mockTime,
testCacheCapacity);
diff --git
a/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinator.java
b/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinator.java
index 4b07209f26e..9c312e924a4 100644
---
a/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinator.java
+++
b/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinator.java
@@ -32,8 +32,8 @@ import org.apache.kafka.image.MetadataDelta;
import org.apache.kafka.image.MetadataImage;
import org.apache.kafka.server.share.SharePartitionKey;
+import java.util.Map;
import java.util.OptionalInt;
-import java.util.Properties;
import java.util.concurrent.CompletableFuture;
import java.util.function.IntSupplier;
@@ -47,11 +47,11 @@ public interface ShareCoordinator {
int partitionFor(SharePartitionKey key);
/**
- * Return the configuration properties of the share-group state topic.
+ * Return the configuration of the share-group state topic.
*
- * @return Properties of the share-group state topic.
+ * @return The configuration of the share-group state topic.
*/
- Properties shareGroupStateTopicConfigs();
+ Map<String, String> shareGroupStateTopicConfigs();
/**
* Start the share coordinator
diff --git
a/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java
b/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java
index bc6632f0adb..702d27b857e 100644
---
a/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java
+++
b/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java
@@ -71,7 +71,6 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.OptionalInt;
-import java.util.Properties;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
@@ -286,15 +285,15 @@ public class ShareCoordinatorService implements
ShareCoordinator {
}
@Override
- public Properties shareGroupStateTopicConfigs() {
- Properties properties = new Properties();
+ public Map<String, String> shareGroupStateTopicConfigs() {
// As defined in KIP-932.
- properties.put(TopicConfig.CLEANUP_POLICY_CONFIG,
TopicConfig.CLEANUP_POLICY_DELETE);
- properties.put(TopicConfig.COMPRESSION_TYPE_CONFIG,
BrokerCompressionType.PRODUCER.name);
- properties.put(TopicConfig.SEGMENT_BYTES_CONFIG,
config.shareCoordinatorStateTopicSegmentBytes());
- properties.put(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG,
config.shareCoordinatorStateTopicMinIsr());
- properties.put(TopicConfig.RETENTION_MS_CONFIG, -1);
- return properties;
+ return Map.of(
+ TopicConfig.CLEANUP_POLICY_CONFIG,
TopicConfig.CLEANUP_POLICY_DELETE,
+ TopicConfig.COMPRESSION_TYPE_CONFIG,
BrokerCompressionType.PRODUCER.name,
+ TopicConfig.SEGMENT_BYTES_CONFIG,
String.valueOf(config.shareCoordinatorStateTopicSegmentBytes()),
+ TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG,
String.valueOf(config.shareCoordinatorStateTopicMinIsr()),
+ TopicConfig.RETENTION_MS_CONFIG, "-1"
+ );
}
/**
diff --git
a/share-coordinator/src/test/java/org/apache/kafka/coordinator/share/ShareCoordinatorServiceTest.java
b/share-coordinator/src/test/java/org/apache/kafka/coordinator/share/ShareCoordinatorServiceTest.java
index 8800fbd233a..4479d76d515 100644
---
a/share-coordinator/src/test/java/org/apache/kafka/coordinator/share/ShareCoordinatorServiceTest.java
+++
b/share-coordinator/src/test/java/org/apache/kafka/coordinator/share/ShareCoordinatorServiceTest.java
@@ -62,7 +62,6 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
-import java.util.Properties;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
@@ -2085,16 +2084,14 @@ class ShareCoordinatorServiceTest {
writer
));
- List<String> propNames = List.of(
+ Set<String> propNames = Set.of(
TopicConfig.CLEANUP_POLICY_CONFIG,
TopicConfig.COMPRESSION_TYPE_CONFIG,
TopicConfig.SEGMENT_BYTES_CONFIG,
TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG,
TopicConfig.RETENTION_MS_CONFIG
);
- Properties actual = service.shareGroupStateTopicConfigs();
- propNames.forEach(actual::remove);
- assertTrue(actual.isEmpty());
+ assertEquals(propNames,
service.shareGroupStateTopicConfigs().keySet());
service.shutdown();
}
