[flink] branch release-1.9 updated: [FLINK-14235][kafka, tests] Change source in at-least-once test from finite to infinite

[jqin]

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


The following commit(s) were added to refs/heads/release-1.9 by this push:
     new 7bed1e2  [FLINK-14235][kafka,tests] Change source in at-least-once 
test from finite to infinite
7bed1e2 is described below

commit 7bed1e229b87670ca9f80778f7b75829b48c91b1
Author: Piotr Nowojski <piotr.nowoj...@gmail.com>
AuthorDate: Fri Oct 18 16:38:26 2019 +0200

    [FLINK-14235][kafka,tests] Change source in at-least-once test from finite 
to infinite
    
    Previously it was possible that the source would end before a first 
chcekpoint could complete.
    If that was the case, any exceptions thrown during checkpointing are 
swallowed, which could explain
    the apparent data loss from FLINK-14235.
---
 .../connectors/kafka/KafkaProducerTestBase.java      | 20 +++++++++++++++++++-
 1 file changed, 19 insertions(+), 1 deletion(-)

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 121d347..311e515 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
@@ -254,7 +254,7 @@ public abstract class KafkaProducerTestBase extends 
KafkaTestBaseWithFlink {
 
                // process exactly failAfterElements number of elements and 
then shutdown Kafka broker and fail application
                DataStream<Integer> inputStream = env
-                       .fromCollection(getIntegersSequence(numElements))
+                       .addSource(new InfiniteIntegerSource())
                        .map(new BrokerRestartingMapper<>(failAfterElements));
 
                StreamSink<Integer> kafkaSink = 
kafkaServer.getProducerSink(topic, keyedSerializationSchema, properties, new 
FlinkKafkaPartitioner<Integer>() {
@@ -539,4 +539,22 @@ public abstract class KafkaProducerTestBase extends 
KafkaTestBaseWithFlink {
                public void initializeState(FunctionInitializationContext 
context) throws Exception {
                }
        }
+
+       private static final class InfiniteIntegerSource implements 
SourceFunction<Integer> {
+
+               private volatile boolean running = true;
+               private int counter = 0;
+
+               @Override
+               public void run(SourceContext<Integer> ctx) throws Exception {
+                       while (running) {
+                               ctx.collect(counter++);
+                       }
+               }
+
+               @Override
+               public void cancel() {
+                       running = false;
+               }
+       }
 }