Skip to content

Commit c9fcad7

Browse files
sjhajhariasjhajharia
authored
MINOR: Cleanup Core Module- Scala Modules (1/n) (#19380)
Now that Kafka Brokers support Java 17, this PR makes some changes in core module. The changes in this PR are limited to only the Scala files in the Core module's tests. The unit tests module is still pending. It shall follow next. The changes mostly include: - Collections.emptyList(), Collections.singletonList() and Arrays.asList() are replaced with List.of() - Collections.emptyMap() and Collections.singletonMap() are replaced with Map.of() - Collections.singleton() is replaced with Set.of() To be clear, the directories being targeted in this PR are: - core/src/test/scala/kafka - core/src/test/scala/integration/kafka Reviewers: Ken Huang <[email protected]>, Chia-Ping Tsai <[email protected]>
1 parent 95d31be commit c9fcad7

File tree

49 files changed

+1018
-1049
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

49 files changed

+1018
-1049
lines changed

‎core/src/test/scala/integration/kafka/admin/RemoteTopicCrudTest.scala

Lines changed: 16 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ import org.junit.jupiter.params.provider.CsvSource
3434

3535
import java.util
3636
import java.util.concurrent.atomic.AtomicInteger
37-
import java.util.{Collections, Optional, Properties}
37+
import java.util.{Optional, Properties}
3838
import scala.collection.Seq
3939
import scala.concurrent.ExecutionException
4040
import scala.util.Random
@@ -195,7 +195,7 @@ class RemoteTopicCrudTest extends IntegrationTestHarness {
195195
topicConfig = topicConfig)
196196

197197
// 4. create a topic with `remote.log.copy.disable=false` and have different local.retention.ms and retention.ms value,
198-
// it should successfully creates the topic.
198+
// it should successfully create the topic.
199199
topicConfig.clear()
200200
topicConfig.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true")
201201
topicConfig.put(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, "100")
@@ -207,7 +207,7 @@ class RemoteTopicCrudTest extends IntegrationTestHarness {
207207
// 5. alter the config to `remote.log.copy.disable=true`, it should fail the config change
208208
val configs = new util.HashMap[ConfigResource, util.Collection[AlterConfigOp]]()
209209
configs.put(new ConfigResource(ConfigResource.Type.TOPIC, testTopicName3),
210-
util.Arrays.asList(
210+
util.List.of(
211211
new AlterConfigOp(new ConfigEntry(TopicConfig.REMOTE_LOG_COPY_DISABLE_CONFIG, "true"),
212212
AlterConfigOp.OpType.SET),
213213
))
@@ -217,7 +217,7 @@ class RemoteTopicCrudTest extends IntegrationTestHarness {
217217

218218
// 6. alter the config to `remote.log.copy.disable=true` and local.retention.ms == retention.ms, it should work without error
219219
configs.put(new ConfigResource(ConfigResource.Type.TOPIC, testTopicName3),
220-
util.Arrays.asList(
220+
util.List.of(
221221
new AlterConfigOp(new ConfigEntry(TopicConfig.REMOTE_LOG_COPY_DISABLE_CONFIG, "true"),
222222
AlterConfigOp.OpType.SET),
223223
new AlterConfigOp(new ConfigEntry(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, "1000"),
@@ -260,7 +260,7 @@ class RemoteTopicCrudTest extends IntegrationTestHarness {
260260
topicConfig = topicConfig)
261261

262262
// 4. create a topic with `remote.log.copy.disable=false` and have different local.retention.bytes and retention.bytes value,
263-
// it should successfully creates the topic.
263+
// it should successfully create the topic.
264264
topicConfig.clear()
265265
topicConfig.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true")
266266
topicConfig.put(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, "100")
@@ -272,7 +272,7 @@ class RemoteTopicCrudTest extends IntegrationTestHarness {
272272
// 5. alter the config to `remote.log.copy.disable=true`, it should fail the config change
273273
val configs = new util.HashMap[ConfigResource, util.Collection[AlterConfigOp]]()
274274
configs.put(new ConfigResource(ConfigResource.Type.TOPIC, testTopicName3),
275-
util.Arrays.asList(
275+
util.List.of(
276276
new AlterConfigOp(new ConfigEntry(TopicConfig.REMOTE_LOG_COPY_DISABLE_CONFIG, "true"),
277277
AlterConfigOp.OpType.SET),
278278
))
@@ -282,7 +282,7 @@ class RemoteTopicCrudTest extends IntegrationTestHarness {
282282

283283
// 6. alter the config to `remote.log.copy.disable=true` and local.retention.bytes == retention.bytes, it should work without error
284284
configs.put(new ConfigResource(ConfigResource.Type.TOPIC, testTopicName3),
285-
util.Arrays.asList(
285+
util.List.of(
286286
new AlterConfigOp(new ConfigEntry(TopicConfig.REMOTE_LOG_COPY_DISABLE_CONFIG, "true"),
287287
AlterConfigOp.OpType.SET),
288288
new AlterConfigOp(new ConfigEntry(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, "1000"),
@@ -300,7 +300,7 @@ class RemoteTopicCrudTest extends IntegrationTestHarness {
300300

301301
val configs = new util.HashMap[ConfigResource, util.Collection[AlterConfigOp]]()
302302
configs.put(new ConfigResource(ConfigResource.Type.TOPIC, testTopicName),
303-
Collections.singleton(
303+
util.Set.of(
304304
new AlterConfigOp(new ConfigEntry(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true"),
305305
AlterConfigOp.OpType.SET))
306306
)
@@ -322,7 +322,7 @@ class RemoteTopicCrudTest extends IntegrationTestHarness {
322322
TestUtils.createTopicWithAdmin(admin, testTopicName, brokers, controllerServers, numPartitions, numReplicationFactor)
323323
val configs = new util.HashMap[ConfigResource, util.Collection[AlterConfigOp]]()
324324
configs.put(new ConfigResource(ConfigResource.Type.TOPIC, testTopicName),
325-
Collections.singleton(
325+
util.Set.of(
326326
new AlterConfigOp(new ConfigEntry(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true"),
327327
AlterConfigOp.OpType.SET))
328328
)
@@ -341,7 +341,7 @@ class RemoteTopicCrudTest extends IntegrationTestHarness {
341341

342342
val configs = new util.HashMap[ConfigResource, util.Collection[AlterConfigOp]]()
343343
configs.put(new ConfigResource(ConfigResource.Type.TOPIC, testTopicName),
344-
util.Arrays.asList(
344+
util.List.of(
345345
new AlterConfigOp(new ConfigEntry(TopicConfig.RETENTION_MS_CONFIG, "200"),
346346
AlterConfigOp.OpType.SET),
347347
new AlterConfigOp(new ConfigEntry(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, "100"),
@@ -361,7 +361,7 @@ class RemoteTopicCrudTest extends IntegrationTestHarness {
361361

362362
val configs = new util.HashMap[ConfigResource, util.Collection[AlterConfigOp]]()
363363
configs.put(new ConfigResource(ConfigResource.Type.TOPIC, testTopicName),
364-
util.Arrays.asList(
364+
util.List.of(
365365
new AlterConfigOp(new ConfigEntry(TopicConfig.RETENTION_BYTES_CONFIG, "200"),
366366
AlterConfigOp.OpType.SET),
367367
new AlterConfigOp(new ConfigEntry(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, "100"),
@@ -382,7 +382,7 @@ class RemoteTopicCrudTest extends IntegrationTestHarness {
382382
// inherited local retention ms is 1000
383383
val configs = new util.HashMap[ConfigResource, util.Collection[AlterConfigOp]]()
384384
configs.put(new ConfigResource(ConfigResource.Type.TOPIC, testTopicName),
385-
util.Arrays.asList(
385+
util.List.of(
386386
new AlterConfigOp(new ConfigEntry(TopicConfig.RETENTION_MS_CONFIG, "200"),
387387
AlterConfigOp.OpType.SET),
388388
))
@@ -401,7 +401,7 @@ class RemoteTopicCrudTest extends IntegrationTestHarness {
401401
// inherited local retention bytes is 1024
402402
val configs = new util.HashMap[ConfigResource, util.Collection[AlterConfigOp]]()
403403
configs.put(new ConfigResource(ConfigResource.Type.TOPIC, testTopicName),
404-
util.Arrays.asList(
404+
util.List.of(
405405
new AlterConfigOp(new ConfigEntry(TopicConfig.RETENTION_BYTES_CONFIG, "512"),
406406
AlterConfigOp.OpType.SET),
407407
))
@@ -420,7 +420,7 @@ class RemoteTopicCrudTest extends IntegrationTestHarness {
420420

421421
val configs = new util.HashMap[ConfigResource, util.Collection[AlterConfigOp]]()
422422
configs.put(new ConfigResource(ConfigResource.Type.TOPIC, testTopicName),
423-
util.Arrays.asList(
423+
util.List.of(
424424
new AlterConfigOp(new ConfigEntry(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "false"),
425425
AlterConfigOp.OpType.SET),
426426
))
@@ -440,7 +440,7 @@ class RemoteTopicCrudTest extends IntegrationTestHarness {
440440

441441
val configs = new util.HashMap[ConfigResource, util.Collection[AlterConfigOp]]()
442442
configs.put(new ConfigResource(ConfigResource.Type.TOPIC, testTopicName),
443-
util.Arrays.asList(
443+
util.List.of(
444444
new AlterConfigOp(new ConfigEntry(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "false"),
445445
AlterConfigOp.OpType.SET),
446446
new AlterConfigOp(new ConfigEntry(TopicConfig.REMOTE_LOG_DELETE_ON_DISABLE_CONFIG, "true"),
@@ -598,7 +598,7 @@ class MyRemoteLogMetadataManager extends NoOpRemoteLogMetadataManager {
598598
val timestamp = time.milliseconds()
599599
val startOffset = idx * recordsPerSegment
600600
val endOffset = startOffset + recordsPerSegment - 1
601-
val segmentLeaderEpochs: util.Map[Integer, java.lang.Long] = Collections.singletonMap(0, 0L)
601+
val segmentLeaderEpochs: util.Map[Integer, java.lang.Long] = util.Map.of(0, 0L)
602602
segmentMetadataList.add(new RemoteLogSegmentMetadata(new RemoteLogSegmentId(topicIdPartition, Uuid.randomUuid()), startOffset, endOffset, timestamp, 0, timestamp, segmentSize, Optional.empty(), RemoteLogSegmentState.COPY_SEGMENT_FINISHED, segmentLeaderEpochs))
603603
}
604604
segmentMetadataList.iterator()

‎core/src/test/scala/integration/kafka/api/AdminClientWithPoliciesIntegrationTest.scala

Lines changed: 18 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@
1414
package kafka.api
1515

1616
import java.util
17-
import java.util.{Collections, Properties}
17+
import java.util.Properties
1818
import kafka.integration.KafkaServerTestHarness
1919
import kafka.server.KafkaConfig
2020
import kafka.utils.{Logging, TestUtils}
@@ -59,7 +59,7 @@ class AdminClientWithPoliciesIntegrationTest extends KafkaServerTestHarness with
5959
}
6060

6161
def createConfig: util.Map[String, Object] =
62-
Map[String, Object](AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG -> bootstrapServers()).asJava
62+
util.Map.of[String, Object](AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers())
6363

6464
override def generateConfigs: collection.Seq[KafkaConfig] = {
6565
val configs = TestUtils.createBrokerConfigs(brokerCount)
@@ -122,34 +122,34 @@ class AdminClientWithPoliciesIntegrationTest extends KafkaServerTestHarness with
122122

123123
// Set a mutable broker config
124124
val brokerResource = new ConfigResource(ConfigResource.Type.BROKER, brokers.head.config.brokerId.toString)
125-
var alterResult = client.incrementalAlterConfigs(Collections.singletonMap(brokerResource,
126-
util.Arrays.asList(new AlterConfigOp(new ConfigEntry(ServerConfigs.MESSAGE_MAX_BYTES_CONFIG, "50000"), OpType.SET))))
125+
var alterResult = client.incrementalAlterConfigs(util.Map.of(brokerResource,
126+
util.List.of(new AlterConfigOp(new ConfigEntry(ServerConfigs.MESSAGE_MAX_BYTES_CONFIG, "50000"), OpType.SET))))
127127
alterResult.all.get
128128
assertEquals(Set(ServerConfigs.MESSAGE_MAX_BYTES_CONFIG), validationsForResource(brokerResource).head.configs().keySet().asScala)
129129
validations.clear()
130130

131131
val alterConfigs = new util.HashMap[ConfigResource, util.Collection[AlterConfigOp]]()
132-
alterConfigs.put(topicResource1, util.Arrays.asList(
132+
alterConfigs.put(topicResource1, util.List.of(
133133
new AlterConfigOp(new ConfigEntry(TopicConfig.MIN_CLEANABLE_DIRTY_RATIO_CONFIG, "0.9"), OpType.SET),
134134
new AlterConfigOp(new ConfigEntry(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "2"), OpType.SET)
135135
))
136136

137-
alterConfigs.put(topicResource2, util.Arrays.asList(
137+
alterConfigs.put(topicResource2, util.List.of(
138138
new AlterConfigOp(new ConfigEntry(TopicConfig.MIN_CLEANABLE_DIRTY_RATIO_CONFIG, "0.8"), OpType.SET),
139139
))
140140

141-
alterConfigs.put(topicResource3, util.Arrays.asList(
141+
alterConfigs.put(topicResource3, util.List.of(
142142
new AlterConfigOp(new ConfigEntry(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "-1"), OpType.SET),
143143
))
144144

145-
alterConfigs.put(brokerResource, util.Arrays.asList(
145+
alterConfigs.put(brokerResource, util.List.of(
146146
new AlterConfigOp(new ConfigEntry(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "12313"), OpType.SET),
147147
))
148148

149149
// Alter configs: second is valid, the others are invalid
150150
alterResult = client.incrementalAlterConfigs(alterConfigs)
151151

152-
assertEquals(Set(topicResource1, topicResource2, topicResource3, brokerResource).asJava, alterResult.values.keySet)
152+
assertEquals(util.Set.of(topicResource1, topicResource2, topicResource3, brokerResource), alterResult.values.keySet)
153153
assertFutureThrows(classOf[PolicyViolationException], alterResult.values.get(topicResource1))
154154
alterResult.values.get(topicResource2).get
155155
assertFutureThrows(classOf[InvalidConfigurationException], alterResult.values.get(topicResource3))
@@ -160,7 +160,7 @@ class AdminClientWithPoliciesIntegrationTest extends KafkaServerTestHarness with
160160

161161
// Verify that the second resource was updated and the others were not
162162
ensureConsistentKRaftMetadata()
163-
var describeResult = client.describeConfigs(Seq(topicResource1, topicResource2, topicResource3, brokerResource).asJava)
163+
var describeResult = client.describeConfigs(util.List.of(topicResource1, topicResource2, topicResource3, brokerResource))
164164
var configs = describeResult.all.get
165165
assertEquals(4, configs.size)
166166

@@ -172,13 +172,13 @@ class AdminClientWithPoliciesIntegrationTest extends KafkaServerTestHarness with
172172
assertNull(configs.get(brokerResource).get(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG).value)
173173

174174
// Alter configs with validateOnly = true: only second is valid
175-
alterConfigs.put(topicResource2, util.Arrays.asList(
175+
alterConfigs.put(topicResource2, util.List.of(
176176
new AlterConfigOp(new ConfigEntry(TopicConfig.MIN_CLEANABLE_DIRTY_RATIO_CONFIG, "0.7"), OpType.SET),
177177
))
178178

179179
alterResult = client.incrementalAlterConfigs(alterConfigs, new AlterConfigsOptions().validateOnly(true))
180180

181-
assertEquals(Set(topicResource1, topicResource2, topicResource3, brokerResource).asJava, alterResult.values.keySet)
181+
assertEquals(util.Set.of(topicResource1, topicResource2, topicResource3, brokerResource), alterResult.values.keySet)
182182
assertFutureThrows(classOf[PolicyViolationException], alterResult.values.get(topicResource1))
183183
alterResult.values.get(topicResource2).get
184184
assertFutureThrows(classOf[InvalidConfigurationException], alterResult.values.get(topicResource3))
@@ -189,7 +189,7 @@ class AdminClientWithPoliciesIntegrationTest extends KafkaServerTestHarness with
189189

190190
// Verify that no resources are updated since validate_only = true
191191
ensureConsistentKRaftMetadata()
192-
describeResult = client.describeConfigs(Seq(topicResource1, topicResource2, topicResource3, brokerResource).asJava)
192+
describeResult = client.describeConfigs(util.List.of(topicResource1, topicResource2, topicResource3, brokerResource))
193193
configs = describeResult.all.get
194194
assertEquals(4, configs.size)
195195

@@ -201,12 +201,12 @@ class AdminClientWithPoliciesIntegrationTest extends KafkaServerTestHarness with
201201
assertNull(configs.get(brokerResource).get(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG).value)
202202

203203
// Do an incremental alter config on the broker, ensure we don't see the broker config we set earlier in the policy
204-
alterResult = client.incrementalAlterConfigs(Map(
205-
brokerResource ->
206-
Seq(new AlterConfigOp(
204+
alterResult = client.incrementalAlterConfigs(util.Map.of(
205+
brokerResource ,
206+
util.List.of(new AlterConfigOp(
207207
new ConfigEntry(SocketServerConfigs.MAX_CONNECTIONS_CONFIG, "9999"), OpType.SET)
208-
).asJavaCollection
209-
).asJava)
208+
)
209+
))
210210
alterResult.all.get
211211
assertEquals(Set(SocketServerConfigs.MAX_CONNECTIONS_CONFIG), validationsForResource(brokerResource).head.configs().keySet().asScala)
212212
}

0 commit comments

Comments
 (0)