You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ar...@apache.org on 2021/09/01 05:56:45 UTC
[flink] branch release-1.14 updated: [FLINK-23351][tests] Harden
FileReadingWatermarkITCase by counting watermarks after source operator
finishes
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.14 by this push:
new e66b92b [FLINK-23351][tests] Harden FileReadingWatermarkITCase by counting watermarks after source operator finishes
e66b92b is described below
commit e66b92be2a5abaa6dc7b1510b21b00a9447754f1
Author: Fabian Paul <fa...@ververica.com>
AuthorDate: Wed Aug 25 16:38:41 2021 +0200
[FLINK-23351][tests] Harden FileReadingWatermarkITCase by counting watermarks after source operator finishes
This PR changes the semantic of FileReadingWatermarkITCase to wait for
an expected number of watermarks after the ContinuousFileReaderOperator
finishes instead of calculating the expected watermark number with an
error margin. The error margin was not realiable enough and caused
instabilities on different systems.
(cherry picked from commit 0b52bfdbf94c4c355fc50cb681a717b51e9ef442)
---
.../streaming/api/FileReadingWatermarkITCase.java | 131 ++++++++-------------
1 file changed, 47 insertions(+), 84 deletions(-)
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/FileReadingWatermarkITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/FileReadingWatermarkITCase.java
index d7c5eac..0163100 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/api/FileReadingWatermarkITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/api/FileReadingWatermarkITCase.java
@@ -17,129 +17,92 @@
package org.apache.flink.test.streaming.api;
-import org.apache.flink.api.common.JobExecutionResult;
-import org.apache.flink.api.common.accumulators.IntCounter;
-import org.apache.flink.api.common.accumulators.LongCounter;
+import org.apache.flink.api.common.eventtime.TimestampAssigner;
import org.apache.flink.api.common.eventtime.Watermark;
-import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.configuration.Configuration;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
-import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
-import org.apache.flink.streaming.api.windowing.time.Time;
-import org.apache.flink.util.Preconditions;
+import org.apache.flink.test.util.InfiniteIntegerInputFormat;
+import org.apache.flink.testutils.junit.SharedObjects;
+import org.apache.flink.testutils.junit.SharedReference;
+import org.apache.flink.util.TestLogger;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
-import java.io.File;
-import java.io.IOException;
-import java.io.PrintWriter;
+import java.util.concurrent.CountDownLatch;
import static org.apache.flink.util.Preconditions.checkState;
-import static org.junit.Assert.assertEquals;
/**
* Tests that watermarks are emitted while file is being read, particularly the last split.
*
* @see <a href="https://issues.apache.org/jira/browse/FLINK-19109">FLINK-19109</a>
*/
-public class FileReadingWatermarkITCase {
- private static final String NUM_WATERMARKS_ACC_NAME = "numWatermarks";
- private static final String RUNTIME_ACC_NAME = "runtime";
- private static final int FILE_SIZE_LINES = 5_000_000;
- private static final int WATERMARK_INTERVAL_MILLIS = 10;
- private static final int MIN_EXPECTED_WATERMARKS = 5;
+public class FileReadingWatermarkITCase extends TestLogger {
+ private static final Logger LOG = LoggerFactory.getLogger(FileReadingWatermarkITCase.class);
+
+ private static final int WATERMARK_INTERVAL_MILLIS = 1_000;
+ private static final int EXPECTED_WATERMARKS = 5;
@Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+ @Rule public final SharedObjects sharedObjects = SharedObjects.create();
+ /**
+ * Adds an infinite split that causes the input of {@link
+ * org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator} to instantly go
+ * idle while data is still being processed.
+ *
+ * <p>Before FLINK-19109, watermarks would not be emitted at this point.
+ */
@Test
public void testWatermarkEmissionWithChaining() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(1);
env.getConfig().setAutoWatermarkInterval(WATERMARK_INTERVAL_MILLIS);
-
+ SharedReference<CountDownLatch> latch =
+ sharedObjects.add(new CountDownLatch(EXPECTED_WATERMARKS));
checkState(env.isChainingEnabled());
-
- Tuple3<File, String, String> sourceFileAndFirstLastLines = getSourceFile(FILE_SIZE_LINES);
- env.readTextFile(sourceFileAndFirstLastLines.f0.getAbsolutePath())
- .assignTimestampsAndWatermarks(getExtractorAssigner())
- .addSink(
- getWatermarkCounter(
- sourceFileAndFirstLastLines.f1, sourceFileAndFirstLastLines.f2));
-
- JobExecutionResult result = env.execute();
-
- int actual = result.getAccumulatorResult(NUM_WATERMARKS_ACC_NAME);
- long expected =
- result.<Long>getAccumulatorResult(RUNTIME_ACC_NAME) / WATERMARK_INTERVAL_MILLIS;
- checkState(expected > MIN_EXPECTED_WATERMARKS);
-
- assertEquals(
- "too few watermarks emitted in "
- + result.<Long>getAccumulatorResult(RUNTIME_ACC_NAME)
- + " ms",
- expected,
- actual,
- .5 * expected);
+ env.createInput(new InfiniteIntegerInputFormat(true))
+ .assignTimestampsAndWatermarks(
+ WatermarkStrategy.<Integer>forMonotonousTimestamps()
+ .withTimestampAssigner(context -> getExtractorAssigner()))
+ .addSink(getWatermarkCounter(latch));
+ env.executeAsync();
+ latch.get().await();
}
- private Tuple3<File, String, String> getSourceFile(int numLines) throws IOException {
- Preconditions.checkArgument(numLines > 0);
- File file = temporaryFolder.newFile();
- String first = "0", last = null;
- try (PrintWriter printWriter = new PrintWriter(file)) {
- for (int i = 0; i < numLines; i++) {
- printWriter.println(i);
- last = Integer.toString(i);
- }
- }
- return Tuple3.of(file, first, last);
- }
-
- private static BoundedOutOfOrdernessTimestampExtractor<String> getExtractorAssigner() {
- return new BoundedOutOfOrdernessTimestampExtractor<String>(Time.hours(1)) {
- private final long started = System.currentTimeMillis();
+ private static TimestampAssigner<Integer> getExtractorAssigner() {
+ return new TimestampAssigner<Integer>() {
+ private long counter = 1;
@Override
- public long extractTimestamp(String line) {
- return started + Long.parseLong(line);
+ public long extractTimestamp(Integer element, long recordTimestamp) {
+ return counter++;
}
};
}
- private static SinkFunction<String> getWatermarkCounter(
- final String firstElement, final String lastElement) {
- return new RichSinkFunction<String>() {
- private long start;
- private final IntCounter numWatermarks = new IntCounter();
- private final LongCounter runtime = new LongCounter();
- private long lastWatermark = -1;
-
+ private static SinkFunction<Integer> getWatermarkCounter(
+ final SharedReference<CountDownLatch> latch) {
+ return new RichSinkFunction<Integer>() {
@Override
- public void open(Configuration parameters) throws Exception {
- super.open(parameters);
- getRuntimeContext().addAccumulator(NUM_WATERMARKS_ACC_NAME, numWatermarks);
- getRuntimeContext().addAccumulator(RUNTIME_ACC_NAME, runtime);
- }
-
- @Override
- public void invoke(String value, SinkFunction.Context context) {
- if (value.equals(firstElement)) {
- start = System.nanoTime();
- }
- if (value.equals(lastElement)) {
- runtime.add((System.nanoTime() - start) / 1_000_000);
+ public void invoke(Integer value, SinkFunction.Context context) {
+ try {
+ Thread.sleep(1000);
+ LOG.info("Sink received record");
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
}
}
@Override
public void writeWatermark(Watermark watermark) {
- if (watermark.getTimestamp() != lastWatermark) {
- lastWatermark = watermark.getTimestamp();
- numWatermarks.add(1);
- }
+ LOG.info("Sink received watermark {}", watermark);
+ latch.get().countDown();
}
};
}