You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by jq...@apache.org on 2019/10/29 05:45:34 UTC
[flink] 02/02: [hotfix][kafka][test-stability] Accelerate the
KafkaProducerTest by reducing the timeout values.
This is an automated email from the ASF dual-hosted git repository.
jqin pushed a commit to branch release-1.8
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 443ffaeb6c828be2f2ad1c434e9873341c44bcef
Author: Jiangjie (Becket) Qin <ji...@alibaba-inc.com>
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 d23523f..9527694 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
@@ -210,6 +210,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);