sohami closed pull request #1463: DRILL-6625: Intermittent failures in Kafka
unit tests
URL: https://github.com/apache/drill/pull/1463
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/contrib/storage-kafka/pom.xml b/contrib/storage-kafka/pom.xml
index 5b5917d4a3a..0260c1c0c07 100644
--- a/contrib/storage-kafka/pom.xml
+++ b/contrib/storage-kafka/pom.xml
@@ -32,6 +32,7 @@
<properties>
<kafka.version>0.11.0.1</kafka.version>
+ <kafka.TestSuite>**/TestKafkaSuit.class</kafka.TestSuite>
</properties>
<dependencies>
@@ -97,4 +98,29 @@
<scope>test</scope>
</dependency>
</dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <includes>
+ <include>${kafka.TestSuite}</include>
+ </includes>
+ <excludes>
+ <exclude>**/KafkaFilterPushdownTest.java</exclude>
+ <exclude>**/KafkaQueriesTest.java</exclude>
+ <exclude>**/MessageIteratorTest.java</exclude>
+ </excludes>
+ <systemProperties>
+ <property>
+ <name>logback.log.dir</name>
+ <value>${project.build.directory}/surefire-reports</value>
+ </property>
+ </systemProperties>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
</project>
diff --git
a/contrib/storage-kafka/src/test/java/org/apache/drill/exec/store/kafka/KafkaFilterPushdownTest.java
b/contrib/storage-kafka/src/test/java/org/apache/drill/exec/store/kafka/KafkaFilterPushdownTest.java
index 7be0ec39415..d874733055a 100644
---
a/contrib/storage-kafka/src/test/java/org/apache/drill/exec/store/kafka/KafkaFilterPushdownTest.java
+++
b/contrib/storage-kafka/src/test/java/org/apache/drill/exec/store/kafka/KafkaFilterPushdownTest.java
@@ -27,6 +27,7 @@
import static org.apache.drill.exec.store.kafka.TestKafkaSuit.NUM_JSON_MSG;
import static
org.apache.drill.exec.store.kafka.TestKafkaSuit.embeddedKafkaCluster;
+import static org.junit.Assert.assertTrue;
@Category({KafkaStorageTest.class, SlowTest.class})
public class KafkaFilterPushdownTest extends KafkaTestBase {
@@ -42,6 +43,10 @@ public static void setup() throws Exception {
KafkaMessageGenerator generator = new
KafkaMessageGenerator(embeddedKafkaCluster.getKafkaBrokerList(),
StringSerializer.class);
generator.populateJsonMsgWithTimestamps(TestQueryConstants.JSON_PUSHDOWN_TOPIC,
NUM_JSON_MSG);
+ String query = String.format(TestQueryConstants.MSG_SELECT_QUERY,
TestQueryConstants.JSON_PUSHDOWN_TOPIC);
+ //Ensure messages are present
+ assertTrue("Kafka server does not have expected number of messages",
+ testSql(query) == NUM_PARTITIONS * NUM_JSON_MSG);
}
/**
diff --git
a/contrib/storage-kafka/src/test/java/org/apache/drill/exec/store/kafka/KafkaMessageGenerator.java
b/contrib/storage-kafka/src/test/java/org/apache/drill/exec/store/kafka/KafkaMessageGenerator.java
index f4a254ebce9..d094531a71c 100644
---
a/contrib/storage-kafka/src/test/java/org/apache/drill/exec/store/kafka/KafkaMessageGenerator.java
+++
b/contrib/storage-kafka/src/test/java/org/apache/drill/exec/store/kafka/KafkaMessageGenerator.java
@@ -55,14 +55,15 @@
public KafkaMessageGenerator (final String broker, Class<?> valueSerializer)
{
producerProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, broker);
producerProperties.put(ProducerConfig.ACKS_CONFIG, "all");
- producerProperties.put(ProducerConfig.RETRIES_CONFIG, 0);
+ producerProperties.put(ProducerConfig.RETRIES_CONFIG, 3);
producerProperties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
producerProperties.put(ProducerConfig.LINGER_MS_CONFIG, 0);
producerProperties.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1);
- producerProperties.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 1000);
+ producerProperties.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 5000);
producerProperties.put(ProducerConfig.CLIENT_ID_CONFIG,
"drill-test-kafka-client");
producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class);
producerProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
valueSerializer);
+ producerProperties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
//So that retries do not cause duplicates
}
public void populateAvroMsgIntoKafka(String topic, int numMsg) throws
IOException {
diff --git
a/contrib/storage-kafka/src/test/java/org/apache/drill/exec/store/kafka/KafkaTestBase.java
b/contrib/storage-kafka/src/test/java/org/apache/drill/exec/store/kafka/KafkaTestBase.java
index 24e6f6d68ae..ef71e619c52 100644
---
a/contrib/storage-kafka/src/test/java/org/apache/drill/exec/store/kafka/KafkaTestBase.java
+++
b/contrib/storage-kafka/src/test/java/org/apache/drill/exec/store/kafka/KafkaTestBase.java
@@ -57,7 +57,7 @@ public static void
initKafkaStoragePlugin(EmbeddedKafkaCluster embeddedKafkaClus
pluginRegistry.createOrUpdate(KafkaStoragePluginConfig.NAME,
storagePluginConfig, true);
testNoResult(String.format("alter session set `%s` = '%s'",
ExecConstants.KAFKA_RECORD_READER,
"org.apache.drill.exec.store.kafka.decoders.JsonMessageReader"));
- testNoResult(String.format("alter session set `%s` = %d",
ExecConstants.KAFKA_POLL_TIMEOUT, 200));
+ testNoResult(String.format("alter session set `%s` = %d",
ExecConstants.KAFKA_POLL_TIMEOUT, 5000));
}
public List<QueryDataBatch> runKafkaSQLWithResults(String sql) throws
Exception {
diff --git
a/contrib/storage-kafka/src/test/java/org/apache/drill/exec/store/kafka/TestKafkaSuit.java
b/contrib/storage-kafka/src/test/java/org/apache/drill/exec/store/kafka/TestKafkaSuit.java
index ecf998e3ad3..784eb4ee506 100644
---
a/contrib/storage-kafka/src/test/java/org/apache/drill/exec/store/kafka/TestKafkaSuit.java
+++
b/contrib/storage-kafka/src/test/java/org/apache/drill/exec/store/kafka/TestKafkaSuit.java
@@ -112,6 +112,7 @@ public static void createTopicHelper(final String
topicName, final int partition
Properties topicProps = new Properties();
topicProps.put(TopicConfig.MESSAGE_TIMESTAMP_TYPE_CONFIG, "CreateTime");
+ topicProps.put(TopicConfig.RETENTION_MS_CONFIG, "-1");
ZkUtils zkUtils = new ZkUtils(zkClient,
new
ZkConnection(embeddedKafkaCluster.getZkServer().getConnectionString()), false);
AdminUtils.createTopic(zkUtils, topicName, partitions, 1,
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
