[FLINK-15490][kafka][test-stability] Enable idempotence producing in …

…KafkaITCase to avoid intermittent test failure.
spotify · Jan 10, 2020 · f0f9343 · f0f9343
1 parent fb12bcf
commit f0f9343
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 0 deletions.
diff --git a/...-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironment.java b/...-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironment.java
@@ -125,6 +125,14 @@ public void createTestTopic(String topic, int numberOfPartitions, int replicatio
 
  public abstract List<KafkaServer> getBrokers();
 
+ public Properties getIdempotentProducerConfig() {
+ Properties props = new Properties();
+ props.put("enable.idempotence", "true");
+ props.put("acks", "all");
+ props.put("retries", "3");
+ return props;
+ }
+
  // -- consumer / producer instances:
  public <T> FlinkKafkaConsumerBase<T> getConsumer(List<String> topics, DeserializationSchema<T> deserializationSchema, Properties props) {
  return getConsumer(topics, new KafkaDeserializationSchemaWrapper<T>(deserializationSchema), props);

diff --git a/...e/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/DataGenerators.java b/...e/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/DataGenerators.java
@@ -104,6 +104,8 @@ public void cancel() {
  if (secureProps != null) {
  props.putAll(testServer.getSecureProperties());
  }
+ // Ensure the producer enables idempotence.
+ props.putAll(testServer.getIdempotentProducerConfig());
 
  stream = stream.rebalance();
  testServer.produceIntoKafka(stream, topic,