Skip to content

KAFKA-12284: increase request timeout to make tests reliable #10547

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
chia7712 merged 4 commits into from
Apr 28, 2021
Merged

KAFKA-12284: increase request timeout to make tests reliable #10547

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
5c5b584
KAFKA-12284: increase request timeout to make tests reliable
showuon Apr 15, 2021
9f6bc9f
Merge branch 'trunk' of https://github.com/apache/kafka into KAFKA-12284
showuon Apr 22, 2021
d0c7bcc
KAFKA-12284: refactor code
showuon Apr 28, 2021
02b401a
KAFKA-12284: refactor
showuon Apr 28, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 23 additions & 7 deletions ...java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
package org.apache.kafka.connect.mirror.integration;

import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.Config;
import org.apache.kafka.clients.admin.DescribeConfigsResult;
import org.apache.kafka.clients.consumer.Consumer;
Expand Down Expand Up @@ -83,7 +84,8 @@ public abstract class MirrorConnectorsIntegrationBaseTest {
private static final int CHECKPOINT_DURATION_MS = 20_000;
private static final int RECORD_CONSUME_DURATION_MS = 20_000;
private static final int OFFSET_SYNC_DURATION_MS = 30_000;
private static final int TOPIC_SYNC_DURATION_MS = 30_000;
private static final int TOPIC_SYNC_DURATION_MS = 60_000;
private static final int REQUEST_TIMEOUT_DURATION_MS = 60_000;
private static final int NUM_WORKERS = 3;
private static final Duration CONSUMER_POLL_TIMEOUT_MS = Duration.ofMillis(500);
protected static final String PRIMARY_CLUSTER_ALIAS = "primary";
Expand Down Expand Up @@ -165,11 +167,19 @@ public void startClusters() throws Exception {
primary.start();
primary.assertions().assertAtLeastNumWorkersAreUp(NUM_WORKERS,
"Workers of " + PRIMARY_CLUSTER_ALIAS + "-connect-cluster did not start in time.");

waitForTopicCreated(primary, "mm2-status.backup.internal");
waitForTopicCreated(primary, "mm2-offsets.backup.internal");
waitForTopicCreated(primary, "mm2-configs.backup.internal");

backup.start();
backup.assertions().assertAtLeastNumWorkersAreUp(NUM_WORKERS,
"Workers of " + BACKUP_CLUSTER_ALIAS + "-connect-cluster did not start in time.");

waitForTopicCreated(backup, "mm2-status.primary.internal");
waitForTopicCreated(backup, "mm2-offsets.primary.internal");
waitForTopicCreated(backup, "mm2-configs.primary.internal");

createTopics();

warmUpConsumer(Collections.singletonMap("group.id", "consumer-group-dummy"));
Expand Down Expand Up @@ -602,13 +612,19 @@ private static Map<String, String> basicMM2Config() {
private void createTopics() {
// to verify topic config will be sync-ed across clusters
Map<String, String> topicConfig = Collections.singletonMap(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT);
Map<String, String> emptyMap = Collections.emptyMap();

// increase admin client request timeout value to make the tests reliable.
Properties adminClientConfig = new Properties();
adminClientConfig.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, REQUEST_TIMEOUT_DURATION_MS);

// create these topics before starting the connectors so we don't need to wait for discovery
primary.kafka().createTopic("test-topic-1", NUM_PARTITIONS, 1, topicConfig);
primary.kafka().createTopic("backup.test-topic-1", 1);
primary.kafka().createTopic("heartbeats", 1);
backup.kafka().createTopic("test-topic-1", NUM_PARTITIONS);
backup.kafka().createTopic("primary.test-topic-1", 1);
backup.kafka().createTopic("heartbeats", 1);
primary.kafka().createTopic("test-topic-1", NUM_PARTITIONS, 1, topicConfig, adminClientConfig);
primary.kafka().createTopic("backup.test-topic-1", 1, 1, emptyMap, adminClientConfig);
primary.kafka().createTopic("heartbeats", 1, 1, emptyMap, adminClientConfig);
backup.kafka().createTopic("test-topic-1", NUM_PARTITIONS, 1, emptyMap, adminClientConfig);
backup.kafka().createTopic("primary.test-topic-1", 1, 1, emptyMap, adminClientConfig);
backup.kafka().createTopic("heartbeats", 1, 1, emptyMap, adminClientConfig);
}

/*
Expand Down
31 changes: 22 additions & 9 deletions ...ct/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -337,18 +337,28 @@ public void createTopic(String topic) {
* @param topic The name of the topic.
*/
public void createTopic(String topic, int partitions) {
createTopic(topic, partitions, 1, new HashMap<>());
createTopic(topic, partitions, 1, Collections.emptyMap());
}

/**
* Create a Kafka topic with the given parameters.
* Create a Kafka topic with given partition, replication factor, and topic config.
*
* @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.
* @param topic The name of the topic.
*/
public void createTopic(String topic, int partitions, int replication, Map<String, String> topicConfig) {
createTopic(topic, partitions, replication, topicConfig, new Properties());
}

/**
* 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.
* @param adminClientConfig Additional admin client configuration settings.
*/
public void createTopic(String topic, int partitions, int replication, Map<String, String> topicConfig, Properties adminClientConfig) {
if (replication > brokers.length) {
throw new InvalidReplicationFactorException("Insufficient brokers ("
+ brokers.length + ") for desired replication (" + replication + ")");
Expand All @@ -359,7 +369,7 @@ public void createTopic(String topic, int partitions, int replication, Map<Strin
final NewTopic newTopic = new NewTopic(topic, partitions, (short) replication);
newTopic.configs(topicConfig);

try (final Admin adminClient = createAdminClient()) {
try (final Admin adminClient = createAdminClient(adminClientConfig)) {
adminClient.createTopics(Collections.singletonList(newTopic)).all().get();
} catch (final InterruptedException | ExecutionException e) {
throw new RuntimeException(e);
Expand All @@ -383,8 +393,7 @@ public void produce(String topic, Integer partition, String key, String value) {
}
}

public Admin createAdminClient() {
final Properties adminClientConfig = new Properties();
public Admin createAdminClient(Properties adminClientConfig) {
adminClientConfig.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers());
final Object listeners = brokerConfig.get(KafkaConfig$.MODULE$.ListenersProp());
if (listeners != null && listeners.toString().contains("SSL")) {
Expand All @@ -395,6 +404,10 @@ public Admin createAdminClient() {
return Admin.create(adminClientConfig);
}

public Admin createAdminClient() {
return createAdminClient(new Properties());
}

/**
* Consume at least n records in a given duration or throw an exception.
*
Expand Down