You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2021/06/09 13:21:34 UTC
[flink] 02/03: [hotfix][test] Deduplicate test case between
SourceStreamTask and SourceOperatorStreamTask
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.13
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 6e1decb077ea6a54b87f335cf4e49b4ad6da954f
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Wed Jun 2 10:06:08 2021 +0200
[hotfix][test] Deduplicate test case between SourceStreamTask and SourceOperatorStreamTask
---
.../tasks/SourceOperatorStreamTaskTest.java | 52 ++------------
.../runtime/tasks/SourceStreamTaskTest.java | 66 +++++------------
.../runtime/tasks/SourceStreamTaskTestBase.java | 83 ++++++++++++++++++++++
3 files changed, 106 insertions(+), 95 deletions(-)
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTaskTest.java
index 87ec424..41e8b44 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTaskTest.java
@@ -32,16 +32,12 @@ import org.apache.flink.api.connector.source.mocks.MockSourceSplit;
import org.apache.flink.api.connector.source.mocks.MockSourceSplitSerializer;
import org.apache.flink.core.io.InputStatus;
import org.apache.flink.core.testutils.OneShotLatch;
-import org.apache.flink.metrics.Gauge;
-import org.apache.flink.metrics.Metric;
import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.checkpoint.CheckpointType;
import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
import org.apache.flink.runtime.jobgraph.OperatorID;
-import org.apache.flink.runtime.metrics.MetricNames;
-import org.apache.flink.runtime.metrics.groups.TaskMetricGroup;
import org.apache.flink.runtime.source.event.AddSplitEvent;
import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
import org.apache.flink.streaming.api.operators.SourceOperator;
@@ -56,23 +52,18 @@ import java.io.Serializable;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
-import java.util.Map;
import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.function.Supplier;
import java.util.stream.IntStream;
-import static org.apache.flink.api.common.typeinfo.BasicTypeInfo.INT_TYPE_INFO;
import static org.apache.flink.streaming.util.TestHarnessUtil.assertOutputEquals;
import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.Matchers.greaterThanOrEqualTo;
import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.Matchers.lessThanOrEqualTo;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
@@ -80,48 +71,17 @@ import static org.junit.Assert.assertTrue;
* Tests for verifying that the {@link SourceOperator} as a task input can be integrated well with
* {@link org.apache.flink.streaming.runtime.io.StreamOneInputProcessor}.
*/
-public class SourceOperatorStreamTaskTest {
+public class SourceOperatorStreamTaskTest extends SourceStreamTaskTestBase {
private static final OperatorID OPERATOR_ID = new OperatorID();
private static final int NUM_RECORDS = 10;
@Test
public void testMetrics() throws Exception {
- long sleepTime = 42;
-
- StreamTaskMailboxTestHarnessBuilder<Integer> builder =
- new StreamTaskMailboxTestHarnessBuilder<>(
- SourceOperatorStreamTask::new, INT_TYPE_INFO);
-
- final Map<String, Metric> metrics = new ConcurrentHashMap<>();
- final TaskMetricGroup taskMetricGroup =
- new StreamTaskTestHarness.TestTaskMetricGroup(metrics);
-
- try (StreamTaskMailboxTestHarness<Integer> harness =
- builder.setupOutputForSingletonOperatorChain(
- new SourceOperatorFactory<>(
- new MockSource(Boundedness.BOUNDED, 1),
- WatermarkStrategy.noWatermarks()))
- .setTaskMetricGroup(taskMetricGroup)
- .build()) {
-
- Future<Boolean> triggerFuture =
- harness.streamTask.triggerCheckpointAsync(
- new CheckpointMetaData(1L, System.currentTimeMillis()),
- CheckpointOptions.forCheckpointWithDefaultLocation());
-
- assertFalse(triggerFuture.isDone());
- Thread.sleep(sleepTime);
- while (!triggerFuture.isDone()) {
- harness.streamTask.runMailboxStep();
- }
- Gauge<Long> checkpointStartDelayGauge =
- (Gauge<Long>) metrics.get(MetricNames.CHECKPOINT_START_DELAY_TIME);
- assertThat(
- checkpointStartDelayGauge.getValue(),
- greaterThanOrEqualTo(sleepTime * 1_000_000));
- Gauge<Double> busyTimeGauge = (Gauge<Double>) metrics.get(MetricNames.TASK_BUSY_TIME);
- assertThat(busyTimeGauge.getValue(), lessThanOrEqualTo(1_000_000d));
- }
+ testMetrics(
+ SourceOperatorStreamTask::new,
+ new SourceOperatorFactory<>(
+ new MockSource(Boundedness.BOUNDED, 1), WatermarkStrategy.noWatermarks()),
+ lessThanOrEqualTo(1_000_000d));
}
/**
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java
index 3dff912..9a41cf9 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java
@@ -27,14 +27,10 @@ import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.testutils.MultiShotLatch;
import org.apache.flink.core.testutils.OneShotLatch;
-import org.apache.flink.metrics.Gauge;
-import org.apache.flink.metrics.Metric;
import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.execution.CancelTaskException;
import org.apache.flink.runtime.jobgraph.OperatorID;
-import org.apache.flink.runtime.metrics.MetricNames;
-import org.apache.flink.runtime.metrics.groups.TaskMetricGroup;
import org.apache.flink.runtime.operators.testutils.ExpectedTestException;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.checkpoint.ListCheckpointed;
@@ -43,6 +39,7 @@ import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
import org.apache.flink.streaming.api.operators.StreamSource;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.util.TestHarnessUtil;
@@ -60,11 +57,9 @@ import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-import java.util.Map;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
@@ -75,21 +70,21 @@ import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
+import static org.apache.flink.api.common.typeinfo.BasicTypeInfo.INT_TYPE_INFO;
import static org.apache.flink.api.common.typeinfo.BasicTypeInfo.STRING_TYPE_INFO;
import static org.apache.flink.runtime.checkpoint.CheckpointType.SAVEPOINT_SUSPEND;
import static org.apache.flink.runtime.state.CheckpointStorageLocationReference.getDefault;
import static org.apache.flink.util.Preconditions.checkState;
-import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
/**
* These tests verify that the RichFunction methods are called (in correct order). And that
* checkpointing/element emission don't occur concurrently.
*/
-public class SourceStreamTaskTest {
+public class SourceStreamTaskTest extends SourceStreamTaskTestBase {
@Test
public void testInputEndedBeforeStopWithSavepointConfirmed() throws Exception {
@@ -101,7 +96,7 @@ public class SourceStreamTaskTest {
new StreamTaskMailboxTestHarnessBuilder<>(SourceStreamTask::new, STRING_TYPE_INFO)
.setupOperatorChain(
new OperatorID(),
- new StreamSource<String, CancelTestSource>(source))
+ new StreamSource<String, CancelTestSource<String>>(source))
.chain(
new OperatorID(),
chainTail,
@@ -147,42 +142,16 @@ public class SourceStreamTaskTest {
Assert.assertEquals(10, resultElements.size());
}
- @Test(timeout = 60_000)
+ @Test
public void testMetrics() throws Exception {
- long sleepTime = 42;
- StreamTaskMailboxTestHarnessBuilder<String> builder =
- new StreamTaskMailboxTestHarnessBuilder<>(SourceStreamTask::new, STRING_TYPE_INFO);
-
- final Map<String, Metric> metrics = new ConcurrentHashMap<>();
- final TaskMetricGroup taskMetricGroup =
- new StreamTaskTestHarness.TestTaskMetricGroup(metrics);
-
- StreamTaskMailboxTestHarness<String> harness =
- builder.setupOutputForSingletonOperatorChain(
- new StreamSource<>(
- new CancelTestSource(
- STRING_TYPE_INFO.createSerializer(
- new ExecutionConfig()),
- "Hello")))
- .setTaskMetricGroup(taskMetricGroup)
- .build();
-
- Future<Boolean> triggerFuture =
- harness.streamTask.triggerCheckpointAsync(
- new CheckpointMetaData(1L, System.currentTimeMillis()),
- CheckpointOptions.forCheckpointWithDefaultLocation());
-
- assertFalse(triggerFuture.isDone());
- Thread.sleep(sleepTime);
- while (!triggerFuture.isDone()) {
- harness.streamTask.runMailboxStep();
- }
- Gauge<Long> checkpointStartDelayGauge =
- (Gauge<Long>) metrics.get(MetricNames.CHECKPOINT_START_DELAY_TIME);
- assertThat(
- checkpointStartDelayGauge.getValue(), greaterThanOrEqualTo(sleepTime * 1_000_000));
- Gauge<Double> busyTimeGauge = (Gauge<Double>) metrics.get(MetricNames.TASK_BUSY_TIME);
- assertTrue(Double.isNaN(busyTimeGauge.getValue()));
+ testMetrics(
+ SourceStreamTask::new,
+ SimpleOperatorFactory.of(
+ new StreamSource<Integer, SourceFunction<Integer>>(
+ new CancelTestSource(
+ INT_TYPE_INFO.createSerializer(new ExecutionConfig()),
+ 42))),
+ is(Double.NaN));
}
/**
@@ -792,20 +761,19 @@ public class SourceStreamTaskTest {
public void cancel() {}
}
- private static class CancelTestSource extends FromElementsFunction<String> {
+ private static class CancelTestSource<T> extends FromElementsFunction<T> {
private static final long serialVersionUID = 8713065281092996067L;
private static MultiShotLatch dataProcessing = new MultiShotLatch();
private static MultiShotLatch cancellationWaiting = new MultiShotLatch();
- public CancelTestSource(TypeSerializer<String> serializer, String... elements)
- throws IOException {
+ public CancelTestSource(TypeSerializer<T> serializer, T... elements) throws IOException {
super(serializer, elements);
}
@Override
- public void run(SourceContext<String> ctx) throws Exception {
+ public void run(SourceContext<T> ctx) throws Exception {
super.run(ctx);
dataProcessing.trigger();
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTestBase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTestBase.java
new file mode 100644
index 0000000..8a1ffbc
--- /dev/null
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTestBase.java
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.tasks;
+
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.Metric;
+import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.metrics.MetricNames;
+import org.apache.flink.runtime.metrics.groups.TaskMetricGroup;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.util.function.FunctionWithException;
+
+import org.hamcrest.Matcher;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Future;
+
+import static org.apache.flink.api.common.typeinfo.BasicTypeInfo.INT_TYPE_INFO;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
+
+/** Common base class for testing source tasks. */
+public class SourceStreamTaskTestBase {
+ public void testMetrics(
+ FunctionWithException<Environment, ? extends StreamTask<Integer, ?>, Exception>
+ taskFactory,
+ StreamOperatorFactory<?> operatorFactory,
+ Matcher<Double> busyTimeMatcher)
+ throws Exception {
+ long sleepTime = 42;
+
+ StreamTaskMailboxTestHarnessBuilder<Integer> builder =
+ new StreamTaskMailboxTestHarnessBuilder<>(taskFactory, INT_TYPE_INFO);
+
+ final Map<String, Metric> metrics = new ConcurrentHashMap<>();
+ final TaskMetricGroup taskMetricGroup =
+ new StreamTaskTestHarness.TestTaskMetricGroup(metrics);
+
+ try (StreamTaskMailboxTestHarness<Integer> harness =
+ builder.setupOutputForSingletonOperatorChain(operatorFactory)
+ .setTaskMetricGroup(taskMetricGroup)
+ .build()) {
+
+ Future<Boolean> triggerFuture =
+ harness.streamTask.triggerCheckpointAsync(
+ new CheckpointMetaData(1L, System.currentTimeMillis()),
+ CheckpointOptions.forCheckpointWithDefaultLocation());
+
+ assertFalse(triggerFuture.isDone());
+ Thread.sleep(sleepTime);
+ while (!triggerFuture.isDone()) {
+ harness.streamTask.runMailboxStep();
+ }
+ Gauge<Long> checkpointStartDelayGauge =
+ (Gauge<Long>) metrics.get(MetricNames.CHECKPOINT_START_DELAY_TIME);
+ assertThat(
+ checkpointStartDelayGauge.getValue(),
+ greaterThanOrEqualTo(sleepTime * 1_000_000));
+ Gauge<Double> busyTimeGauge = (Gauge<Double>) metrics.get(MetricNames.TASK_BUSY_TIME);
+ assertThat(busyTimeGauge.getValue(), busyTimeMatcher);
+ }
+ }
+}