You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by zh...@apache.org on 2020/06/05 17:12:00 UTC
[flink] branch master updated: [FLINK-18139][checkpointing] Fixing
unaligned checkpoints checks wrong channels for inflight data.
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new d25609d [FLINK-18139][checkpointing] Fixing unaligned checkpoints checks wrong channels for inflight data.
d25609d is described below
commit d25609d07f52ba9bac3f7bb33ce0635e255e3d9d
Author: Arvid Heise <ar...@ververica.com>
AuthorDate: Thu Jun 4 22:51:18 2020 +0200
[FLINK-18139][checkpointing] Fixing unaligned checkpoints checks wrong channels for inflight data.
CheckpointBarrierUnaligner#hasInflightData was not called with input gate contextual information, such that only the same first few channels are checked during initial snapshotting of inflight data for multi-gate setups.
---
.../partition/consumer/SingleInputGateBuilder.java | 4 +-
.../io/AlternatingCheckpointBarrierHandler.java | 5 +-
.../runtime/io/CheckpointBarrierHandler.java | 3 +-
.../runtime/io/CheckpointBarrierUnaligner.java | 4 +-
.../runtime/io/CheckpointedInputGate.java | 5 +-
.../AlternatingCheckpointBarrierHandlerTest.java | 4 +-
.../runtime/io/StreamTaskNetworkInputTest.java | 73 +++++++++++++++++++++-
7 files changed, 85 insertions(+), 13 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java
index ad607f7..b279998 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java
@@ -70,12 +70,12 @@ public class SingleInputGateBuilder {
return this;
}
- SingleInputGateBuilder setConsumedSubpartitionIndex(int consumedSubpartitionIndex) {
+ public SingleInputGateBuilder setConsumedSubpartitionIndex(int consumedSubpartitionIndex) {
this.consumedSubpartitionIndex = consumedSubpartitionIndex;
return this;
}
- SingleInputGateBuilder setSingleInputGateIndex(int gateIndex) {
+ public SingleInputGateBuilder setSingleInputGateIndex(int gateIndex) {
this.gateIndex = gateIndex;
return this;
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/AlternatingCheckpointBarrierHandler.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/AlternatingCheckpointBarrierHandler.java
index 09f05b9..8ed6788 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/AlternatingCheckpointBarrierHandler.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/AlternatingCheckpointBarrierHandler.java
@@ -18,6 +18,7 @@
package org.apache.flink.streaming.runtime.io;
import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo;
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.buffer.BufferReceivedListener;
@@ -99,9 +100,9 @@ class AlternatingCheckpointBarrierHandler extends CheckpointBarrierHandler {
}
@Override
- public boolean hasInflightData(long checkpointId, int channelIndex) {
+ public boolean hasInflightData(long checkpointId, InputChannelInfo channelInfo) {
// should only be called for unaligned checkpoint
- return unalignedHandler.hasInflightData(checkpointId, channelIndex);
+ return unalignedHandler.hasInflightData(checkpointId, channelInfo);
}
@Override
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierHandler.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierHandler.java
index 07f6f7a..952af24 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierHandler.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierHandler.java
@@ -22,6 +22,7 @@ import org.apache.flink.runtime.checkpoint.CheckpointException;
import org.apache.flink.runtime.checkpoint.CheckpointFailureReason;
import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
+import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo;
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.buffer.BufferReceivedListener;
@@ -88,7 +89,7 @@ public abstract class CheckpointBarrierHandler implements Closeable {
* this method returns true iff the unaligner still expects the respective barrier to be <i>consumed</i> on the
* that channel.
*/
- public boolean hasInflightData(long checkpointId, int channelIndex) {
+ public boolean hasInflightData(long checkpointId, InputChannelInfo channelInfo) {
return false;
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java
index 1d4bf82..0a6e661 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java
@@ -223,14 +223,14 @@ public class CheckpointBarrierUnaligner extends CheckpointBarrierHandler {
}
@Override
- public boolean hasInflightData(long checkpointId, int channelIndex) {
+ public boolean hasInflightData(long checkpointId, InputChannelInfo channelInfo) {
if (checkpointId < currentConsumedCheckpointId) {
return false;
}
if (checkpointId > currentConsumedCheckpointId) {
return true;
}
- return hasInflightBuffers[channelIndex];
+ return hasInflightBuffers[getFlattenedChannelIndex(channelInfo)];
}
@Override
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointedInputGate.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointedInputGate.java
index 578cf21..1e2ccf3 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointedInputGate.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointedInputGate.java
@@ -140,8 +140,9 @@ public class CheckpointedInputGate implements PullingAsyncDataInput<BufferOrEven
long checkpointId,
int channelIndex,
ChannelStateWriter channelStateWriter) throws IOException {
- if (barrierHandler.hasInflightData(checkpointId, channelIndex)) {
- inputGate.getChannel(channelIndex).spillInflightBuffers(checkpointId, channelStateWriter);
+ InputChannel channel = inputGate.getChannel(channelIndex);
+ if (barrierHandler.hasInflightData(checkpointId, channel.getChannelInfo())) {
+ channel.spillInflightBuffers(checkpointId, channelStateWriter);
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/AlternatingCheckpointBarrierHandlerTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/AlternatingCheckpointBarrierHandlerTest.java
index fdaacb5..d723d06 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/AlternatingCheckpointBarrierHandlerTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/AlternatingCheckpointBarrierHandlerTest.java
@@ -22,6 +22,7 @@ import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.checkpoint.CheckpointType;
import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
+import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo;
import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.partition.consumer.InputChannel;
@@ -107,7 +108,8 @@ public class AlternatingCheckpointBarrierHandlerTest {
private static void assertInflightDataEquals(CheckpointBarrierHandler expected, CheckpointBarrierHandler actual, long barrierId, int numChannels) {
for (int channelId = 0; channelId < numChannels; channelId++) {
- assertEquals(expected.hasInflightData(barrierId, channelId), actual.hasInflightData(barrierId, channelId));
+ InputChannelInfo channelInfo = new InputChannelInfo(0, channelId);
+ assertEquals(expected.hasInflightData(barrierId, channelInfo), actual.hasInflightData(barrierId, channelInfo));
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/StreamTaskNetworkInputTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/StreamTaskNetworkInputTest.java
index 0237aef..0793bf1 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/StreamTaskNetworkInputTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/StreamTaskNetworkInputTest.java
@@ -22,17 +22,24 @@ import org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.core.io.InputStatus;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
+import org.apache.flink.runtime.checkpoint.channel.RecordingChannelStateWriter;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
import org.apache.flink.runtime.io.network.api.EndOfPartitionEvent;
+import org.apache.flink.runtime.io.network.api.serialization.EventSerializer;
import org.apache.flink.runtime.io.network.api.serialization.RecordSerializer;
import org.apache.flink.runtime.io.network.api.serialization.SpanningRecordSerializer;
import org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer;
+import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.buffer.BufferBuilder;
import org.apache.flink.runtime.io.network.buffer.BufferBuilderTestUtils;
import org.apache.flink.runtime.io.network.buffer.BufferConsumer;
import org.apache.flink.runtime.io.network.partition.consumer.BufferOrEvent;
+import org.apache.flink.runtime.io.network.partition.consumer.InputChannelBuilder;
+import org.apache.flink.runtime.io.network.partition.consumer.RemoteInputChannel;
+import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate;
+import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGateBuilder;
import org.apache.flink.runtime.io.network.partition.consumer.StreamTestSingleInputGate;
import org.apache.flink.runtime.operators.testutils.DummyCheckpointInvokable;
import org.apache.flink.runtime.plugable.DeserializationDelegate;
@@ -55,6 +62,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
import java.util.stream.IntStream;
import static org.hamcrest.Matchers.is;
@@ -111,6 +119,67 @@ public class StreamTaskNetworkInputTest {
}
@Test
+ public void testSnapshotWithTwoInputGates() throws Exception {
+ CheckpointBarrierUnaligner unaligner = new CheckpointBarrierUnaligner(
+ new int[]{ 1, 1 },
+ ChannelStateWriter.NO_OP,
+ "test",
+ new DummyCheckpointInvokable());
+
+ SingleInputGate inputGate1 = new SingleInputGateBuilder().setSingleInputGateIndex(0).build();
+ RemoteInputChannel channel1 = InputChannelBuilder.newBuilder().buildRemoteChannel(inputGate1);
+ inputGate1.setInputChannels(channel1);
+ inputGate1.registerBufferReceivedListener(unaligner.getBufferReceivedListener().get());
+ StreamTaskNetworkInput<Long> input1 = createInput(unaligner, inputGate1);
+
+ SingleInputGate inputGate2 = new SingleInputGateBuilder().setSingleInputGateIndex(1).build();
+ RemoteInputChannel channel2 = InputChannelBuilder.newBuilder().buildRemoteChannel(inputGate2);
+ inputGate2.setInputChannels(channel2);
+ inputGate2.registerBufferReceivedListener(unaligner.getBufferReceivedListener().get());
+ StreamTaskNetworkInput<Long> input2 = createInput(unaligner, inputGate2);
+
+ CheckpointBarrier barrier = new CheckpointBarrier(0, 0L, CheckpointOptions.forCheckpointWithDefaultLocation());
+ channel1.onBuffer(EventSerializer.toBuffer(barrier), 0, 0);
+ channel1.onBuffer(BufferBuilderTestUtils.buildSomeBuffer(1), 1, 0);
+
+ // all records on inputGate2 are now in-flight
+ channel2.onBuffer(BufferBuilderTestUtils.buildSomeBuffer(2), 0, 0);
+ channel2.onBuffer(BufferBuilderTestUtils.buildSomeBuffer(3), 1, 0);
+
+ // now snapshot all inflight buffers
+ RecordingChannelStateWriter channelStateWriter = new RecordingChannelStateWriter();
+ channelStateWriter.start(0, CheckpointOptions.forCheckpointWithDefaultLocation());
+ CompletableFuture<Void> completableFuture1 = input1.prepareSnapshot(channelStateWriter, 0);
+ CompletableFuture<Void> completableFuture2 = input2.prepareSnapshot(channelStateWriter, 0);
+
+ // finish unaligned checkpoint on input side
+ channel2.onBuffer(EventSerializer.toBuffer(barrier), 2, 0);
+
+ // futures should be completed
+ completableFuture1.join();
+ completableFuture2.join();
+
+ assertEquals(channelStateWriter.getAddedInput().get(channel1.getChannelInfo()), Collections.emptyList());
+ List<Buffer> storedBuffers = channelStateWriter.getAddedInput().get(channel2.getChannelInfo());
+ assertEquals(Arrays.asList(2, 3), storedBuffers.stream().map(Buffer::getSize).collect(Collectors.toList()));
+ }
+
+ private StreamTaskNetworkInput<Long> createInput(CheckpointBarrierHandler handler, SingleInputGate inputGate) {
+ return new StreamTaskNetworkInput<>(
+ new CheckpointedInputGate(inputGate, handler),
+ LongSerializer.INSTANCE,
+ new StatusWatermarkValve(inputGate.getNumberOfInputChannels(), new NoOpDataOutput<>()),
+ inputGate.getGateIndex(),
+ createDeserializers(inputGate.getNumberOfInputChannels()));
+ }
+
+ private TestRecordDeserializer[] createDeserializers(int numberOfInputChannels) {
+ return IntStream.range(0, numberOfInputChannels)
+ .mapToObj(index -> new TestRecordDeserializer(ioManager.getSpillingDirectoriesPaths()))
+ .toArray(TestRecordDeserializer[]::new);
+ }
+
+ @Test
public void testSnapshotAfterEndOfPartition() throws Exception {
int numInputChannels = 1;
int channelId = 0;
@@ -119,9 +188,7 @@ public class StreamTaskNetworkInputTest {
VerifyRecordsDataOutput<Long> output = new VerifyRecordsDataOutput<>();
LongSerializer inSerializer = LongSerializer.INSTANCE;
StreamTestSingleInputGate<Long> inputGate = new StreamTestSingleInputGate<>(numInputChannels, 0, inSerializer, 1024);
- TestRecordDeserializer[] deserializers = IntStream.range(0, numInputChannels)
- .mapToObj(index -> new TestRecordDeserializer(ioManager.getSpillingDirectoriesPaths()))
- .toArray(TestRecordDeserializer[]::new);
+ TestRecordDeserializer[] deserializers = createDeserializers(numInputChannels);
StreamTaskNetworkInput<Long> input = new StreamTaskNetworkInput<>(
new CheckpointedInputGate(
inputGate.getInputGate(),