...
| Code Block |
|---|
| language | java |
|---|
| title | EmbeddedCluster.java |
|---|
|
package org.apache.kafka.testkit;
public class EmbeddedCluster extends ExternalResource {
/** Create a Kafka cluster with the requested number of brokers and with the provided configs */
public static EmbeddedCluster create(int numBrokers, Map<String, String> configs) { ... }
/** Create a Kafka cluster with one broker per passed map of configs. */
public EmbeddedCluster(Collection<Map<String, String>> configs) {
this.configs = configs;
}
public void start() { ... }
public TestAdmin admin() { ... }
public boolean stopBroker(int brokerId) { ... }
public void stop() { ... }
public List<String> bootstrapServers(String listenerName) { ... }
public String zooKeeperConnectString() { ... }
public List<EmbeddedBroker> brokers() { ... }
public Integer leader(TopicPartition topicPartition) { ... }
}
|
| Code Block |
|---|
| language | java |
|---|
| title | TestAdmin.java |
|---|
|
package org.apache.kafka.testkit;
public class TestAdmin {
/**
* Create a Kafka topic with the given parameters.
*
* @param topic The name of the topic.
* @param partitions The number of partitions for this topic.
* @param replication The replication factor for (the partitions of) this topic.
*/
public void createTopic(final String topic, final int partitions, final int replication) { ... }
/**
* Create a Kafka topic with the given parameters.
*
* @param topic The name of the topic.
* @param partitions The number of partitions for this topic.
* @param replication The replication factor for (partitions of) this topic.
* @param topicConfig Additional topic-level configuration settings.
*/
public void createTopic(final String topic,
final int partitions,
final int replication,
final Properties topicConfig) { ... }
public void deleteTopic(final String topic) { ... }
}
|
| Code Block |
|---|
| language | java |
|---|
| title | TimeUtils.java |
|---|
|
package org.apache.kafka.testkit;
public class TimeUtils {
/**
* 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 BooleanSupplier testCondition, final long maxWaitMs, String conditionDetails) { ... }
}
|
Proposed Changes
We will introduce a new artifact called kafka-testkit that will contain the Java classes for the test library. This module will depend on the clients and core kafka artifacts, but it won't depend on any of the Kafka test artifacts. As such, users won't have our tests on their classpath.
...