fix unstable ResetIntegrationTest
Reviewers: Guozhang Wang Damian Guy
@@ -24,6 +24,7 @@ import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.SslConfigs;
import org.apache.kafka.common.config.types.Password;
+import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.LongSerializer;
import org.apache.kafka.common.serialization.Serdes;
@@ -167,11 +168,24 @@ public abstract class AbstractResetIntegrationTest {
private final static Properties RESULT_CONSUMER_CONFIG = new Properties();
void prepareTest() throws Exception {
- cluster.deleteAndRecreateTopics(INPUT_TOPIC, OUTPUT_TOPIC, OUTPUT_TOPIC_2, OUTPUT_TOPIC_2_RERUN);
-
prepareConfigs();
prepareEnvironment();
+ // busy wait until cluster (ie, ConsumerGroupCoordinator) is available
+ while (true) {
+ Thread.sleep(50);
+
+ try {
+ TestUtils.waitForCondition(consumerGroupInactiveCondition, TIMEOUT_MULTIPLIER * CLEANUP_CONSUMER_TIMEOUT,
+ "Test consumer group active even after waiting " + (TIMEOUT_MULTIPLIER * CLEANUP_CONSUMER_TIMEOUT) + " ms.");
+ } catch (final TimeoutException e) {
+ continue;
+ }
+ break;
+ }
+
+ cluster.deleteAndRecreateTopics(INPUT_TOPIC, OUTPUT_TOPIC, OUTPUT_TOPIC_2, OUTPUT_TOPIC_2_RERUN);
+
add10InputElements();
}
