You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by li...@apache.org on 2023/01/26 07:30:29 UTC
[flink] 03/03: [FLINK-30623][runtime][bug] The canEmitBatchOfRecords should check recordWriter and changelogWriterAvailabilityProvider are available
This is an automated email from the ASF dual-hosted git repository.
lindong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 560f314703858464f5089e24e065347d00704af5
Author: 1996fanrui <19...@gmail.com>
AuthorDate: Wed Jan 25 16:24:46 2023 +0800
[FLINK-30623][runtime][bug] The canEmitBatchOfRecords should check recordWriter and changelogWriterAvailabilityProvider are available
This closes #21690.
---
...esultPartitionWriterWithAvailabilityHelper.java | 46 ++++++++++++++++++++++
.../flink/streaming/runtime/tasks/StreamTask.java | 12 ++++--
.../runtime/tasks/MultipleInputStreamTaskTest.java | 44 +++++++++++++++++++++
.../runtime/tasks/OneInputStreamTaskTest.java | 42 ++++++++++++++++++++
.../runtime/tasks/TwoInputStreamTaskTest.java | 44 +++++++++++++++++++++
5 files changed, 184 insertions(+), 4 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/writer/ResultPartitionWriterWithAvailabilityHelper.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/writer/ResultPartitionWriterWithAvailabilityHelper.java
new file mode 100644
index 00000000000..8bc5f7321ed
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/writer/ResultPartitionWriterWithAvailabilityHelper.java
@@ -0,0 +1,46 @@
+/*
+ * 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.runtime.io.network.api.writer;
+
+import org.apache.flink.runtime.io.network.partition.MockResultPartitionWriter;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * A specific result partition writer implementation only used to control the output availability
+ * state in tests.
+ */
+public class ResultPartitionWriterWithAvailabilityHelper extends MockResultPartitionWriter {
+
+ private final AvailabilityHelper availabilityHelper;
+
+ public ResultPartitionWriterWithAvailabilityHelper(AvailabilityHelper availabilityHelper) {
+ this.availabilityHelper = availabilityHelper;
+ }
+
+ @Override
+ public CompletableFuture<?> getAvailableFuture() {
+ return availabilityHelper.getAvailableFuture();
+ }
+
+ @Override
+ public boolean isAvailable() {
+ return availabilityHelper.isAvailable();
+ }
+}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
index 63f21f86f4b..6b52ed5da3b 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
@@ -547,9 +547,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
DataInputStatus status = inputProcessor.processInput();
switch (status) {
case MORE_AVAILABLE:
- if (recordWriter.isAvailable()
- && (changelogWriterAvailabilityProvider == null
- || changelogWriterAvailabilityProvider.isAvailable())) {
+ if (taskIsAvailable()) {
return;
}
break;
@@ -994,8 +992,14 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
return this.mailboxProcessor::getMailboxExecutor;
}
+ private boolean taskIsAvailable() {
+ return recordWriter.isAvailable()
+ && (changelogWriterAvailabilityProvider == null
+ || changelogWriterAvailabilityProvider.isAvailable());
+ }
+
public CanEmitBatchOfRecordsChecker getCanEmitBatchOfRecords() {
- return () -> !this.mailboxProcessor.hasMail();
+ return () -> !this.mailboxProcessor.hasMail() && taskIsAvailable();
}
public final boolean isRunning() {
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java
index 1a015d4fb84..8c46fd5fb92 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java
@@ -50,12 +50,14 @@ import org.apache.flink.runtime.checkpoint.CheckpointType;
import org.apache.flink.runtime.checkpoint.SavepointType;
import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.io.AvailabilityProvider;
import org.apache.flink.runtime.io.network.api.CancelCheckpointMarker;
import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
import org.apache.flink.runtime.io.network.api.EndOfData;
import org.apache.flink.runtime.io.network.api.EndOfPartitionEvent;
import org.apache.flink.runtime.io.network.api.StopMode;
import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
+import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriterWithAvailabilityHelper;
import org.apache.flink.runtime.io.network.partition.PartitionTestUtils;
import org.apache.flink.runtime.io.network.partition.ResultPartition;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
@@ -130,6 +132,7 @@ import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
/**
@@ -892,6 +895,47 @@ public class MultipleInputStreamTaskTest {
}
}
+ /** The CanEmitBatchOfRecords should always be false for {@link MultipleInputStreamTask}. */
+ @Test
+ public void testCanEmitBatchOfRecords() throws Exception {
+ AvailabilityProvider.AvailabilityHelper availabilityHelper =
+ new AvailabilityProvider.AvailabilityHelper();
+ try (StreamTaskMailboxTestHarness<String> testHarness =
+ new StreamTaskMailboxTestHarnessBuilder<>(
+ MultipleInputStreamTask::new, BasicTypeInfo.STRING_TYPE_INFO)
+ .addInput(BasicTypeInfo.STRING_TYPE_INFO)
+ .addInput(BasicTypeInfo.INT_TYPE_INFO)
+ .addInput(BasicTypeInfo.DOUBLE_TYPE_INFO)
+ .addAdditionalOutput(
+ new ResultPartitionWriterWithAvailabilityHelper(availabilityHelper))
+ .setupOperatorChain(new MapToStringMultipleInputOperatorFactory(3))
+ .finishForSingletonOperatorChain(IntSerializer.INSTANCE)
+ .build()) {
+ StreamTask.CanEmitBatchOfRecordsChecker canEmitBatchOfRecordsChecker =
+ testHarness.streamTask.getCanEmitBatchOfRecords();
+ testHarness.processAll();
+
+ availabilityHelper.resetAvailable();
+ assertFalse(canEmitBatchOfRecordsChecker.check());
+
+ // The canEmitBatchOfRecordsChecker should be the false after the record writer is
+ // unavailable.
+ availabilityHelper.resetUnavailable();
+ assertFalse(canEmitBatchOfRecordsChecker.check());
+
+ // Restore record writer to available
+ availabilityHelper.resetAvailable();
+ assertFalse(canEmitBatchOfRecordsChecker.check());
+
+ // The canEmitBatchOfRecordsChecker should be the false after add the mail to mail box.
+ testHarness.streamTask.mainMailboxExecutor.execute(() -> {}, "mail");
+ assertFalse(canEmitBatchOfRecordsChecker.check());
+
+ testHarness.processAll();
+ assertFalse(canEmitBatchOfRecordsChecker.check());
+ }
+ }
+
@Test
public void testLatencyMarker() throws Exception {
final Map<String, Metric> metrics = new ConcurrentHashMap<>();
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTaskTest.java
index bc7bb15ae21..f8e60522b31 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTaskTest.java
@@ -37,8 +37,10 @@ import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.io.AvailabilityProvider;
import org.apache.flink.runtime.io.network.api.CancelCheckpointMarker;
import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
+import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriterWithAvailabilityHelper;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.metrics.MetricNames;
import org.apache.flink.runtime.metrics.NoOpMetricRegistry;
@@ -61,6 +63,7 @@ import org.apache.flink.streaming.api.operators.StreamMap;
import org.apache.flink.streaming.api.operators.StreamOperator;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask.CanEmitBatchOfRecordsChecker;
import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
import org.apache.flink.streaming.util.TestHarnessUtil;
import org.apache.flink.util.Preconditions;
@@ -87,6 +90,7 @@ import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.cr
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -954,6 +958,44 @@ public class OneInputStreamTaskTest extends TestLogger {
testHarness.waitForTaskCompletion();
}
+ @Test
+ public void testCanEmitBatchOfRecords() throws Exception {
+ AvailabilityProvider.AvailabilityHelper availabilityHelper =
+ new AvailabilityProvider.AvailabilityHelper();
+ try (StreamTaskMailboxTestHarness<Integer> testHarness =
+ new StreamTaskMailboxTestHarnessBuilder<>(
+ OneInputStreamTask::new, BasicTypeInfo.INT_TYPE_INFO)
+ .addInput(BasicTypeInfo.INT_TYPE_INFO)
+ .addAdditionalOutput(
+ new ResultPartitionWriterWithAvailabilityHelper(availabilityHelper))
+ .setupOperatorChain(new TestOperator())
+ .finishForSingletonOperatorChain(IntSerializer.INSTANCE)
+ .build()) {
+ CanEmitBatchOfRecordsChecker canEmitBatchOfRecordsChecker =
+ testHarness.streamTask.getCanEmitBatchOfRecords();
+ testHarness.processAll();
+
+ availabilityHelper.resetAvailable();
+ assertTrue(canEmitBatchOfRecordsChecker.check());
+
+ // The canEmitBatchOfRecordsChecker should be the false after the record writer is
+ // unavailable.
+ availabilityHelper.resetUnavailable();
+ assertFalse(canEmitBatchOfRecordsChecker.check());
+
+ // Restore record writer to available
+ availabilityHelper.resetAvailable();
+ assertTrue(canEmitBatchOfRecordsChecker.check());
+
+ // The canEmitBatchOfRecordsChecker should be the false after add the mail to mail box.
+ testHarness.streamTask.mainMailboxExecutor.execute(() -> {}, "mail");
+ assertFalse(canEmitBatchOfRecordsChecker.check());
+
+ testHarness.processAll();
+ assertTrue(canEmitBatchOfRecordsChecker.check());
+ }
+ }
+
static class WatermarkMetricOperator extends AbstractStreamOperator<String>
implements OneInputStreamOperator<String, String> {
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TwoInputStreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TwoInputStreamTaskTest.java
index d98827d3137..9d215a4b1a6 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TwoInputStreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TwoInputStreamTaskTest.java
@@ -21,6 +21,7 @@ package org.apache.flink.streaming.runtime.tasks;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.api.common.typeutils.base.StringSerializer;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.Counter;
@@ -29,10 +30,12 @@ import org.apache.flink.metrics.Metric;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.io.AvailabilityProvider;
import org.apache.flink.runtime.io.network.api.CancelCheckpointMarker;
import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
import org.apache.flink.runtime.io.network.api.EndOfData;
import org.apache.flink.runtime.io.network.api.StopMode;
+import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriterWithAvailabilityHelper;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.metrics.MetricNames;
import org.apache.flink.runtime.metrics.NoOpMetricRegistry;
@@ -75,6 +78,7 @@ import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.contains;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
/**
* Tests for {@link TwoInputStreamTask}.
@@ -781,6 +785,46 @@ public class TwoInputStreamTaskTest {
testHarness.waitForTaskCompletion();
}
+ /** The CanEmitBatchOfRecords should always be false for {@link TwoInputStreamTask}. */
+ @Test
+ public void testCanEmitBatchOfRecords() throws Exception {
+ AvailabilityProvider.AvailabilityHelper availabilityHelper =
+ new AvailabilityProvider.AvailabilityHelper();
+ try (StreamTaskMailboxTestHarness<String> testHarness =
+ new StreamTaskMailboxTestHarnessBuilder<>(
+ TwoInputStreamTask::new, BasicTypeInfo.STRING_TYPE_INFO)
+ .addInput(BasicTypeInfo.STRING_TYPE_INFO)
+ .addInput(BasicTypeInfo.STRING_TYPE_INFO)
+ .addAdditionalOutput(
+ new ResultPartitionWriterWithAvailabilityHelper(availabilityHelper))
+ .setupOperatorChain(new DuplicatingOperator())
+ .finishForSingletonOperatorChain(IntSerializer.INSTANCE)
+ .build()) {
+ StreamTask.CanEmitBatchOfRecordsChecker canEmitBatchOfRecordsChecker =
+ testHarness.streamTask.getCanEmitBatchOfRecords();
+ testHarness.processAll();
+
+ availabilityHelper.resetAvailable();
+ assertFalse(canEmitBatchOfRecordsChecker.check());
+
+ // The canEmitBatchOfRecordsChecker should be the false after the record writer is
+ // unavailable.
+ availabilityHelper.resetUnavailable();
+ assertFalse(canEmitBatchOfRecordsChecker.check());
+
+ // Restore record writer to available
+ availabilityHelper.resetAvailable();
+ assertFalse(canEmitBatchOfRecordsChecker.check());
+
+ // The canEmitBatchOfRecordsChecker should be the false after add the mail to mail box.
+ testHarness.streamTask.mainMailboxExecutor.execute(() -> {}, "mail");
+ assertFalse(canEmitBatchOfRecordsChecker.check());
+
+ testHarness.processAll();
+ assertFalse(canEmitBatchOfRecordsChecker.check());
+ }
+ }
+
// This must only be used in one test, otherwise the static fields will be changed
// by several tests concurrently
private static class TestOpenCloseMapFunction