This is an automated email from the ASF dual-hosted git repository. jqin pushed a commit to branch release-1.9 in repository https://gitbox.apache.org/repos/asf/flink.git
commit 9db954365ac0ac317fdda4c11f9204c5df3aaebc Author: Jiangjie (Becket) Qin <[email protected]> AuthorDate: Sat Oct 26 18:55:44 2019 +0800 [hotfix][kafka][test-stability] Accelerate the KafkaProducerTest by reducing the timeout values. --- .../flink/streaming/connectors/kafka/KafkaProducerTestBase.java | 4 ++++ .../org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java | 4 ++++ 2 files changed, 8 insertions(+) diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java index d89e8ad..1d7630a 100644 --- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java +++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java @@ -245,6 +245,10 @@ public abstract class KafkaProducerTestBase extends KafkaTestBaseWithFlink { properties.putAll(secureProps); // decrease timeout and block time from 60s down to 10s - this is how long KafkaProducer will try send pending (not flushed) data on close() properties.setProperty("timeout.ms", "10000"); + // KafkaProducer prior to KIP-91 (release 2.1) uses request timeout to expire the unsent records. + properties.setProperty("request.timeout.ms", "3000"); + // KafkaProducer in 2.1.0 and above uses delivery timeout to expire the the records. + properties.setProperty("delivery.timeout.ms", "5000"); properties.setProperty("max.block.ms", "10000"); // increase batch.size and linger.ms - this tells KafkaProducer to batch produced events instead of flushing them immediately properties.setProperty("batch.size", "10240000"); diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java index c86dd08..b85481f 100644 --- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java +++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java @@ -222,6 +222,10 @@ public abstract class KafkaTestBase extends TestLogger { while (System.currentTimeMillis() < startMillis + timeoutMillis) { properties.put("key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer"); properties.put("value.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer"); + // We need to set these two properties so that they are lower than request.timeout.ms. This is + // required for some old KafkaConsumer versions. + properties.put("session.timeout.ms", "2000"); + properties.put("heartbeat.interval.ms", "500"); // query kafka for new records ... Collection<ConsumerRecord<Integer, Integer>> records = kafkaServer.getAllRecordsFromTopic(properties, topic, partition, 100);
