root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

KAFKA-17589 Move JUnit extensions to test-common module (#17318) 

This patch completely removes the compile-time dependency on core for both test and main sources by introducing two new modules.

1) `test-common` include all the common test implementation code (including dependency on :core for BrokerServer, ControllerServer, etc)
2) `test-common:api` new sub-module that just includes interfaces including our junit extension

Reviewers: David Arthur <mumrah@gmail.com>
This commit is contained in:
Chia-Ping Tsai 2024-10-03 10:28:37 +08:00 committed by GitHub
parent 22a14d75c2
commit 979740b49d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
100 changed files with 796 additions and 417 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

156
build.gradle
View File

@ -1062,6 +1062,8 @@ project(':core') {
testImplementation project(':storage:storage-api').sourceSets.test.output
testImplementation project(':server').sourceSets.test.output
testImplementation project(':share').sourceSets.test.output
testImplementation project(':test-common')
testImplementation project(':test-common:test-common-api')
testImplementation libs.bcpkix
testImplementation libs.mockitoCore
testImplementation(libs.apacheda) {
@ -1111,8 +1113,7 @@ project(':core') {
}
tasks.create(name: "copyDependantLibs", type: Copy) {
from (configurations.testRuntimeClasspath) {
include('slf4j-log4j12*')
from (configurations.compileClasspath) {
include('reload4j*jar')
}
from (configurations.runtimeClasspath) {
@ -1264,7 +1265,7 @@ project(':core') {
}
jar {
dependsOn('copyDependantLibs')
dependsOn copyDependantLibs
}
jar.manifest {
@ -1521,6 +1522,91 @@ project(':group-coordinator') {
srcJar.dependsOn 'processMessages'
}
project(':test-common') {
base {
archivesName = "kafka-test-common"
}
dependencies {
implementation libs.scalaJava8Compat
implementation project(':core')
implementation project(':metadata')
implementation project(':server')
implementation project(':raft')
implementation project(':storage')
implementation project(':server-common')
implementation libs.slf4jApi
testImplementation libs.junitJupiter
testImplementation libs.mockitoCore
testRuntimeOnly libs.slf4jReload4j
testRuntimeOnly libs.junitPlatformLanucher
}
sourceSets {
main {
java {
srcDirs = ["src/main/java"]
}
}
test {
java {
srcDirs = ["src/test/java"]
}
}
}
checkstyle {
sourceSets = []
}
javadoc {
enabled = false
}
}
project(':test-common:test-common-api') {
base {
archivesName = "kafka-test-common-api"
}
dependencies {
implementation libs.scalaJava8Compat
implementation project(':clients')
implementation project(':core')
implementation project(':group-coordinator')
implementation project(':metadata')
implementation project(':raft')
implementation project(':server')
implementation project(':server-common')
implementation project(':test-common')
implementation libs.junitJupiter
testImplementation libs.mockitoCore
testRuntimeOnly libs.junitPlatformLanucher
}
sourceSets {
main {
java {
srcDirs = ["src/main/java"]
}
}
test {
java {
srcDirs = ["src/test/java"]
}
}
}
checkstyle {
sourceSets = []
}
javadoc {
enabled = false
}
}
project(':transaction-coordinator') {
base {
archivesName = "kafka-transaction-coordinator"
@ -2166,6 +2252,7 @@ project(':storage') {
testImplementation project(':clients').sourceSets.test.output
testImplementation project(':core')
testImplementation project(':core').sourceSets.test.output
testImplementation project(':test-common:test-common-api')
testImplementation project(':server')
testImplementation project(':server-common')
testImplementation project(':server-common').sourceSets.test.output
@ -2356,6 +2443,7 @@ project(':tools') {
testImplementation project(':server').sourceSets.test.output
testImplementation project(':core')
testImplementation project(':core').sourceSets.test.output
testImplementation project(':test-common:test-common-api')
testImplementation project(':server-common')
testImplementation project(':server-common').sourceSets.test.output
testImplementation project(':connect:api')
@ -2386,10 +2474,6 @@ project(':tools') {
}
tasks.create(name: "copyDependantLibs", type: Copy) {
from (configurations.testRuntimeClasspath) {
include('slf4j-log4j12*')
include('reload4j*jar')
}
from (configurations.runtimeClasspath) {
exclude('kafka-clients*')
}
@ -2413,6 +2497,7 @@ project(':trogdor') {
implementation libs.jacksonDatabind
implementation libs.jacksonJDK8Datatypes
implementation libs.slf4jApi
runtimeOnly libs.reload4j
implementation libs.jacksonJaxrsJsonProvider
implementation libs.jerseyContainerServlet
@ -2437,10 +2522,6 @@ project(':trogdor') {
}
tasks.create(name: "copyDependantLibs", type: Copy) {
from (configurations.testRuntimeClasspath) {
include('slf4j-log4j12*')
include('reload4j*jar')
}
from (configurations.runtimeClasspath) {
exclude('kafka-clients*')
}
@ -2476,7 +2557,6 @@ project(':shell') {
testImplementation project(':clients')
testImplementation project(':clients').sourceSets.test.output
testImplementation project(':core')
testImplementation project(':core').sourceSets.test.output
testImplementation project(':server-common')
testImplementation project(':server-common').sourceSets.test.output
testImplementation libs.junitJupiter
@ -2490,9 +2570,6 @@ project(':shell') {
}
tasks.create(name: "copyDependantLibs", type: Copy) {
from (configurations.testRuntimeClasspath) {
include('jline-*jar')
}
from (configurations.runtimeClasspath) {
include('jline-*jar')
}
@ -2533,7 +2610,7 @@ project(':streams') {
testImplementation project(':server')
testImplementation project(':core')
testImplementation project(':tools')
testImplementation project(':core').sourceSets.test.output
testImplementation project(':test-common')
testImplementation project(':storage')
testImplementation project(':group-coordinator')
testImplementation project(':transaction-coordinator')
@ -2692,7 +2769,7 @@ project(':streams:streams-scala') {
}
testImplementation project(':group-coordinator')
testImplementation project(':core')
testImplementation project(':core').sourceSets.test.output
testImplementation project(':test-common')
testImplementation project(':server-common').sourceSets.test.output
testImplementation project(':streams').sourceSets.test.output
testImplementation project(':clients').sourceSets.test.output
@ -3285,6 +3362,7 @@ project(':connect:api') {
dependencies {
api project(':clients')
implementation libs.slf4jApi
runtimeOnly libs.reload4j
implementation libs.jaxrsApi
testImplementation libs.junitJupiter
@ -3298,10 +3376,6 @@ project(':connect:api') {
}
tasks.create(name: "copyDependantLibs", type: Copy) {
from (configurations.testRuntimeClasspath) {
include('slf4j-log4j12*')
include('reload4j*jar')
}
from (configurations.runtimeClasspath) {
exclude('kafka-clients*')
exclude('connect-*')
@ -3324,6 +3398,7 @@ project(':connect:transforms') {
api project(':connect:api')
implementation libs.slf4jApi
runtimeOnly libs.reload4j
testImplementation libs.junitJupiter
@ -3337,10 +3412,6 @@ project(':connect:transforms') {
}
tasks.create(name: "copyDependantLibs", type: Copy) {
from (configurations.testRuntimeClasspath) {
include('slf4j-log4j12*')
include('reload4j*jar')
}
from (configurations.runtimeClasspath) {
exclude('kafka-clients*')
exclude('connect-*')
@ -3367,6 +3438,7 @@ project(':connect:json') {
api libs.jacksonAfterburner
implementation libs.slf4jApi
runtimeOnly libs.reload4j
testImplementation libs.junitJupiter
@ -3380,10 +3452,6 @@ project(':connect:json') {
}
tasks.create(name: "copyDependantLibs", type: Copy) {
from (configurations.testRuntimeClasspath) {
include('slf4j-log4j12*')
include('reload4j*jar')
}
from (configurations.runtimeClasspath) {
exclude('kafka-clients*')
exclude('connect-*')
@ -3440,7 +3508,7 @@ project(':connect:runtime') {
testImplementation project(':server')
testImplementation project(':metadata')
testImplementation project(':server-common')
testImplementation project(':core').sourceSets.test.output
testImplementation project(':test-common')
testImplementation project(':server-common')
testImplementation project(':server')
testImplementation project(':group-coordinator')
@ -3464,10 +3532,6 @@ project(':connect:runtime') {
}
tasks.create(name: "copyDependantLibs", type: Copy) {
from (configurations.testRuntimeClasspath) {
// No need to copy log4j since the module has an explicit dependency on that
include('slf4j-log4j12*')
}
from (configurations.runtimeClasspath) {
exclude('kafka-clients*')
exclude('connect-*')
@ -3553,6 +3617,7 @@ project(':connect:file') {
dependencies {
implementation project(':connect:api')
implementation libs.slf4jApi
runtimeOnly libs.reload4j
testImplementation libs.junitJupiter
testImplementation libs.mockitoCore
@ -3563,7 +3628,7 @@ project(':connect:file') {
testImplementation project(':connect:runtime')
testImplementation project(':connect:runtime').sourceSets.test.output
testImplementation project(':core')
testImplementation project(':core').sourceSets.test.output
testImplementation project(':test-common')
testImplementation project(':server-common').sourceSets.test.output
}
@ -3572,10 +3637,6 @@ project(':connect:file') {
}
tasks.create(name: "copyDependantLibs", type: Copy) {
from (configurations.testRuntimeClasspath) {
include('slf4j-log4j12*')
include('reload4j*jar')
}
from (configurations.runtimeClasspath) {
exclude('kafka-clients*')
exclude('connect-*')
@ -3597,6 +3658,7 @@ project(':connect:basic-auth-extension') {
dependencies {
implementation project(':connect:api')
implementation libs.slf4jApi
runtimeOnly libs.reload4j
implementation libs.jaxrsApi
implementation libs.jaxAnnotationApi
@ -3615,10 +3677,6 @@ project(':connect:basic-auth-extension') {
}
tasks.create(name: "copyDependantLibs", type: Copy) {
from (configurations.testRuntimeClasspath) {
include('slf4j-log4j12*')
include('reload4j*jar')
}
from (configurations.runtimeClasspath) {
exclude('kafka-clients*')
exclude('connect-*')
@ -3646,6 +3704,7 @@ project(':connect:mirror') {
implementation libs.argparse4j
implementation libs.jacksonAnnotations
implementation libs.slf4jApi
runtimeOnly libs.reload4j
implementation libs.jacksonAnnotations
implementation libs.jacksonJaxrsJsonProvider
implementation libs.jerseyContainerServlet
@ -3664,7 +3723,7 @@ project(':connect:mirror') {
testImplementation project(':clients').sourceSets.test.output
testImplementation project(':connect:runtime').sourceSets.test.output
testImplementation project(':core')
testImplementation project(':core').sourceSets.test.output
testImplementation project(':test-common')
testImplementation project(':server')
testImplementation project(':server-common').sourceSets.test.output
@ -3679,10 +3738,6 @@ project(':connect:mirror') {
}
tasks.create(name: "copyDependantLibs", type: Copy) {
from (configurations.testRuntimeClasspath) {
include('slf4j-log4j12*')
include('reload4j*jar')
}
from (configurations.runtimeClasspath) {
exclude('kafka-clients*')
exclude('connect-*')
@ -3732,6 +3787,7 @@ project(':connect:mirror-client') {
dependencies {
implementation project(':clients')
implementation libs.slf4jApi
runtimeOnly libs.reload4j
testImplementation libs.junitJupiter
testImplementation project(':clients').sourceSets.test.output
@ -3745,10 +3801,6 @@ project(':connect:mirror-client') {
}
tasks.create(name: "copyDependantLibs", type: Copy) {
from (configurations.testRuntimeClasspath) {
include('slf4j-log4j12*')
include('reload4j*jar')
}
from (configurations.runtimeClasspath) {
exclude('kafka-clients*')
exclude('connect-*')

20
checkstyle/import-control-core.xml
View File

@ -102,9 +102,9 @@
<allow pkg="org.apache.kafka.server.authorizer"/>
<allow pkg="org.apache.kafka.server.common" />
<allow pkg="org.apache.kafka.test" />
<allow pkg="kafka.testkit"/>
<allow pkg="kafka.test.annotation"/>
<allow pkg="kafka.test.junit"/>
<allow pkg="org.apache.kafka.common.test"/>
<allow pkg="org.apache.kafka.common.test.api"/>
<allow pkg="org.apache.kafka.common.test.api"/>
<allow pkg="kafka.network"/>
<allow pkg="kafka.api"/>
<allow pkg="kafka.server"/>
@ -113,14 +113,6 @@
<allow pkg="org.apache.kafka.clients.consumer"/>
<allow pkg="org.apache.kafka.coordinator.group"/>
<allow pkg="org.apache.kafka.coordinator.transaction"/>
<subpackage name="annotation">
<allow pkg="kafka.test"/>
</subpackage>
<subpackage name="junit">
<allow pkg="kafka.test"/>
<allow pkg="org.apache.kafka.clients"/>
<allow pkg="org.apache.kafka.metadata" />
</subpackage>
<subpackage name="server">
<allow pkg="kafka.test" />
</subpackage>
@ -139,9 +131,9 @@
<allow pkg="org.apache.kafka.server"/>
<allow pkg="org.apache.kafka.test"/>
<allow pkg="org.apache.log4j"/>
<allow pkg="kafka.test"/>
<allow pkg="kafka.test.annotation"/>
<allow pkg="kafka.test.junit"/>
<allow pkg="org.apache.kafka.common.test"/>
<allow pkg="org.apache.kafka.common.test.api"/>
<allow pkg="org.apache.kafka.common.test.api"/>
</subpackage>
<subpackage name="security">

4
checkstyle/import-control-storage.xml
View File

@ -53,7 +53,7 @@
<allow pkg="com.fasterxml.jackson" />
<allow pkg="kafka.api" />
<allow pkg="kafka.utils" />
<allow pkg="kafka.test" />
<allow pkg="org.apache.kafka.common.test" />
<allow pkg="org.apache.kafka.clients" />
<allow pkg="org.apache.kafka.server.common" />
<allow pkg="org.apache.kafka.server.config" />
@ -69,7 +69,7 @@
<subpackage name="storage">
<allow pkg="com.yammer.metrics.core" />
<allow pkg="org.apache.kafka.server.metrics" />
<allow pkg="kafka.test" />
<allow pkg="org.apache.kafka.common.test" />
</subpackage>
</subpackage>
</subpackage>

6
checkstyle/import-control.xml
View File

@ -305,7 +305,7 @@
<allow pkg="org.jose4j" />
<allow pkg="net.sourceforge.argparse4j" />
<allow pkg="org.apache.log4j" />
<allow pkg="kafka.test" />
<allow pkg="org.apache.kafka.common.test" />
<allow pkg="joptsimple" />
<allow pkg="javax.rmi.ssl"/>
<allow pkg="kafka.utils" />
@ -393,7 +393,7 @@
</subpackage>
<subpackage name="integration">
<allow pkg="kafka.testkit"/>
<allow pkg="org.apache.kafka.common.test"/>
<allow pkg="org.apache.kafka.metadata"/>
<allow pkg="kafka.admin" />
<allow pkg="kafka.api" />
@ -603,7 +603,7 @@
<allow pkg="org.apache.kafka.server.config" />
<allow pkg="kafka.cluster" />
<allow pkg="kafka.server" />
<allow pkg="kafka.testkit" />
<allow pkg="org.apache.kafka.common.test" />
<allow pkg="kafka.zk" />
<allow pkg="kafka.utils" />
<allow class="javax.servlet.http.HttpServletResponse" />

4
connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java
View File

@ -17,8 +17,6 @@
package org.apache.kafka.connect.util.clusters;
import kafka.server.BrokerServer;
import kafka.testkit.KafkaClusterTestKit;
import kafka.testkit.TestKitNodes;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.Admin;
@ -48,6 +46,8 @@ import org.apache.kafka.common.config.types.Password;
import org.apache.kafka.common.errors.InvalidReplicationFactorException;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.test.KafkaClusterTestKit;
import org.apache.kafka.common.test.TestKitNodes;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;

12
core/src/test/java/kafka/admin/AclCommandTest.java
View File

@ -17,12 +17,6 @@
package kafka.admin;
import kafka.admin.AclCommand.AclCommandOptions;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterConfigProperty;
import kafka.test.annotation.ClusterTest;
import kafka.test.annotation.ClusterTestDefaults;
import kafka.test.annotation.Type;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.common.acl.AccessControlEntry;
import org.apache.kafka.common.acl.AclBindingFilter;
@ -33,6 +27,12 @@ import org.apache.kafka.common.resource.Resource;
import org.apache.kafka.common.resource.ResourcePattern;
import org.apache.kafka.common.resource.ResourceType;
import org.apache.kafka.common.security.auth.KafkaPrincipal;
import org.apache.kafka.common.test.api.ClusterConfigProperty;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestDefaults;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.common.test.api.Type;
import org.apache.kafka.common.utils.AppInfoParser;
import org.apache.kafka.common.utils.Exit;
import org.apache.kafka.common.utils.LogCaptureAppender;

11
core/src/test/java/kafka/admin/AdminFenceProducersTest.java
View File

@ -17,12 +17,6 @@
package kafka.admin;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterConfigProperty;
import kafka.test.annotation.ClusterTest;
import kafka.test.annotation.ClusterTestDefaults;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.FenceProducersOptions;
@ -34,6 +28,11 @@ import org.apache.kafka.common.errors.InvalidProducerEpochException;
import org.apache.kafka.common.errors.ProducerFencedException;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.test.api.ClusterConfigProperty;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestDefaults;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.coordinator.transaction.TransactionLogConfig;
import org.apache.kafka.coordinator.transaction.TransactionStateManagerConfig;
import org.apache.kafka.server.config.ServerLogConfigs;

11
core/src/test/java/kafka/admin/ClientTelemetryTest.java
View File

@ -17,12 +17,6 @@
package kafka.admin;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterConfigProperty;
import kafka.test.annotation.ClusterTest;
import kafka.test.annotation.Type;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
@ -42,6 +36,11 @@ import org.apache.kafka.common.metrics.KafkaMetric;
import org.apache.kafka.common.metrics.MetricsReporter;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.test.api.ClusterConfigProperty;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.common.test.api.Type;
import org.apache.kafka.server.telemetry.ClientTelemetry;
import org.apache.kafka.server.telemetry.ClientTelemetryReceiver;

9
core/src/test/java/kafka/admin/ConfigCommandIntegrationTest.java
View File

@ -16,14 +16,13 @@
*/
package kafka.admin;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterTest;
import kafka.test.annotation.Type;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.ConfigEntry;
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.common.test.api.Type;
import org.apache.kafka.common.utils.Exit;
import org.apache.kafka.test.TestUtils;

7
core/src/test/java/kafka/admin/UserScramCredentialsCommandTest.java
View File

@ -16,10 +16,9 @@
*/
package kafka.admin;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterTest;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.common.utils.Exit;
import org.apache.kafka.test.NoRetryException;
import org.apache.kafka.test.TestUtils;

11
core/src/test/java/kafka/server/LogManagerIntegrationTest.java
View File

@ -16,12 +16,6 @@
*/
package kafka.server;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterTest;
import kafka.test.annotation.Type;
import kafka.test.junit.ClusterTestExtensions;
import kafka.test.junit.RaftClusterInvocationContext;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.Consumer;
@ -37,6 +31,11 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.TopicPartitionInfo;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.common.test.api.RaftClusterInvocationContext;
import org.apache.kafka.common.test.api.Type;
import org.apache.kafka.storage.internals.checkpoint.PartitionMetadataFile;
import org.apache.kafka.test.TestUtils;

13
core/src/test/java/kafka/test/server/BootstrapControllersIntegrationTest.java
View File

@ -17,13 +17,6 @@
package kafka.test.server;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterConfigProperty;
import kafka.test.annotation.ClusterTest;
import kafka.test.annotation.ClusterTestDefaults;
import kafka.test.annotation.Type;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AlterConfigOp;
import org.apache.kafka.clients.admin.Config;
@ -58,6 +51,12 @@ import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.resource.PatternType;
import org.apache.kafka.common.resource.ResourcePattern;
import org.apache.kafka.common.resource.ResourceType;
import org.apache.kafka.common.test.api.ClusterConfigProperty;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestDefaults;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.common.test.api.Type;
import org.apache.kafka.metadata.authorizer.StandardAuthorizer;
import org.apache.kafka.server.common.MetadataVersion;
import org.apache.kafka.test.TestUtils;

6
core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala
View File

@ -13,9 +13,9 @@
package kafka.api
import kafka.log.UnifiedLog
import kafka.test.ClusterInstance
import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, Type}
import kafka.test.junit.ClusterTestExtensions
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.test.api.{ClusterConfigProperty, ClusterTest, Type}
import org.apache.kafka.common.test.api.ClusterTestExtensions
import kafka.utils.TestUtils
import org.apache.kafka.clients.admin.{Admin, ConsumerGroupDescription}
import org.apache.kafka.clients.consumer.{Consumer, GroupProtocol, OffsetAndMetadata}

6
core/src/test/scala/integration/kafka/coordinator/transaction/ProducerIdsIntegrationTest.scala
View File

@ -19,9 +19,9 @@ package kafka.coordinator.transaction
import kafka.network.SocketServer
import kafka.server.IntegrationTestUtils
import kafka.test.ClusterInstance
import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, ClusterTests, Type}
import kafka.test.junit.ClusterTestExtensions
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.test.api.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, ClusterTests, Type}
import org.apache.kafka.common.test.api.ClusterTestExtensions
import org.apache.kafka.common.message.InitProducerIdRequestData
import org.apache.kafka.common.network.ListenerName
import org.apache.kafka.common.protocol.Errors

2
core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala
View File

@ -20,7 +20,7 @@ package kafka.server
import kafka.log.UnifiedLog
import kafka.network.SocketServer
import kafka.server.IntegrationTestUtils.connectAndReceive
import kafka.testkit.{KafkaClusterTestKit, TestKitNodes}
import org.apache.kafka.common.test.{KafkaClusterTestKit, TestKitNodes}
import kafka.utils.TestUtils
import org.apache.commons.io.FileUtils
import org.apache.kafka.clients.admin.AlterConfigOp.OpType

6
core/src/test/scala/integration/kafka/server/MetadataVersionIntegrationTest.scala
View File

@ -17,9 +17,9 @@
package kafka.server
import kafka.test.ClusterInstance
import kafka.test.annotation.{ClusterTest, ClusterTests, Type}
import kafka.test.junit.ClusterTestExtensions
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.test.api.{ClusterTest, ClusterTests, Type}
import org.apache.kafka.common.test.api.ClusterTestExtensions
import kafka.utils.TestUtils
import org.apache.kafka.clients.admin.FeatureUpdate.UpgradeType
import org.apache.kafka.clients.admin.{FeatureUpdate, UpdateFeaturesOptions}

4
core/src/test/scala/integration/kafka/server/RaftClusterSnapshotTest.scala
View File

@ -17,8 +17,8 @@
package kafka.server
import kafka.testkit.KafkaClusterTestKit
import kafka.testkit.TestKitNodes
import org.apache.kafka.common.test.KafkaClusterTestKit
import org.apache.kafka.common.test.TestKitNodes
import kafka.utils.TestUtils
import org.apache.kafka.common.utils.BufferSupplier
import org.apache.kafka.metadata.MetadataRecordSerde

2
core/src/test/scala/unit/kafka/server/AbstractApiVersionsRequestTest.scala
View File

@ -16,7 +16,7 @@
*/
package kafka.server
import kafka.test.ClusterInstance
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.clients.NodeApiVersions
import org.apache.kafka.common.message.ApiMessageType.ListenerType
import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersion

8
core/src/test/scala/unit/kafka/server/AllocateProducerIdsRequestTest.scala
View File

@ -18,10 +18,10 @@ package unit.kafka.server
import kafka.network.SocketServer
import kafka.server.{BrokerServer, ControllerServer, IntegrationTestUtils}
import kafka.test.ClusterInstance
import kafka.test.annotation.{ClusterTest, ClusterTestDefaults, Type}
import kafka.test.junit.ClusterTestExtensions
import kafka.test.junit.RaftClusterInvocationContext.RaftClusterInstance
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.test.api.{ClusterTest, ClusterTestDefaults, Type}
import org.apache.kafka.common.test.api.ClusterTestExtensions
import org.apache.kafka.common.test.api.RaftClusterInvocationContext.RaftClusterInstance
import org.apache.kafka.common.message.AllocateProducerIdsRequestData
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.requests._

6
core/src/test/scala/unit/kafka/server/ApiVersionsRequestTest.scala
View File

@ -17,12 +17,12 @@
package kafka.server
import kafka.test.ClusterInstance
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.message.ApiVersionsRequestData
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.common.requests.ApiVersionsRequest
import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, Type}
import kafka.test.junit.ClusterTestExtensions
import org.apache.kafka.common.test.api.{ClusterConfigProperty, ClusterTest, Type}
import org.apache.kafka.common.test.api.ClusterTestExtensions
import org.apache.kafka.server.common.MetadataVersion
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.extension.ExtendWith

6
core/src/test/scala/unit/kafka/server/BrokerMetricNamesTest.scala
View File

@ -17,9 +17,9 @@
package kafka.server
import kafka.test.ClusterInstance
import kafka.test.annotation.ClusterTest
import kafka.test.junit.ClusterTestExtensions
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.test.api.ClusterTest
import org.apache.kafka.common.test.api.ClusterTestExtensions
import kafka.utils.TestUtils
import org.apache.kafka.server.metrics.KafkaYammerMetrics
import org.junit.jupiter.api.AfterEach

10
core/src/test/scala/unit/kafka/server/BrokerRegistrationRequestTest.scala
View File

@ -17,10 +17,8 @@
package kafka.server
import kafka.test.ClusterInstance
import kafka.test.annotation._
import kafka.test.junit.ClusterTestExtensions
import kafka.test.junit.RaftClusterInvocationContext.RaftClusterInstance
import org.apache.kafka.common.test.api.{AutoStart, ClusterConfigProperty, ClusterInstance, ClusterTest, ClusterTestExtensions, ClusterTests, Type}
import org.apache.kafka.common.test.api.RaftClusterInvocationContext.RaftClusterInstance
import org.apache.kafka.clients.ClientResponse
import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic
import org.apache.kafka.common.message.{BrokerRegistrationRequestData, CreateTopicsRequestData}
@ -34,6 +32,7 @@ import org.apache.kafka.common.{Node, Uuid}
import org.apache.kafka.server.{ControllerRequestCompletionHandler, NodeToControllerChannelManager}
import org.apache.kafka.server.common.{Features, MetadataVersion}
import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows}
import org.junit.jupiter.api.Disabled
import org.junit.jupiter.api.extension.ExtendWith
import java.util
@ -207,6 +206,7 @@ class BrokerRegistrationRequestTest {
}
}
@Disabled("zk migration is removed from ClusterTestExtensions")
@ClusterTest(
types = Array(Type.KRAFT),
brokers = 1,
@ -243,6 +243,7 @@ class BrokerRegistrationRequestTest {
}
}
@Disabled("zk migration is removed from ClusterTestExtensions")
@ClusterTest(
types = Array(Type.KRAFT),
brokers = 1,
@ -275,6 +276,7 @@ class BrokerRegistrationRequestTest {
* Start a KRaft cluster with migrations enabled, verify that the controller does not accept metadata changes
* through the RPCs. The migration never proceeds past pre-migration since no ZK brokers are registered.
*/
@Disabled("zk migration is removed from ClusterTestExtensions")
@ClusterTests(Array(
new ClusterTest(types = Array(Type.KRAFT), autoStart = AutoStart.NO, controllers = 1, metadataVersion = MetadataVersion.IBP_3_4_IV0,
serverProperties = Array(new ClusterConfigProperty(key = "zookeeper.metadata.migration.enable", value = "true")))

6
core/src/test/scala/unit/kafka/server/ClientQuotasRequestTest.scala
View File

@ -20,9 +20,9 @@ package kafka.server
import java.net.InetAddress
import java.util
import java.util.concurrent.{ExecutionException, TimeUnit}
import kafka.test.ClusterInstance
import kafka.test.annotation.ClusterTest
import kafka.test.junit.ClusterTestExtensions
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.test.api.ClusterTest
import org.apache.kafka.common.test.api.ClusterTestExtensions
import kafka.utils.TestUtils
import org.apache.kafka.clients.admin.{ScramCredentialInfo, ScramMechanism, UserScramCredentialUpsertion}
import org.apache.kafka.common.errors.{InvalidRequestException, UnsupportedVersionException}

6
core/src/test/scala/unit/kafka/server/ConsumerGroupDescribeRequestTest.scala
View File

@ -16,9 +16,9 @@
*/
package kafka.server
import kafka.test.ClusterInstance
import kafka.test.annotation._
import kafka.test.junit.ClusterTestExtensions
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.test.api._
import org.apache.kafka.common.test.api.ClusterTestExtensions
import kafka.utils.TestUtils
import org.apache.kafka.common.ConsumerGroupState
import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData.{Assignment, DescribedGroup, TopicPartitions}

8
core/src/test/scala/unit/kafka/server/ConsumerGroupHeartbeatRequestTest.scala
View File

@ -16,10 +16,10 @@
*/
package kafka.server
import kafka.test.ClusterInstance
import kafka.test.annotation.{ClusterConfigProperty, ClusterFeature, ClusterTest, Type}
import kafka.test.junit.ClusterTestExtensions
import kafka.test.junit.RaftClusterInvocationContext.RaftClusterInstance
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.test.api.{ClusterConfigProperty, ClusterFeature, ClusterTest, Type}
import org.apache.kafka.common.test.api.ClusterTestExtensions
import org.apache.kafka.common.test.api.RaftClusterInvocationContext.RaftClusterInstance
import kafka.utils.TestUtils
import org.apache.kafka.clients.admin.AlterConfigOp.OpType
import org.apache.kafka.clients.admin.{AlterConfigOp, ConfigEntry}

6
core/src/test/scala/unit/kafka/server/ConsumerProtocolMigrationTest.scala
View File

@ -16,9 +16,9 @@
*/
package kafka.server
import kafka.test.ClusterInstance
import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
import kafka.test.junit.ClusterTestExtensions
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.test.api.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
import org.apache.kafka.common.test.api.ClusterTestExtensions
import org.apache.kafka.common.message.ListGroupsResponseData
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.coordinator.group.{Group, GroupCoordinatorConfig}

2
core/src/test/scala/unit/kafka/server/ControllerApisTest.scala
View File

@ -21,7 +21,7 @@ import kafka.network.RequestChannel
import kafka.raft.RaftManager
import kafka.server.QuotaFactory.QuotaManagers
import kafka.server.metadata.KRaftMetadataCache
import kafka.test.MockController
import org.apache.kafka.common.test.MockController
import org.apache.kafka.clients.admin.AlterConfigOp
import org.apache.kafka.common.Uuid.ZERO_UUID
import org.apache.kafka.common.acl.AclOperation

6
core/src/test/scala/unit/kafka/server/DeleteGroupsRequestTest.scala
View File

@ -16,9 +16,9 @@
*/
package kafka.server
import kafka.test.ClusterInstance
import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
import kafka.test.junit.ClusterTestExtensions
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.test.api.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
import org.apache.kafka.common.test.api.ClusterTestExtensions
import org.apache.kafka.common.message.DescribeGroupsResponseData.DescribedGroup
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.coordinator.group.GroupCoordinatorConfig

6
core/src/test/scala/unit/kafka/server/DescribeGroupsRequestTest.scala
View File

@ -16,9 +16,9 @@
*/
package kafka.server
import kafka.test.ClusterInstance
import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
import kafka.test.junit.ClusterTestExtensions
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.test.api.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
import org.apache.kafka.common.test.api.ClusterTestExtensions
import org.apache.kafka.common.message.DescribeGroupsResponseData.{DescribedGroup, DescribedGroupMember}
import org.apache.kafka.common.protocol.ApiKeys
import org.apache.kafka.coordinator.group.GroupCoordinatorConfig

6
core/src/test/scala/unit/kafka/server/DescribeQuorumRequestTest.scala
View File

@ -16,9 +16,9 @@
*/
package kafka.server
import kafka.test.ClusterInstance
import kafka.test.annotation.{ClusterTest, ClusterTestDefaults, Type}
import kafka.test.junit.ClusterTestExtensions
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.test.api.{ClusterTest, ClusterTestDefaults, Type}
import org.apache.kafka.common.test.api.ClusterTestExtensions
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.common.requests.DescribeQuorumRequest.singletonRequest
import org.apache.kafka.common.requests.{AbstractRequest, AbstractResponse, DescribeQuorumRequest, DescribeQuorumResponse}

2
core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala
View File

@ -17,7 +17,7 @@
package kafka.server
import kafka.network.SocketServer
import kafka.test.ClusterInstance
import org.apache.kafka.common.test.api.ClusterInstance
import kafka.utils.TestUtils
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord, RecordMetadata}
import org.apache.kafka.common.{TopicIdPartition, TopicPartition, Uuid}

6
core/src/test/scala/unit/kafka/server/HeartbeatRequestTest.scala
View File

@ -16,9 +16,9 @@
*/
package kafka.server
import kafka.test.ClusterInstance
import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
import kafka.test.junit.ClusterTestExtensions
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.test.api.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
import org.apache.kafka.common.test.api.ClusterTestExtensions
import kafka.utils.TestUtils
import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor
import org.apache.kafka.clients.consumer.internals.ConsumerProtocol

6
core/src/test/scala/unit/kafka/server/JoinGroupRequestTest.scala
View File

@ -16,9 +16,9 @@
*/
package kafka.server
import kafka.test.ClusterInstance
import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, Type}
import kafka.test.junit.ClusterTestExtensions
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.test.api.{ClusterConfigProperty, ClusterTest, Type}
import org.apache.kafka.common.test.api.ClusterTestExtensions
import kafka.utils.TestUtils
import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor
import org.apache.kafka.clients.consumer.internals.ConsumerProtocol

6
core/src/test/scala/unit/kafka/server/LeaveGroupRequestTest.scala
View File

@ -16,9 +16,9 @@
*/
package kafka.server
import kafka.test.ClusterInstance
import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
import kafka.test.junit.ClusterTestExtensions
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.test.api.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
import org.apache.kafka.common.test.api.ClusterTestExtensions
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.common.requests.JoinGroupRequest
import org.apache.kafka.coordinator.group.GroupCoordinatorConfig

6
core/src/test/scala/unit/kafka/server/ListGroupsRequestTest.scala
View File

@ -16,9 +16,9 @@
*/
package kafka.server
import kafka.test.ClusterInstance
import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
import kafka.test.junit.ClusterTestExtensions
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.test.api.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
import org.apache.kafka.common.test.api.ClusterTestExtensions
import org.apache.kafka.common.message.ListGroupsResponseData
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.coordinator.group.classic.ClassicGroupState

6
core/src/test/scala/unit/kafka/server/OffsetCommitRequestTest.scala
View File

@ -16,9 +16,9 @@
*/
package kafka.server
import kafka.test.ClusterInstance
import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
import kafka.test.junit.ClusterTestExtensions
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.test.api.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
import org.apache.kafka.common.test.api.ClusterTestExtensions
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.coordinator.group.GroupCoordinatorConfig
import org.junit.jupiter.api.Assertions.fail

6
core/src/test/scala/unit/kafka/server/OffsetDeleteRequestTest.scala
View File

@ -16,9 +16,9 @@
*/
package kafka.server
import kafka.test.ClusterInstance
import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
import kafka.test.junit.ClusterTestExtensions
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.test.api.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
import org.apache.kafka.common.test.api.ClusterTestExtensions
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.coordinator.group.GroupCoordinatorConfig
import org.junit.jupiter.api.Assertions.fail

6
core/src/test/scala/unit/kafka/server/OffsetFetchRequestTest.scala
View File

@ -17,9 +17,9 @@
package kafka.server
import kafka.test.ClusterInstance
import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
import kafka.test.junit.ClusterTestExtensions
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.test.api.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
import org.apache.kafka.common.test.api.ClusterTestExtensions
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.message.OffsetFetchResponseData
import org.apache.kafka.common.protocol.{ApiKeys, Errors}

4
core/src/test/scala/unit/kafka/server/SaslApiVersionsRequestTest.scala
View File

@ -19,9 +19,7 @@ package kafka.server
import kafka.api.{KafkaSasl, SaslSetup}
import kafka.security.JaasTestUtils
import kafka.server.SaslApiVersionsRequestTest.{kafkaClientSaslMechanism, kafkaServerSaslMechanisms}
import kafka.test.annotation.{ClusterTemplate, Type}
import kafka.test.junit.ClusterTestExtensions
import kafka.test.{ClusterConfig, ClusterInstance}
import org.apache.kafka.common.test.api.{ClusterTemplate, Type, ClusterTestExtensions, ClusterConfig, ClusterInstance}
import org.apache.kafka.common.config.SaslConfigs
import org.apache.kafka.common.config.internals.BrokerSecurityConfigs
import org.apache.kafka.common.message.SaslHandshakeRequestData

6
core/src/test/scala/unit/kafka/server/ShareFetchAcknowledgeRequestTest.scala
View File

@ -16,9 +16,9 @@
*/
package kafka.server
import kafka.test.ClusterInstance
import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
import kafka.test.junit.ClusterTestExtensions
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.test.api.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
import org.apache.kafka.common.test.api.ClusterTestExtensions
import org.apache.kafka.common.message.ShareFetchResponseData.AcquiredRecords
import org.apache.kafka.common.message.{ShareAcknowledgeRequestData, ShareAcknowledgeResponseData, ShareFetchRequestData, ShareFetchResponseData}

6
core/src/test/scala/unit/kafka/server/ShareGroupDescribeRequestTest.scala
View File

@ -16,9 +16,9 @@
*/
package kafka.server
import kafka.test.ClusterInstance
import kafka.test.annotation._
import kafka.test.junit.ClusterTestExtensions
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.test.api._
import org.apache.kafka.common.test.api.ClusterTestExtensions
import kafka.utils.TestUtils
import org.apache.kafka.common.ShareGroupState
import org.apache.kafka.common.message.ShareGroupDescribeResponseData.DescribedGroup

8
core/src/test/scala/unit/kafka/server/ShareGroupHeartbeatRequestTest.scala
View File

@ -16,10 +16,10 @@
*/
package kafka.server
import kafka.test.ClusterInstance
import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, Type}
import kafka.test.junit.ClusterTestExtensions
import kafka.test.junit.RaftClusterInvocationContext.RaftClusterInstance
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.test.api.{ClusterConfigProperty, ClusterTest, Type}
import org.apache.kafka.common.test.api.ClusterTestExtensions
import org.apache.kafka.common.test.api.RaftClusterInvocationContext.RaftClusterInstance
import kafka.utils.TestUtils
import kafka.utils.TestUtils.waitForAllPartitionsMetadata
import org.apache.kafka.clients.admin.{Admin, NewPartitions}

6
core/src/test/scala/unit/kafka/server/SyncGroupRequestTest.scala
View File

@ -16,9 +16,9 @@
*/
package kafka.server
import kafka.test.ClusterInstance
import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
import kafka.test.junit.ClusterTestExtensions
import org.apache.kafka.common.test.api.ClusterInstance
import org.apache.kafka.common.test.api.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
import org.apache.kafka.common.test.api.ClusterTestExtensions
import kafka.utils.TestUtils
import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor
import org.apache.kafka.clients.consumer.internals.ConsumerProtocol

2
core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataPublisherTest.scala
View File

@ -24,7 +24,7 @@ import java.util.Properties
import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
import kafka.log.LogManager
import kafka.server.{BrokerServer, KafkaConfig, ReplicaManager}
import kafka.testkit.{KafkaClusterTestKit, TestKitNodes}
import org.apache.kafka.common.test.{KafkaClusterTestKit, TestKitNodes}
import kafka.utils.TestUtils
import org.apache.kafka.clients.admin.AlterConfigOp.OpType.SET
import org.apache.kafka.clients.admin.{Admin, AlterConfigOp, ConfigEntry, NewTopic}

4
settings.gradle
View File

@ -115,7 +115,9 @@ include 'clients',
'tools',
'tools:tools-api',
'transaction-coordinator',
'trogdor'
'trogdor',
'test-common',
'test-common:test-common-api'
project(":storage:api").name = "storage-api"
rootProject.name = 'kafka'

9
storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogSegmentLifecycleTest.java
View File

@ -16,14 +16,13 @@
*/
package org.apache.kafka.server.log.remote.metadata.storage;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterTest;
import kafka.test.annotation.ClusterTestDefaults;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestDefaults;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.server.log.remote.storage.RemoteLogMetadataManager;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentId;

9
storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerMultipleSubscriptionsTest.java
View File

@ -17,17 +17,16 @@
package org.apache.kafka.server.log.remote.metadata.storage;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterTest;
import kafka.test.annotation.ClusterTestDefaults;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestDefaults;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentId;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadata;

7
storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerRestartTest.java
View File

@ -16,15 +16,14 @@
*/
package org.apache.kafka.server.log.remote.metadata.storage;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterTest;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentId;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadata;

9
storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerTest.java
View File

@ -16,16 +16,15 @@
*/
package org.apache.kafka.server.log.remote.metadata.storage;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterTest;
import kafka.test.annotation.ClusterTestDefaults;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestDefaults;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentId;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadata;

9
storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogMetadataManagerTest.java
View File

@ -16,14 +16,13 @@
*/
package org.apache.kafka.server.log.remote.storage;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterTest;
import kafka.test.annotation.ClusterTestDefaults;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestDefaults;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.server.log.remote.metadata.storage.RemoteLogMetadataManagerTestUtils;
import org.apache.kafka.server.log.remote.metadata.storage.RemotePartitionMetadataStore;

7
streams/src/test/java/org/apache/kafka/streams/integration/utils/EmbeddedKafkaCluster.java
View File

@ -16,9 +16,6 @@
*/
package org.apache.kafka.streams.integration.utils;
import kafka.testkit.KafkaClusterTestKit;
import kafka.testkit.TestKitNodes;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.Config;
@ -40,6 +37,8 @@ import org.apache.kafka.common.errors.InvalidReplicationFactorException;
import org.apache.kafka.common.errors.TopicExistsException;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.test.KafkaClusterTestKit;
import org.apache.kafka.common.test.TestKitNodes;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
import org.apache.kafka.coordinator.transaction.TransactionLogConfig;
@ -77,7 +76,7 @@ import static org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZE
import static org.apache.kafka.common.utils.Utils.mkProperties;
/**
* Setup an embedded Kafka KRaft cluster for integration tests (using {@link kafka.testkit.KafkaClusterTestKit} internally) with the
* Setup an embedded Kafka KRaft cluster for integration tests (using {@link org.apache.kafka.common.test.KafkaClusterTestKit} internally) with the
* specified number of brokers and the specified broker properties.
* Additional Kafka client properties can also be supplied if required.
* This class also provides various utility methods to easily create Kafka topics, produce data, consume data etc.

8
core/src/test/java/kafka/testkit/KafkaClusterTestKit.java → test-common/src/main/java/org/apache/kafka/common/test/KafkaClusterTestKit.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package kafka.testkit;
package org.apache.kafka.common.test;
import kafka.raft.KafkaRaftManager;
import kafka.server.BrokerServer;
@ -43,9 +43,7 @@ import org.apache.kafka.server.common.ApiMessageAndVersion;
import org.apache.kafka.server.config.KRaftConfigs;
import org.apache.kafka.server.config.ServerConfigs;
import org.apache.kafka.server.fault.FaultHandler;
import org.apache.kafka.server.fault.MockFaultHandler;
import org.apache.kafka.storage.internals.log.CleanerConfig;
import org.apache.kafka.test.TestUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -77,7 +75,6 @@ import java.util.stream.Collectors;
import static org.apache.kafka.server.config.ReplicationConfigs.INTER_BROKER_LISTENER_NAME_CONFIG;
import static org.apache.kafka.server.config.ServerLogConfigs.LOG_DIRS_CONFIG;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@SuppressWarnings("deprecation") // Needed for Scala 2.12 compatibility
@ -569,7 +566,8 @@ public class KafkaClusterTestKit implements AutoCloseable {
active.set(controllerServer.controller());
}
}
assertNotNull(active.get(), "No active controller found");
if (active.get() == null)
throw new RuntimeException("Controller not active");
});
return active.get();
}

2
core/src/test/java/kafka/testkit/KafkaClusterThreadFactory.java → test-common/src/main/java/org/apache/kafka/common/test/KafkaClusterThreadFactory.java
View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package kafka.testkit;
package org.apache.kafka.common.test;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

4
core/src/test/java/kafka/test/MockController.java → test-common/src/main/java/org/apache/kafka/common/test/MockController.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package kafka.test;
package org.apache.kafka.common.test;
import org.apache.kafka.clients.admin.AlterConfigOp;
import org.apache.kafka.common.Uuid;
@ -390,6 +390,8 @@ public class MockController implements Controller {
case DELETE:
configs.getOrDefault(resource, Collections.emptyMap()).remove(key);
break;
default:
break;
}
}
}

72
test-common/src/main/java/org/apache/kafka/common/test/MockFaultHandler.java Normal file
View File

@ -0,0 +1,72 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.common.test;
import org.apache.kafka.server.fault.FaultHandler;
import org.apache.kafka.server.fault.FaultHandlerException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* This is a fault handler suitable for use in JUnit tests. It will store the result of the first
* call to handleFault that was made.
*/
public class MockFaultHandler implements FaultHandler {
private static final Logger log = LoggerFactory.getLogger(MockFaultHandler.class);
private final String name;
private FaultHandlerException firstException = null;
private boolean ignore = false;
public MockFaultHandler(String name) {
this.name = name;
}
@Override
public synchronized RuntimeException handleFault(String failureMessage, Throwable cause) {
if (cause == null) {
log.error("Encountered {} fault: {}", name, failureMessage);
} else {
log.error("Encountered {} fault: {}", name, failureMessage, cause);
}
FaultHandlerException e = (cause == null) ?
new FaultHandlerException(name + ": " + failureMessage) :
new FaultHandlerException(name + ": " + failureMessage +
": " + cause.getMessage(), cause);
if (firstException == null) {
firstException = e;
}
return firstException;
}
public synchronized void maybeRethrowFirstException() {
if (firstException != null && !ignore) {
throw firstException;
}
}
public synchronized FaultHandlerException firstException() {
return firstException;
}
public synchronized void setIgnore(boolean ignore) {
this.ignore = ignore;
}
}

35
test-common/src/main/java/org/apache/kafka/common/test/NoRetryException.java Normal file
View File

@ -0,0 +1,35 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.common.test;
/**
* This class can be used in the callback given to {@link TestUtils#retryOnExceptionWithTimeout(long, long, ValuelessCallable)}
* to indicate that a particular exception should not be retried. Instead the retry operation will
* be aborted immediately and the exception will be rethrown.
*/
public class NoRetryException extends RuntimeException {
private final Throwable cause;
public NoRetryException(Throwable cause) {
this.cause = cause;
}
@Override
public Throwable getCause() {
return this.cause;
}
}

27
test-common/src/main/java/org/apache/kafka/common/test/TestCondition.java Normal file
View File

@ -0,0 +1,27 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.common.test;
/**
* Interface to wrap actions that are required to wait until a condition is met
* for testing purposes. Note that this is not intended to do any assertions.
*/
@FunctionalInterface
public interface TestCondition {
boolean conditionMet() throws Exception;
}

2
core/src/test/java/kafka/testkit/TestKitNode.java → test-common/src/main/java/org/apache/kafka/common/test/TestKitNode.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package kafka.testkit;
package org.apache.kafka.common.test;
import org.apache.kafka.metadata.properties.MetaPropertiesEnsemble;

4
core/src/test/java/kafka/testkit/TestKitNodes.java → test-common/src/main/java/org/apache/kafka/common/test/TestKitNodes.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package kafka.testkit;
package org.apache.kafka.common.test;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.network.ListenerName;
@ -24,7 +24,6 @@ import org.apache.kafka.metadata.properties.MetaProperties;
import org.apache.kafka.metadata.properties.MetaPropertiesEnsemble;
import org.apache.kafka.metadata.properties.MetaPropertiesVersion;
import org.apache.kafka.server.common.MetadataVersion;
import org.apache.kafka.test.TestUtils;
import java.io.File;
import java.nio.file.Paths;
@ -41,6 +40,7 @@ import java.util.stream.IntStream;
import java.util.stream.Stream;
public class TestKitNodes {
public static final int CONTROLLER_ID_OFFSET = 3000;
public static final int BROKER_ID_OFFSET = 0;

258
test-common/src/main/java/org/apache/kafka/common/test/TestUtils.java Normal file
View File

@ -0,0 +1,258 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.common.test;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.utils.Exit;
import org.apache.kafka.common.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Random;
import java.util.function.Supplier;
/**
* Helper functions for writing unit tests
*/
public class TestUtils {
private static final Logger log = LoggerFactory.getLogger(TestUtils.class);
public static final File IO_TMP_DIR = new File(System.getProperty("java.io.tmpdir"));
public static final String LETTERS = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz";
public static final String DIGITS = "0123456789";
public static final String LETTERS_AND_DIGITS = LETTERS + DIGITS;
/* A consistent random number generator to make tests repeatable */
public static final Random SEEDED_RANDOM = new Random(192348092834L);
public static final Random RANDOM = new Random();
public static final long DEFAULT_POLL_INTERVAL_MS = 100;
public static final long DEFAULT_MAX_WAIT_MS = 15000;
/**
* Create an empty file in the default temporary-file directory, using the given prefix and suffix
* to generate its name.
* @throws IOException
*/
public static File tempFile(final String prefix, final String suffix) throws IOException {
final File file = Files.createTempFile(prefix, suffix).toFile();
file.deleteOnExit();
return file;
}
/**
* Create an empty file in the default temporary-file directory, using `kafka` as the prefix and `tmp` as the
* suffix to generate its name.
*/
public static File tempFile() throws IOException {
return tempFile("kafka", ".tmp");
}
/**
* Create a file with the given contents in the default temporary-file directory,
* using `kafka` as the prefix and `tmp` as the suffix to generate its name.
*/
public static File tempFile(final String contents) throws IOException {
final File file = tempFile();
Files.write(file.toPath(), contents.getBytes(StandardCharsets.UTF_8));
return file;
}
/**
* Create a temporary relative directory in the default temporary-file directory with the given prefix.
*
* @param prefix The prefix of the temporary directory, if null using "kafka-" as default prefix
*/
public static File tempDirectory(final String prefix) {
return tempDirectory(null, prefix);
}
/**
* Create a temporary relative directory in the default temporary-file directory with a
* prefix of "kafka-"
*
* @return the temporary directory just created.
*/
public static File tempDirectory() {
return tempDirectory(null);
}
/**
* Create a temporary directory under the given root directory.
* The root directory is removed on JVM exit if it doesn't already exist
* when this function is invoked.
*
* @param root path to create temporary directory under
* @return the temporary directory created within {@code root}
*/
public static File tempRelativeDir(String root) {
File rootFile = new File(root);
if (rootFile.mkdir()) {
rootFile.deleteOnExit();
}
return tempDirectory(rootFile.toPath(), null);
}
/**
* Create a temporary relative directory in the specified parent directory with the given prefix.
*
* @param parent The parent folder path name, if null using the default temporary-file directory
* @param prefix The prefix of the temporary directory, if null using "kafka-" as default prefix
*/
public static File tempDirectory(final Path parent, String prefix) {
final File file;
prefix = prefix == null ? "kafka-" : prefix;
try {
file = parent == null ?
Files.createTempDirectory(prefix).toFile() : Files.createTempDirectory(parent, prefix).toFile();
} catch (final IOException ex) {
throw new RuntimeException("Failed to create a temp dir", ex);
}
Exit.addShutdownHook("delete-temp-file-shutdown-hook", () -> {
try {
Utils.delete(file);
} catch (IOException e) {
log.error("Error deleting {}", file.getAbsolutePath(), e);
}
});
return file;
}
/**
* uses default value of 15 seconds for timeout
*/
public static void waitForCondition(final TestCondition testCondition, final String conditionDetails) throws InterruptedException {
waitForCondition(testCondition, DEFAULT_MAX_WAIT_MS, () -> conditionDetails);
}
/**
* uses default value of 15 seconds for timeout
*/
public static void waitForCondition(final TestCondition testCondition, final Supplier<String> conditionDetailsSupplier) throws InterruptedException {
waitForCondition(testCondition, DEFAULT_MAX_WAIT_MS, conditionDetailsSupplier);
}
/**
* Wait for condition to be met for at most {@code maxWaitMs} and throw assertion failure otherwise.
* This should be used instead of {@code Thread.sleep} whenever possible as it allows a longer timeout to be used
* without unnecessarily increasing test time (as the condition is checked frequently). The longer timeout is needed to
* avoid transient failures due to slow or overloaded machines.
*/
public static void waitForCondition(final TestCondition testCondition, final long maxWaitMs, String conditionDetails) throws InterruptedException {
waitForCondition(testCondition, maxWaitMs, () -> conditionDetails);
}
/**
* Wait for condition to be met for at most {@code maxWaitMs} and throw assertion failure otherwise.
* This should be used instead of {@code Thread.sleep} whenever possible as it allows a longer timeout to be used
* without unnecessarily increasing test time (as the condition is checked frequently). The longer timeout is needed to
* avoid transient failures due to slow or overloaded machines.
*/
public static void waitForCondition(final TestCondition testCondition, final long maxWaitMs, Supplier<String> conditionDetailsSupplier) throws InterruptedException {
waitForCondition(testCondition, maxWaitMs, DEFAULT_POLL_INTERVAL_MS, conditionDetailsSupplier);
}
/**
* Wait for condition to be met for at most {@code maxWaitMs} with a polling interval of {@code pollIntervalMs}
* and throw assertion failure otherwise. This should be used instead of {@code Thread.sleep} whenever possible
* as it allows a longer timeout to be used without unnecessarily increasing test time (as the condition is
* checked frequently). The longer timeout is needed to avoid transient failures due to slow or overloaded
* machines.
*/
public static void waitForCondition(
final TestCondition testCondition,
final long maxWaitMs,
final long pollIntervalMs,
Supplier<String> conditionDetailsSupplier
) throws InterruptedException {
retryOnExceptionWithTimeout(maxWaitMs, pollIntervalMs, () -> {
String conditionDetailsSupplied = conditionDetailsSupplier != null ? conditionDetailsSupplier.get() : null;
String conditionDetails = conditionDetailsSupplied != null ? conditionDetailsSupplied : "";
if (!testCondition.conditionMet())
throw new TimeoutException("Condition not met within timeout " + maxWaitMs + ". " + conditionDetails);
});
}
/**
* Wait for the given runnable to complete successfully, i.e. throw now {@link Exception}s or
* {@link AssertionError}s, or for the given timeout to expire. If the timeout expires then the
* last exception or assertion failure will be thrown thus providing context for the failure.
*
* @param timeoutMs the total time in milliseconds to wait for {@code runnable} to complete successfully.
* @param runnable the code to attempt to execute successfully.
* @throws InterruptedException if the current thread is interrupted while waiting for {@code runnable} to complete successfully.
*/
public static void retryOnExceptionWithTimeout(final long timeoutMs,
final ValuelessCallable runnable) throws InterruptedException {
retryOnExceptionWithTimeout(timeoutMs, DEFAULT_POLL_INTERVAL_MS, runnable);
}
/**
* Wait for the given runnable to complete successfully, i.e. throw now {@link Exception}s or
* {@link AssertionError}s, or for the default timeout to expire. If the timeout expires then the
* last exception or assertion failure will be thrown thus providing context for the failure.
*
* @param runnable the code to attempt to execute successfully.
* @throws InterruptedException if the current thread is interrupted while waiting for {@code runnable} to complete successfully.
*/
public static void retryOnExceptionWithTimeout(final ValuelessCallable runnable) throws InterruptedException {
retryOnExceptionWithTimeout(DEFAULT_MAX_WAIT_MS, DEFAULT_POLL_INTERVAL_MS, runnable);
}
/**
* Wait for the given runnable to complete successfully, i.e. throw now {@link Exception}s or
* {@link AssertionError}s, or for the given timeout to expire. If the timeout expires then the
* last exception or assertion failure will be thrown thus providing context for the failure.
*
* @param timeoutMs the total time in milliseconds to wait for {@code runnable} to complete successfully.
* @param pollIntervalMs the interval in milliseconds to wait between invoking {@code runnable}.
* @param runnable the code to attempt to execute successfully.
* @throws InterruptedException if the current thread is interrupted while waiting for {@code runnable} to complete successfully.
*/
public static void retryOnExceptionWithTimeout(final long timeoutMs,
final long pollIntervalMs,
final ValuelessCallable runnable) throws InterruptedException {
final long expectedEnd = System.currentTimeMillis() + timeoutMs;
while (true) {
try {
runnable.call();
return;
} catch (final NoRetryException e) {
throw e;
} catch (final AssertionError t) {
if (expectedEnd <= System.currentTimeMillis()) {
throw t;
}
} catch (final Exception e) {
if (expectedEnd <= System.currentTimeMillis()) {
throw new AssertionError(String.format("Assertion failed with an exception after %s ms", timeoutMs), e);
}
}
Thread.sleep(Math.min(pollIntervalMs, timeoutMs));
}
}
}

25
test-common/src/main/java/org/apache/kafka/common/test/ValuelessCallable.java Normal file
View File

@ -0,0 +1,25 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.common.test;
/**
* Like a {@link Runnable} that allows exceptions to be thrown or a {@link java.util.concurrent.Callable}
* that does not return a value.
*/
public interface ValuelessCallable {
void call() throws Exception;
}

2
core/src/test/java/kafka/testkit/KafkaClusterTestKitTest.java → test-common/src/test/java/org/apache/kafka/common/test/KafkaClusterTestKitTest.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package kafka.testkit;
package org.apache.kafka.common.test;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;

2
core/src/test/java/kafka/testkit/KafkaClusterThreadFactoryTest.java → test-common/src/test/java/org/apache/kafka/common/test/KafkaClusterThreadFactoryTest.java
View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package kafka.testkit;
package org.apache.kafka.common.test;
import org.junit.jupiter.api.Test;

2
core/src/test/java/kafka/test/annotation/AutoStart.java → test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/AutoStart.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package kafka.test.annotation;
package org.apache.kafka.common.test.api;
public enum AutoStart {
YES,

4
core/src/test/java/kafka/test/ClusterConfig.java → test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/ClusterConfig.java
View File

@ -15,9 +15,7 @@
* limitations under the License.
*/
package kafka.test;
import kafka.test.annotation.Type;
package org.apache.kafka.common.test.api;
import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.server.common.Features;

2
core/src/test/java/kafka/test/annotation/ClusterConfigProperty.java → test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/ClusterConfigProperty.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package kafka.test.annotation;
package org.apache.kafka.common.test.api;
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;

2
core/src/test/java/kafka/test/annotation/ClusterFeature.java → test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/ClusterFeature.java
View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package kafka.test.annotation;
package org.apache.kafka.common.test.api;
import org.apache.kafka.server.common.Features;

2
core/src/test/java/kafka/test/ClusterGenerator.java → test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/ClusterGenerator.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package kafka.test;
package org.apache.kafka.common.test.api;
import java.util.function.Consumer;

6
core/src/test/java/kafka/test/ClusterInstance.java → test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/ClusterInstance.java
View File

@ -15,14 +15,12 @@
* limitations under the License.
*/
package kafka.test;
package org.apache.kafka.common.test.api;
import kafka.network.SocketServer;
import kafka.server.BrokerServer;
import kafka.server.ControllerServer;
import kafka.server.KafkaBroker;
import kafka.test.annotation.ClusterTest;
import kafka.test.annotation.Type;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.NewTopic;
@ -30,8 +28,8 @@ import org.apache.kafka.clients.consumer.GroupProtocol;
import org.apache.kafka.common.acl.AccessControlEntry;
import org.apache.kafka.common.acl.AclBindingFilter;
import org.apache.kafka.common.network.ListenerName;
import org.apache.kafka.common.test.TestUtils;
import org.apache.kafka.server.authorizer.Authorizer;
import org.apache.kafka.test.TestUtils;
import java.util.ArrayList;
import java.util.Collection;

3
core/src/test/java/kafka/test/junit/ClusterInstanceParameterResolver.java → test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/ClusterInstanceParameterResolver.java
View File

@ -15,9 +15,8 @@
* limitations under the License.
*/
package kafka.test.junit;
package org.apache.kafka.common.test.api;
import kafka.test.ClusterInstance;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtensionContext;

5
core/src/test/java/kafka/test/annotation/ClusterTemplate.java → test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/ClusterTemplate.java
View File

@ -15,10 +15,7 @@
* limitations under the License.
*/
package kafka.test.annotation;
import kafka.test.ClusterConfig;
import kafka.test.ClusterGenerator;
package org.apache.kafka.common.test.api;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.TestTemplate;

2
core/src/test/java/kafka/test/annotation/ClusterTest.java → test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/ClusterTest.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package kafka.test.annotation;
package org.apache.kafka.common.test.api;
import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.server.common.MetadataVersion;

3
core/src/test/java/kafka/test/annotation/ClusterTestDefaults.java → test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/ClusterTestDefaults.java
View File

@ -15,9 +15,8 @@
* limitations under the License.
*/
package kafka.test.annotation;
package org.apache.kafka.common.test.api;
import kafka.test.junit.ClusterTestExtensions;
import java.lang.annotation.Documented;
import java.lang.annotation.Retention;

13
core/src/test/java/kafka/test/junit/ClusterTestExtensions.java → test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/ClusterTestExtensions.java
View File

@ -14,18 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package kafka.test.junit;
import kafka.test.ClusterConfig;
import kafka.test.annotation.AutoStart;
import kafka.test.annotation.ClusterConfigProperty;
import kafka.test.annotation.ClusterFeature;
import kafka.test.annotation.ClusterTemplate;
import kafka.test.annotation.ClusterTest;
import kafka.test.annotation.ClusterTestDefaults;
import kafka.test.annotation.ClusterTests;
import kafka.test.annotation.Type;
package org.apache.kafka.common.test.api;
import org.apache.kafka.server.common.Features;
import org.apache.kafka.server.util.timer.SystemTimer;

3
core/src/test/java/kafka/test/annotation/ClusterTests.java → test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/ClusterTests.java
View File

@ -14,8 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package kafka.test.annotation;
package org.apache.kafka.common.test.api;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.TestTemplate;

3
core/src/test/java/kafka/test/junit/DetectThreadLeak.java → test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/DetectThreadLeak.java
View File

@ -14,8 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package kafka.test.junit;
package org.apache.kafka.common.test.api;
import java.util.List;
import java.util.Set;

2
core/src/test/java/kafka/test/junit/README.md → test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/README.md
View File

@ -76,7 +76,7 @@ a JUnit extension called `ClusterTestExtensions` which knows how to process thes
invocations. Test classes that wish to make use of these annotations need to explicitly register this extension:
```scala
import kafka.test.junit.ClusterTestExtensions
import org.apache.kafka.common.test.api.ClusterTestExtensions
@ExtendWith(value = Array(classOf[ClusterTestExtensions]))
class ApiVersionsRequestTest {

28
core/src/test/java/kafka/test/junit/RaftClusterInvocationContext.java → test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/RaftClusterInvocationContext.java
View File

@ -14,22 +14,18 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package kafka.test.junit;
package org.apache.kafka.common.test.api;
import kafka.network.SocketServer;
import kafka.server.BrokerServer;
import kafka.server.ControllerServer;
import kafka.server.KafkaBroker;
import kafka.test.ClusterConfig;
import kafka.test.ClusterInstance;
import kafka.test.annotation.Type;
import kafka.testkit.KafkaClusterTestKit;
import kafka.testkit.TestKitNodes;
import kafka.zk.EmbeddedZookeeper;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.common.network.ListenerName;
import org.apache.kafka.common.test.KafkaClusterTestKit;
import org.apache.kafka.common.test.TestKitNodes;
import org.apache.kafka.common.test.TestUtils;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.metadata.BrokerState;
import org.apache.kafka.metadata.bootstrap.BootstrapMetadata;
@ -110,7 +106,6 @@ public class RaftClusterInvocationContext implements TestTemplateInvocationConte
final AtomicBoolean stopped = new AtomicBoolean(false);
final AtomicBoolean formated = new AtomicBoolean(false);
private final ConcurrentLinkedQueue<Admin> admins = new ConcurrentLinkedQueue<>();
private EmbeddedZookeeper embeddedZookeeper;
private KafkaClusterTestKit clusterTestKit;
private final boolean isCombined;
@ -189,13 +184,10 @@ public class RaftClusterInvocationContext implements TestTemplateInvocationConte
format();
if (started.compareAndSet(false, true)) {
clusterTestKit.startup();
kafka.utils.TestUtils.waitUntilTrue(
TestUtils.waitForCondition(
() -> this.clusterTestKit.brokers().values().stream().allMatch(
brokers -> brokers.brokerState() == BrokerState.RUNNING
),
() -> "Broker never made it to RUNNING state.",
org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS,
100L);
), "Broker never made it to RUNNING state.");
}
} catch (Exception e) {
throw new RuntimeException("Failed to start Raft server", e);
@ -208,9 +200,6 @@ public class RaftClusterInvocationContext implements TestTemplateInvocationConte
admins.forEach(admin -> Utils.closeQuietly(admin, "admin"));
admins.clear();
Utils.closeQuietly(clusterTestKit, "cluster");
if (embeddedZookeeper != null) {
Utils.closeQuietly(embeddedZookeeper, "zk");
}
}
}
@ -294,11 +283,6 @@ public class RaftClusterInvocationContext implements TestTemplateInvocationConte
.setPerServerProperties(clusterConfig.perServerOverrideProperties())
.setNumControllerNodes(clusterConfig.numControllers()).build();
KafkaClusterTestKit.Builder builder = new KafkaClusterTestKit.Builder(nodes);
if (Boolean.parseBoolean(clusterConfig.serverProperties()
.getOrDefault("zookeeper.metadata.migration.enable", "false"))) {
this.embeddedZookeeper = new EmbeddedZookeeper();
builder.setConfigProp("zookeeper.connect", String.format("localhost:%d", embeddedZookeeper.port()));
}
// Copy properties into the TestKit builder
clusterConfig.serverProperties().forEach(builder::setConfigProp);
// KAFKA-12512 need to pass security protocol and listener name here

6
core/src/test/java/kafka/test/annotation/Type.java → test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/Type.java
View File

@ -14,11 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package kafka.test.annotation;
import kafka.test.ClusterConfig;
import kafka.test.junit.RaftClusterInvocationContext;
package org.apache.kafka.common.test.api;
import org.junit.jupiter.api.extension.TestTemplateInvocationContext;

6
core/src/test/java/kafka/test/ClusterConfigTest.java → test-common/test-common-api/src/test/java/org/apache/kafka/common/test/api/ClusterConfigTest.java
View File

@ -15,13 +15,11 @@
* limitations under the License.
*/
package kafka.test;
import kafka.test.annotation.Type;
package org.apache.kafka.common.test.api;
import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.common.test.TestUtils;
import org.apache.kafka.server.common.MetadataVersion;
import org.apache.kafka.test.TestUtils;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;

11
core/src/test/java/kafka/test/ClusterTestExtensionsTest.java → test-common/test-common-api/src/test/java/org/apache/kafka/common/test/api/ClusterTestExtensionsTest.java
View File

@ -15,16 +15,7 @@
* limitations under the License.
*/
package kafka.test;
import kafka.test.annotation.AutoStart;
import kafka.test.annotation.ClusterConfigProperty;
import kafka.test.annotation.ClusterTemplate;
import kafka.test.annotation.ClusterTest;
import kafka.test.annotation.ClusterTestDefaults;
import kafka.test.annotation.ClusterTests;
import kafka.test.annotation.Type;
import kafka.test.junit.ClusterTestExtensions;
package org.apache.kafka.common.test.api;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;

5
core/src/test/java/kafka/test/junit/ClusterTestExtensionsUnitTest.java → test-common/test-common-api/src/test/java/org/apache/kafka/common/test/api/ClusterTestExtensionsUnitTest.java
View File

@ -15,10 +15,7 @@
* limitations under the License.
*/
package kafka.test.junit;
import kafka.test.ClusterConfig;
import kafka.test.annotation.ClusterTemplate;
package org.apache.kafka.common.test.api;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;

2
core/src/test/java/kafka/test/junit/DetectThreadLeakTest.java → test-common/test-common-api/src/test/java/org/apache/kafka/common/test/api/DetectThreadLeakTest.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package kafka.test.junit;
package org.apache.kafka.common.test.api;
import org.junit.jupiter.api.Test;

11
tools/src/test/java/org/apache/kafka/tools/BrokerApiVersionsCommandTest.java
View File

@ -16,12 +16,6 @@
*/
package org.apache.kafka.tools;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterConfigProperty;
import kafka.test.annotation.ClusterTest;
import kafka.test.annotation.ClusterTestDefaults;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.NodeApiVersions;
import org.apache.kafka.common.KafkaFuture;
@ -30,6 +24,11 @@ import org.apache.kafka.common.message.ApiMessageType;
import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersion;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.requests.ApiVersionsResponse;
import org.apache.kafka.common.test.api.ClusterConfigProperty;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestDefaults;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.server.config.ServerConfigs;
import org.junit.jupiter.api.extension.ExtendWith;

9
tools/src/test/java/org/apache/kafka/tools/ClusterToolTest.java
View File

@ -16,14 +16,13 @@
*/
package org.apache.kafka.tools;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterTest;
import kafka.test.annotation.Type;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.MockAdminClient;
import org.apache.kafka.common.errors.UnsupportedEndpointTypeException;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.common.test.api.Type;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;

7
tools/src/test/java/org/apache/kafka/tools/DeleteRecordsCommandTest.java
View File

@ -16,10 +16,6 @@
*/
package org.apache.kafka.tools;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterTest;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
@ -28,6 +24,9 @@ import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.server.common.AdminCommandFailedException;
import org.apache.kafka.server.common.AdminOperationException;

9
tools/src/test/java/org/apache/kafka/tools/FeatureCommandTest.java
View File

@ -16,12 +16,11 @@
*/
package org.apache.kafka.tools;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterTest;
import kafka.test.annotation.Type;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.clients.admin.MockAdminClient;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.common.test.api.Type;
import org.apache.kafka.server.common.Features;
import org.apache.kafka.server.common.MetadataVersion;

19
tools/src/test/java/org/apache/kafka/tools/GetOffsetShellTest.java
View File

@ -17,14 +17,6 @@
package org.apache.kafka.tools;
import kafka.test.ClusterConfig;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterConfigProperty;
import kafka.test.annotation.ClusterTemplate;
import kafka.test.annotation.ClusterTest;
import kafka.test.annotation.ClusterTestDefaults;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.NewTopic;
@ -36,6 +28,13 @@ import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.config.TopicConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.test.api.ClusterConfig;
import org.apache.kafka.common.test.api.ClusterConfigProperty;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTemplate;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestDefaults;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.common.utils.AppInfoParser;
import org.apache.kafka.common.utils.Exit;
import org.apache.kafka.server.config.ServerLogConfigs;
@ -60,8 +59,8 @@ import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;
import static kafka.test.annotation.Type.CO_KRAFT;
import static kafka.test.annotation.Type.KRAFT;
import static org.apache.kafka.common.test.api.Type.CO_KRAFT;
import static org.apache.kafka.common.test.api.Type.KRAFT;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;

10
tools/src/test/java/org/apache/kafka/tools/LeaderElectionCommandTest.java
View File

@ -16,11 +16,6 @@
*/
package org.apache.kafka.tools;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterConfigProperty;
import kafka.test.annotation.ClusterTest;
import kafka.test.annotation.ClusterTestDefaults;
import kafka.test.junit.ClusterTestExtensions;
import kafka.utils.TestUtils;
import org.apache.kafka.clients.admin.Admin;
@ -29,6 +24,11 @@ import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
import org.apache.kafka.common.test.api.ClusterConfigProperty;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestDefaults;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.server.common.AdminCommandFailedException;
import org.junit.jupiter.api.extension.ExtendWith;

7
tools/src/test/java/org/apache/kafka/tools/LogDirsCommandTest.java
View File

@ -16,10 +16,6 @@
*/
package org.apache.kafka.tools;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterTest;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.LogDirDescription;
@ -27,6 +23,9 @@ import org.apache.kafka.clients.admin.MockAdminClient;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;

11
tools/src/test/java/org/apache/kafka/tools/MetadataQuorumCommandTest.java
View File

@ -16,13 +16,12 @@
*/
package org.apache.kafka.tools;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterTest;
import kafka.test.annotation.ClusterTests;
import kafka.test.annotation.Type;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.common.test.api.ClusterTests;
import org.apache.kafka.common.test.api.Type;
import org.apache.kafka.test.TestUtils;
import org.junit.jupiter.api.Test;

11
tools/src/test/java/org/apache/kafka/tools/TopicCommandTest.java
View File

@ -17,12 +17,6 @@
package org.apache.kafka.tools;
import kafka.test.ClusterConfig;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterTemplate;
import kafka.test.annotation.Type;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientTestUtils;
import org.apache.kafka.clients.admin.Config;
@ -55,6 +49,11 @@ import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.FetchRequest;
import org.apache.kafka.common.requests.MetadataResponse;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.test.api.ClusterConfig;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTemplate;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.common.test.api.Type;
import org.apache.kafka.common.utils.Exit;
import org.apache.kafka.server.common.AdminCommandFailedException;
import org.apache.kafka.server.common.AdminOperationException;

7
tools/src/test/java/org/apache/kafka/tools/consumer/ConsoleConsumerTest.java
View File

@ -16,10 +16,6 @@
*/
package org.apache.kafka.tools.consumer;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterTest;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.Consumer;
@ -38,6 +34,9 @@ import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.internals.Topic;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.coordinator.group.generated.GroupMetadataKey;
import org.apache.kafka.coordinator.group.generated.GroupMetadataKeyJsonConverter;

5
tools/src/test/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommandTestUtils.java
View File

@ -17,11 +17,10 @@
package org.apache.kafka.tools.consumer.group;
import kafka.test.ClusterConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.test.api.ClusterConfig;
import org.apache.kafka.common.utils.Utils;
import java.time.Duration;
@ -38,7 +37,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.function.Supplier;
import static kafka.test.annotation.Type.CO_KRAFT;
import static org.apache.kafka.common.test.api.Type.CO_KRAFT;
import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.CONSUMER_GROUP_HEARTBEAT_INTERVAL_MS_CONFIG;
import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.CONSUMER_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG;
import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG;

9
tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteConsumerGroupsTest.java
View File

@ -16,11 +16,6 @@
*/
package org.apache.kafka.tools.consumer.group;
import kafka.test.ClusterConfig;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterTemplate;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.consumer.GroupProtocol;
import org.apache.kafka.clients.consumer.KafkaConsumer;
@ -30,6 +25,10 @@ import org.apache.kafka.common.errors.GroupIdNotFoundException;
import org.apache.kafka.common.errors.GroupNotEmptyException;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.test.api.ClusterConfig;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTemplate;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.test.TestUtils;
import org.apache.kafka.tools.ToolsTestUtils;

9
tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteOffsetsConsumerGroupCommandIntegrationTest.java
View File

@ -16,11 +16,6 @@
*/
package org.apache.kafka.tools.consumer.group;
import kafka.test.ClusterConfig;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterTemplate;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
@ -37,6 +32,10 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.test.api.ClusterConfig;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTemplate;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
import org.junit.jupiter.api.Assertions;

9
tools/src/test/java/org/apache/kafka/tools/consumer/group/DescribeConsumerGroupTest.java
View File

@ -16,11 +16,6 @@
*/
package org.apache.kafka.tools.consumer.group;
import kafka.test.ClusterConfig;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterTemplate;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
@ -35,6 +30,10 @@ import org.apache.kafka.common.ConsumerGroupState;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.test.api.ClusterConfig;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTemplate;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.common.utils.AppInfoParser;
import org.apache.kafka.common.utils.Exit;
import org.apache.kafka.test.TestUtils;

9
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java
View File

@ -16,11 +16,6 @@
*/
package org.apache.kafka.tools.consumer.group;
import kafka.test.ClusterConfig;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterTemplate;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
@ -33,6 +28,10 @@ import org.apache.kafka.common.ConsumerGroupState;
import org.apache.kafka.common.GroupType;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.test.api.ClusterConfig;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTemplate;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.test.TestUtils;
import org.apache.kafka.tools.ToolsTestUtils;

9
tools/src/test/java/org/apache/kafka/tools/consumer/group/ResetConsumerGroupOffsetTest.java
View File

@ -16,11 +16,6 @@
*/
package org.apache.kafka.tools.consumer.group;
import kafka.test.ClusterConfig;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterTemplate;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.GroupProtocol;
@ -34,6 +29,10 @@ import org.apache.kafka.common.ConsumerGroupState;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.test.api.ClusterConfig;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTemplate;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.test.TestUtils;
import org.junit.jupiter.api.extension.ExtendWith;

15
tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommandTest.java
View File

@ -16,14 +16,6 @@
*/
package org.apache.kafka.tools.reassign;
import kafka.test.ClusterInstance;
import kafka.test.annotation.ClusterConfigProperty;
import kafka.test.annotation.ClusterTest;
import kafka.test.annotation.ClusterTestDefaults;
import kafka.test.annotation.ClusterTests;
import kafka.test.annotation.Type;
import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AlterConfigOp;
@ -47,6 +39,13 @@ import org.apache.kafka.common.TopicPartitionReplica;
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.test.api.ClusterConfigProperty;
import org.apache.kafka.common.test.api.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestDefaults;
import org.apache.kafka.common.test.api.ClusterTestExtensions;
import org.apache.kafka.common.test.api.ClusterTests;
import org.apache.kafka.common.test.api.Type;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.server.config.QuotaConfigs;
import org.apache.kafka.test.TestUtils;