Skip to content

Commit f74f596

Browse files
frankvickyfrankvicky
authored
KAFKA-16640 Replace TestUtils#resource by scala.util.Using (apache#15881)
Reviewers: Chia-Ping Tsai <[email protected]>
1 parent f7b242f commit f74f596

13 files changed

+79
-77
lines changed

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

+2-2
Original file line numberDiff line numberDiff line change
@@ -105,8 +105,8 @@ class DelegationTokenEndToEndAuthorizationWithOwnerTest extends DelegationTokenE
105105
@ParameterizedTest
106106
@ValueSource(strings = Array("kraft", "zk"))
107107
def testDescribeTokenForOtherUserFails(quorum: String): Unit = {
108-
TestUtils.resource(createScramAdminClient(kafkaClientSaslMechanism, describeTokenFailPrincipal.getName, describeTokenFailPassword)) { describeTokenFailAdminClient =>
109-
TestUtils.resource(createScramAdminClient(kafkaClientSaslMechanism, otherClientPrincipal.getName, otherClientPassword)) { otherClientAdminClient =>
108+
Using(createScramAdminClient(kafkaClientSaslMechanism, describeTokenFailPrincipal.getName, describeTokenFailPassword)) { describeTokenFailAdminClient =>
109+
Using(createScramAdminClient(kafkaClientSaslMechanism, otherClientPrincipal.getName, otherClientPassword)) { otherClientAdminClient =>
110110
otherClientAdminClient.createDelegationToken().delegationToken().get()
111111
val tokens = describeTokenFailAdminClient.describeDelegationToken(
112112
new DescribeDelegationTokenOptions().owners(Collections.singletonList(otherClientPrincipal))

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

+3-1
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,8 @@ import org.apache.kafka.common.utils.Time
3838
import org.apache.kafka.server.config.{ConfigType, KafkaSecurityConfigs}
3939
import org.apache.zookeeper.client.ZKClientConfig
4040

41+
import scala.util.Using
42+
4143
/*
4244
* Implements an enumeration for the modes enabled here:
4345
* zk only, kafka only, both, custom KafkaServer.
@@ -199,7 +201,7 @@ trait SaslSetup {
199201

200202
def createScramCredentials(zkConnect: String, userName: String, password: String): Unit = {
201203
val zkClientConfig = new ZKClientConfig()
202-
TestUtils.resource(KafkaZkClient(
204+
Using(KafkaZkClient(
203205
zkConnect, JaasUtils.isZkSaslEnabled || KafkaConfig.zkTlsClientAuthEnabled(zkClientConfig), 30000, 30000,
204206
Int.MaxValue, Time.SYSTEM, name = "SaslSetup", zkClientConfig = zkClientConfig, enableEntityConfigControllerCheck = false)) { zkClient =>
205207
val adminZkClient = new AdminZkClient(zkClient)

‎core/src/test/scala/integration/kafka/server/RaftClusterSnapshotTest.scala

+3-2
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@ import org.junit.jupiter.api.Test
3131
import org.junit.jupiter.api.Timeout
3232

3333
import scala.jdk.CollectionConverters._
34+
import scala.util.Using
3435

3536
@Timeout(120)
3637
class RaftClusterSnapshotTest {
@@ -40,7 +41,7 @@ class RaftClusterSnapshotTest {
4041
val numberOfBrokers = 3
4142
val numberOfControllers = 3
4243

43-
TestUtils.resource(
44+
Using(
4445
new KafkaClusterTestKit
4546
.Builder(
4647
new TestKitNodes.Builder()
@@ -73,7 +74,7 @@ class RaftClusterSnapshotTest {
7374

7475
// For every controller and broker perform some sanity checks against the latest snapshot
7576
for ((_, raftManager) <- cluster.raftManagers().asScala) {
76-
TestUtils.resource(
77+
Using(
7778
RecordsSnapshotReader.of(
7879
raftManager.replicatedLog.latestSnapshot.get(),
7980
new MetadataRecordSerde(),

‎core/src/test/scala/kafka/raft/KafkaMetadataLogTest.scala

+32-31
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@ import java.nio.file.{Files, Path}
4141
import java.util
4242
import java.util.{Collections, Optional, Properties}
4343
import scala.jdk.CollectionConverters._
44+
import scala.util.Using
4445

4546
final class KafkaMetadataLogTest {
4647
import KafkaMetadataLogTest._
@@ -123,7 +124,7 @@ final class KafkaMetadataLogTest {
123124
append(log, numberOfRecords, epoch)
124125
log.updateHighWatermark(new LogOffsetMetadata(numberOfRecords))
125126

126-
TestUtils.resource(log.createNewSnapshot(snapshotId).get()) { snapshot =>
127+
Using(log.createNewSnapshot(snapshotId).get()) { snapshot =>
127128
snapshot.freeze()
128129
}
129130

@@ -191,7 +192,7 @@ final class KafkaMetadataLogTest {
191192

192193
append(log, numberOfRecords, epoch)
193194
log.updateHighWatermark(new LogOffsetMetadata(numberOfRecords))
194-
TestUtils.resource(log.createNewSnapshot(snapshotId).get()) { snapshot =>
195+
Using(log.createNewSnapshot(snapshotId).get()) { snapshot =>
195196
snapshot.freeze()
196197
}
197198

@@ -227,7 +228,7 @@ final class KafkaMetadataLogTest {
227228
append(log, numberOfRecords, epoch)
228229
log.updateHighWatermark(new LogOffsetMetadata(numberOfRecords))
229230

230-
TestUtils.resource(log.createNewSnapshot(snapshotId).get()) { snapshot =>
231+
Using(log.createNewSnapshot(snapshotId).get()) { snapshot =>
231232
snapshot.freeze()
232233
}
233234

@@ -276,7 +277,7 @@ final class KafkaMetadataLogTest {
276277
append(log, numberOfRecords, epoch)
277278
log.updateHighWatermark(new LogOffsetMetadata(numberOfRecords))
278279

279-
TestUtils.resource(log.createNewSnapshot(snapshotId).get()) { snapshot =>
280+
Using(log.createNewSnapshot(snapshotId).get()) { snapshot =>
280281
snapshot.freeze()
281282
}
282283

@@ -323,7 +324,7 @@ final class KafkaMetadataLogTest {
323324

324325
append(log, numberOfRecords, epoch)
325326

326-
TestUtils.resource(log.createNewSnapshotUnchecked(sameEpochSnapshotId).get()) { snapshot =>
327+
Using(log.createNewSnapshotUnchecked(sameEpochSnapshotId).get()) { snapshot =>
327328
snapshot.freeze()
328329
}
329330

@@ -337,7 +338,7 @@ final class KafkaMetadataLogTest {
337338

338339
append(log, numberOfRecords, epoch)
339340

340-
TestUtils.resource(log.createNewSnapshotUnchecked(greaterEpochSnapshotId).get()) { snapshot =>
341+
Using(log.createNewSnapshotUnchecked(greaterEpochSnapshotId).get()) { snapshot =>
341342
snapshot.freeze()
342343
}
343344

@@ -356,25 +357,25 @@ final class KafkaMetadataLogTest {
356357

357358
append(log, 1, epoch - 1)
358359
val oldSnapshotId1 = new OffsetAndEpoch(1, epoch - 1)
359-
TestUtils.resource(log.createNewSnapshotUnchecked(oldSnapshotId1).get()) { snapshot =>
360+
Using(log.createNewSnapshotUnchecked(oldSnapshotId1).get()) { snapshot =>
360361
snapshot.freeze()
361362
}
362363

363364
append(log, 1, epoch)
364365
val oldSnapshotId2 = new OffsetAndEpoch(2, epoch)
365-
TestUtils.resource(log.createNewSnapshotUnchecked(oldSnapshotId2).get()) { snapshot =>
366+
Using(log.createNewSnapshotUnchecked(oldSnapshotId2).get()) { snapshot =>
366367
snapshot.freeze()
367368
}
368369

369370
append(log, numberOfRecords - 2, epoch)
370371
val oldSnapshotId3 = new OffsetAndEpoch(numberOfRecords, epoch)
371-
TestUtils.resource(log.createNewSnapshotUnchecked(oldSnapshotId3).get()) { snapshot =>
372+
Using(log.createNewSnapshotUnchecked(oldSnapshotId3).get()) { snapshot =>
372373
snapshot.freeze()
373374
}
374375

375376
val greaterSnapshotId = new OffsetAndEpoch(3 * numberOfRecords, epoch)
376377
append(log, numberOfRecords, epoch)
377-
TestUtils.resource(log.createNewSnapshotUnchecked(greaterSnapshotId).get()) { snapshot =>
378+
Using(log.createNewSnapshotUnchecked(greaterSnapshotId).get()) { snapshot =>
378379
snapshot.freeze()
379380
}
380381

@@ -467,7 +468,7 @@ final class KafkaMetadataLogTest {
467468
metadataDir: File,
468469
snapshotId: OffsetAndEpoch
469470
): Unit = {
470-
TestUtils.resource(FileRawSnapshotWriter.create(metadataDir.toPath, snapshotId))(_.freeze())
471+
Using(FileRawSnapshotWriter.create(metadataDir.toPath, snapshotId))(_.freeze())
471472
}
472473

473474
@Test
@@ -479,7 +480,7 @@ final class KafkaMetadataLogTest {
479480
append(log, numberOfRecords, epoch)
480481

481482
val olderEpochSnapshotId = new OffsetAndEpoch(numberOfRecords, epoch - 1)
482-
TestUtils.resource(log.createNewSnapshotUnchecked(olderEpochSnapshotId).get()) { snapshot =>
483+
Using(log.createNewSnapshotUnchecked(olderEpochSnapshotId).get()) { snapshot =>
483484
snapshot.freeze()
484485
}
485486

@@ -488,7 +489,7 @@ final class KafkaMetadataLogTest {
488489
append(log, numberOfRecords, epoch)
489490

490491
val olderOffsetSnapshotId = new OffsetAndEpoch(numberOfRecords, epoch)
491-
TestUtils.resource(log.createNewSnapshotUnchecked(olderOffsetSnapshotId).get()) { snapshot =>
492+
Using(log.createNewSnapshotUnchecked(olderOffsetSnapshotId).get()) { snapshot =>
492493
snapshot.freeze()
493494
}
494495

@@ -503,7 +504,7 @@ final class KafkaMetadataLogTest {
503504
val snapshotId = new OffsetAndEpoch(1, epoch)
504505

505506
append(log, numberOfRecords, epoch)
506-
TestUtils.resource(log.createNewSnapshotUnchecked(snapshotId).get()) { snapshot =>
507+
Using(log.createNewSnapshotUnchecked(snapshotId).get()) { snapshot =>
507508
snapshot.freeze()
508509
}
509510

@@ -540,25 +541,25 @@ final class KafkaMetadataLogTest {
540541

541542
append(log, 1, epoch - 1)
542543
val oldSnapshotId1 = new OffsetAndEpoch(1, epoch - 1)
543-
TestUtils.resource(log.createNewSnapshotUnchecked(oldSnapshotId1).get()) { snapshot =>
544+
Using(log.createNewSnapshotUnchecked(oldSnapshotId1).get()) { snapshot =>
544545
snapshot.freeze()
545546
}
546547

547548
append(log, 1, epoch)
548549
val oldSnapshotId2 = new OffsetAndEpoch(2, epoch)
549-
TestUtils.resource(log.createNewSnapshotUnchecked(oldSnapshotId2).get()) { snapshot =>
550+
Using(log.createNewSnapshotUnchecked(oldSnapshotId2).get()) { snapshot =>
550551
snapshot.freeze()
551552
}
552553

553554
append(log, numberOfRecords - 2, epoch)
554555
val oldSnapshotId3 = new OffsetAndEpoch(numberOfRecords, epoch)
555-
TestUtils.resource(log.createNewSnapshotUnchecked(oldSnapshotId3).get()) { snapshot =>
556+
Using(log.createNewSnapshotUnchecked(oldSnapshotId3).get()) { snapshot =>
556557
snapshot.freeze()
557558
}
558559

559560
val greaterSnapshotId = new OffsetAndEpoch(3 * numberOfRecords, epoch)
560561
append(log, numberOfRecords, epoch)
561-
TestUtils.resource(log.createNewSnapshotUnchecked(greaterSnapshotId).get()) { snapshot =>
562+
Using(log.createNewSnapshotUnchecked(greaterSnapshotId).get()) { snapshot =>
562563
snapshot.freeze()
563564
}
564565

@@ -589,7 +590,7 @@ final class KafkaMetadataLogTest {
589590
val snapshotId = new OffsetAndEpoch(numberOfRecords + 1, epoch + 1)
590591

591592
append(log, numberOfRecords, epoch)
592-
TestUtils.resource(log.createNewSnapshotUnchecked(snapshotId).get()) { snapshot =>
593+
Using(log.createNewSnapshotUnchecked(snapshotId).get()) { snapshot =>
593594
snapshot.freeze()
594595
}
595596

@@ -688,7 +689,7 @@ final class KafkaMetadataLogTest {
688689
log.updateHighWatermark(new LogOffsetMetadata(numberOfRecords))
689690

690691
val snapshotId = new OffsetAndEpoch(numberOfRecords, epoch)
691-
TestUtils.resource(log.createNewSnapshot(snapshotId).get()) { snapshot =>
692+
Using(log.createNewSnapshot(snapshotId).get()) { snapshot =>
692693
snapshot.freeze()
693694
}
694695

@@ -708,7 +709,7 @@ final class KafkaMetadataLogTest {
708709
log.updateHighWatermark(new LogOffsetMetadata(offset))
709710

710711
val snapshotId = new OffsetAndEpoch(offset, epoch)
711-
TestUtils.resource(log.createNewSnapshot(snapshotId).get()) { snapshot =>
712+
Using(log.createNewSnapshot(snapshotId).get()) { snapshot =>
712713
snapshot.freeze()
713714
}
714715
// Simulate log cleaning advancing the LSO
@@ -730,7 +731,7 @@ final class KafkaMetadataLogTest {
730731
log.updateHighWatermark(new LogOffsetMetadata(offset))
731732

732733
val snapshotId = new OffsetAndEpoch(offset, epoch)
733-
TestUtils.resource(log.createNewSnapshot(snapshotId).get()) { snapshot =>
734+
Using(log.createNewSnapshot(snapshotId).get()) { snapshot =>
734735
snapshot.freeze()
735736
}
736737

@@ -747,7 +748,7 @@ final class KafkaMetadataLogTest {
747748
val log = buildMetadataLog(tempDir, mockTime)
748749
log.updateHighWatermark(new LogOffsetMetadata(offset))
749750
val snapshotId = new OffsetAndEpoch(offset, 1)
750-
TestUtils.resource(log.createNewSnapshotUnchecked(snapshotId).get()) { snapshot =>
751+
Using(log.createNewSnapshotUnchecked(snapshotId).get()) { snapshot =>
751752
snapshot.freeze()
752753
}
753754
log.truncateToLatestSnapshot()
@@ -771,7 +772,7 @@ final class KafkaMetadataLogTest {
771772
val log = buildMetadataLog(tempDir, mockTime)
772773
log.updateHighWatermark(new LogOffsetMetadata(offset))
773774
val snapshotId = new OffsetAndEpoch(offset, 1)
774-
TestUtils.resource(log.createNewSnapshotUnchecked(snapshotId).get()) { snapshot =>
775+
Using(log.createNewSnapshotUnchecked(snapshotId).get()) { snapshot =>
775776
snapshot.freeze()
776777
}
777778
log.truncateToLatestSnapshot()
@@ -853,13 +854,13 @@ final class KafkaMetadataLogTest {
853854
assertFalse(log.maybeClean(), "Should not clean since no snapshots exist")
854855

855856
val snapshotId1 = new OffsetAndEpoch(1000, 1)
856-
TestUtils.resource(log.createNewSnapshotUnchecked(snapshotId1).get()) { snapshot =>
857+
Using(log.createNewSnapshotUnchecked(snapshotId1).get()) { snapshot =>
857858
append(snapshot, 100)
858859
snapshot.freeze()
859860
}
860861

861862
val snapshotId2 = new OffsetAndEpoch(2000, 1)
862-
TestUtils.resource(log.createNewSnapshotUnchecked(snapshotId2).get()) { snapshot =>
863+
Using(log.createNewSnapshotUnchecked(snapshotId2).get()) { snapshot =>
863864
append(snapshot, 100)
864865
snapshot.freeze()
865866
}
@@ -891,7 +892,7 @@ final class KafkaMetadataLogTest {
891892

892893
for (offset <- Seq(100, 200, 300, 400, 500, 600)) {
893894
val snapshotId = new OffsetAndEpoch(offset, 1)
894-
TestUtils.resource(log.createNewSnapshotUnchecked(snapshotId).get()) { snapshot =>
895+
Using(log.createNewSnapshotUnchecked(snapshotId).get()) { snapshot =>
895896
append(snapshot, 10)
896897
snapshot.freeze()
897898
}
@@ -926,14 +927,14 @@ final class KafkaMetadataLogTest {
926927

927928
// Then generate two snapshots
928929
val snapshotId1 = new OffsetAndEpoch(1000, 1)
929-
TestUtils.resource(log.createNewSnapshotUnchecked(snapshotId1).get()) { snapshot =>
930+
Using(log.createNewSnapshotUnchecked(snapshotId1).get()) { snapshot =>
930931
append(snapshot, 500)
931932
snapshot.freeze()
932933
}
933934

934935
// Then generate a snapshot
935936
val snapshotId2 = new OffsetAndEpoch(2000, 1)
936-
TestUtils.resource(log.createNewSnapshotUnchecked(snapshotId2).get()) { snapshot =>
937+
Using(log.createNewSnapshotUnchecked(snapshotId2).get()) { snapshot =>
937938
append(snapshot, 500)
938939
snapshot.freeze()
939940
}
@@ -973,15 +974,15 @@ final class KafkaMetadataLogTest {
973974
log.log.logSegments.asScala.drop(1).head.baseOffset,
974975
1
975976
)
976-
TestUtils.resource(log.createNewSnapshotUnchecked(snapshotId1).get()) { snapshot =>
977+
Using(log.createNewSnapshotUnchecked(snapshotId1).get()) { snapshot =>
977978
snapshot.freeze()
978979
}
979980
// Generate second snapshots that includes the second segment by using the base offset of the third segment
980981
val snapshotId2 = new OffsetAndEpoch(
981982
log.log.logSegments.asScala.drop(2).head.baseOffset,
982983
1
983984
)
984-
TestUtils.resource(log.createNewSnapshotUnchecked(snapshotId2).get()) { snapshot =>
985+
Using(log.createNewSnapshotUnchecked(snapshotId2).get()) { snapshot =>
985986
snapshot.freeze()
986987
}
987988

‎core/src/test/scala/kafka/server/metadata/BrokerServerMetricsTest.scala

+6-6
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@
1818
package kafka.server.metadata
1919

2020
import java.util.Collections
21-
import kafka.utils.TestUtils
2221
import org.apache.kafka.common.MetricName
2322
import org.apache.kafka.common.metrics.Metrics
2423
import org.apache.kafka.common.utils.MockTime
@@ -28,6 +27,7 @@ import org.junit.jupiter.api.Assertions.assertTrue
2827
import org.junit.jupiter.api.Test
2928

3029
import scala.jdk.CollectionConverters._
30+
import scala.util.Using
3131

3232
final class BrokerServerMetricsTest {
3333
@Test
@@ -44,7 +44,7 @@ final class BrokerServerMetricsTest {
4444
new MetricName("metadata-apply-error-count", expectedGroup, "", Collections.emptyMap())
4545
)
4646

47-
TestUtils.resource(BrokerServerMetrics(metrics)) { brokerMetrics =>
47+
Using(BrokerServerMetrics(metrics)) { brokerMetrics =>
4848
val metricsMap = metrics.metrics().asScala.filter{ case (name, _) => name.group == expectedGroup }
4949
assertEquals(expectedMetrics.size, metricsMap.size)
5050
metricsMap.foreach { case (name, metric) =>
@@ -59,7 +59,7 @@ final class BrokerServerMetricsTest {
5959
@Test
6060
def testLastAppliedRecordOffset(): Unit = {
6161
val metrics = new Metrics()
62-
TestUtils.resource(BrokerServerMetrics(metrics)) { brokerMetrics =>
62+
Using(BrokerServerMetrics(metrics)) { brokerMetrics =>
6363
val offsetMetric = metrics.metrics().get(brokerMetrics.lastAppliedRecordOffsetName)
6464
assertEquals(-1L, offsetMetric.metricValue.asInstanceOf[Long])
6565

@@ -77,7 +77,7 @@ final class BrokerServerMetricsTest {
7777
def testLastAppliedRecordTimestamp(): Unit = {
7878
val time = new MockTime()
7979
val metrics = new Metrics(time)
80-
TestUtils.resource(BrokerServerMetrics(metrics)) { brokerMetrics =>
80+
Using(BrokerServerMetrics(metrics)) { brokerMetrics =>
8181
time.sleep(1000)
8282
val timestampMetric = metrics.metrics().get(brokerMetrics.lastAppliedRecordTimestampName)
8383
val lagMetric = metrics.metrics().get(brokerMetrics.lastAppliedRecordLagMsName)
@@ -101,7 +101,7 @@ final class BrokerServerMetricsTest {
101101
def testMetadataLoadErrorCount(): Unit = {
102102
val time = new MockTime()
103103
val metrics = new Metrics(time)
104-
TestUtils.resource(BrokerServerMetrics(metrics)) { brokerMetrics =>
104+
Using(BrokerServerMetrics(metrics)) { brokerMetrics =>
105105
val metadataLoadErrorCountMetric = metrics.metrics().get(brokerMetrics.metadataLoadErrorCountName)
106106

107107
assertEquals(0L, metadataLoadErrorCountMetric.metricValue.asInstanceOf[Long])
@@ -117,7 +117,7 @@ final class BrokerServerMetricsTest {
117117
def testMetadataApplyErrorCount(): Unit = {
118118
val time = new MockTime()
119119
val metrics = new Metrics(time)
120-
TestUtils.resource(BrokerServerMetrics(metrics)) { brokerMetrics =>
120+
Using(BrokerServerMetrics(metrics)) { brokerMetrics =>
121121
val metadataApplyErrorCountMetric = metrics.metrics().get(brokerMetrics.metadataApplyErrorCountName)
122122

123123
assertEquals(0L, metadataApplyErrorCountMetric.metricValue.asInstanceOf[Long])

0 commit comments

Comments
 (0)