HOTFIX: Do Not use unlimited num messages in IntegrationTestUtils
Removed readKeyValues() that give UNLIMITED_MESSAGES which will doom to exhaust all wait time, as all its callers actually do provide the expected number of messages.
Author: Guozhang Wang
Reviewers: Matthias J. Sax Jason Gustafson
Closes from guozhangwang/KHotfix-not-use-limited-num-messages
@@ -53,7 +53,6 @@ import java.util.concurrent.Future;
*/
public class IntegrationTestUtils {
- public static final int UNLIMITED_MESSAGES = -1;
public static final long DEFAULT_TIMEOUT = 30 * 1000L;
/**
@@ -74,19 +73,6 @@ public class IntegrationTestUtils {
return returnList;
}
- /**
- * Returns as many messages as possible from the topic until a (currently hardcoded) timeout is
- * reached.
- *
- * @param topic Kafka topic to read messages from
- * @param consumerConfig Kafka consumer configuration
- * @param waitTime Maximum wait time in milliseconds
- * @return The KeyValue elements retrieved via the consumer.
- */
- public static <K, V> List<KeyValue<K, V>> readKeyValues(final String topic, final Properties consumerConfig, final long waitTime) {
- return readKeyValues(topic, consumerConfig, waitTime, UNLIMITED_MESSAGES);
- }
-
/**
* Returns up to `maxMessages` by reading via the provided consumer (the topic(s) to read from
* are already configured in the consumer).
@@ -210,7 +196,7 @@ public class IntegrationTestUtils {
final TestCondition valuesRead = new TestCondition() {
@Override
public boolean conditionMet() {
- final List<KeyValue<K, V>> readData = readKeyValues(topic, consumerConfig, waitTime);
+ final List<KeyValue<K, V>> readData = readKeyValues(topic, consumerConfig, waitTime, expectedNumRecords);
accumData.addAll(readData);
return accumData.size() >= expectedNumRecords;
}
