You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sr...@apache.org on 2018/02/22 13:29:27 UTC
[4/4] flink git commit: [FLINK-8547][network] Implement
CheckpointBarrierHandler not to spill data for exactly-once
[FLINK-8547][network] Implement CheckpointBarrierHandler not to spill data for exactly-once
This closes #5400.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/3126bf52
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/3126bf52
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/3126bf52
Branch: refs/heads/master
Commit: 3126bf522925a4f9a5cbf4c33077dbe7664363fa
Parents: 831349a
Author: Zhijiang <wa...@aliyun.com>
Authored: Fri Feb 2 15:45:49 2018 +0800
Committer: Stefan Richter <s....@data-artisans.com>
Committed: Thu Feb 22 14:27:36 2018 +0100
----------------------------------------------------------------------
.../flink/configuration/TaskManagerOptions.java | 12 +
.../streaming/runtime/io/BarrierBuffer.java | 36 +-
.../streaming/runtime/io/BufferBlocker.java | 68 +
.../runtime/io/BufferOrEventSequence.java | 59 +
.../streaming/runtime/io/BufferSpiller.java | 78 +-
.../runtime/io/CachedBufferBlocker.java | 147 ++
.../runtime/io/InputProcessorUtil.java | 71 +
.../runtime/io/StreamInputProcessor.java | 25 +-
.../runtime/io/StreamTwoInputProcessor.java | 25 +-
.../io/BarrierBufferAlignmentLimitTest.java | 4 +-
.../io/BarrierBufferMassiveRandomTest.java | 2 +-
.../streaming/runtime/io/BarrierBufferTest.java | 1545 ------------------
.../runtime/io/BarrierBufferTestBase.java | 1426 ++++++++++++++++
.../runtime/io/BufferBlockerTestBase.java | 325 ++++
.../streaming/runtime/io/BufferSpillerTest.java | 341 +---
.../runtime/io/CachedBufferBlockerTest.java | 53 +
.../io/CreditBasedBarrierBufferTest.java | 49 +
.../runtime/io/SpillingBarrierBufferTest.java | 81 +
18 files changed, 2361 insertions(+), 1986 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/3126bf52/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
index e01cf0f..cc3284c 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
@@ -317,6 +317,18 @@ public class TaskManagerOptions {
key("taskmanager.network.credit-based-flow-control.enabled")
.defaultValue(true);
+ /**
+ * Config parameter defining whether to spill data for channels with barrier or not in exactly-once
+ * mode based on credit-based flow control.
+ *
+ * @deprecated Will be removed for Flink 1.6 when the old code will be dropped in favour of
+ * credit-based flow control.
+ */
+ @Deprecated
+ public static final ConfigOption<Boolean> EXACTLY_ONCE_BLOCKING_DATA_ENABLED =
+ key("taskmanager.exactly-once.blocking.data.enabled")
+ .defaultValue(true);
+
// ------------------------------------------------------------------------
// Task Options
// ------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/3126bf52/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/BarrierBuffer.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/BarrierBuffer.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/BarrierBuffer.java
index 7ef9fef..78852b8 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/BarrierBuffer.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/BarrierBuffer.java
@@ -25,7 +25,6 @@ import org.apache.flink.runtime.checkpoint.decline.CheckpointDeclineException;
import org.apache.flink.runtime.checkpoint.decline.CheckpointDeclineOnCancellationBarrierException;
import org.apache.flink.runtime.checkpoint.decline.CheckpointDeclineSubsumedException;
import org.apache.flink.runtime.checkpoint.decline.InputEndOfStreamException;
-import org.apache.flink.runtime.io.disk.iomanager.IOManager;
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.EndOfPartitionEvent;
@@ -41,6 +40,7 @@ import java.util.ArrayDeque;
import java.util.Optional;
import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
/**
* The barrier buffer is {@link CheckpointBarrierHandler} that blocks inputs with barriers until
@@ -65,13 +65,13 @@ public class BarrierBuffer implements CheckpointBarrierHandler {
private final int totalNumberOfInputChannels;
/** To utility to write blocked data to a file channel. */
- private final BufferSpiller bufferSpiller;
+ private final BufferBlocker bufferBlocker;
/**
* The pending blocked buffer/event sequences. Must be consumed before requesting further data
* from the input gate.
*/
- private final ArrayDeque<BufferSpiller.SpilledBufferOrEventSequence> queuedBuffered;
+ private final ArrayDeque<BufferOrEventSequence> queuedBuffered;
/**
* The maximum number of bytes that may be buffered before an alignment is broken. -1 means
@@ -83,7 +83,7 @@ public class BarrierBuffer implements CheckpointBarrierHandler {
* The sequence of buffers/events that has been unblocked and must now be consumed before
* requesting further data from the input gate.
*/
- private BufferSpiller.SpilledBufferOrEventSequence currentBuffered;
+ private BufferOrEventSequence currentBuffered;
/** Handler that receives the checkpoint notifications. */
private AbstractInvokable toNotifyOnCheckpoint;
@@ -118,12 +118,12 @@ public class BarrierBuffer implements CheckpointBarrierHandler {
* <p>There is no limit to how much data may be buffered during an alignment.
*
* @param inputGate The input gate to draw the buffers and events from.
- * @param ioManager The I/O manager that gives access to the temp directories.
+ * @param bufferBlocker The buffer blocker to hold the buffers and events for channels with barrier.
*
* @throws IOException Thrown, when the spilling to temp files cannot be initialized.
*/
- public BarrierBuffer(InputGate inputGate, IOManager ioManager) throws IOException {
- this (inputGate, ioManager, -1);
+ public BarrierBuffer(InputGate inputGate, BufferBlocker bufferBlocker) throws IOException {
+ this (inputGate, bufferBlocker, -1);
}
/**
@@ -134,12 +134,13 @@ public class BarrierBuffer implements CheckpointBarrierHandler {
* checkpoint has been cancelled.
*
* @param inputGate The input gate to draw the buffers and events from.
- * @param ioManager The I/O manager that gives access to the temp directories.
+ * @param bufferBlocker The buffer blocker to hold the buffers and events for channels with barrier.
* @param maxBufferedBytes The maximum bytes to be buffered before the checkpoint aborts.
*
* @throws IOException Thrown, when the spilling to temp files cannot be initialized.
*/
- public BarrierBuffer(InputGate inputGate, IOManager ioManager, long maxBufferedBytes) throws IOException {
+ public BarrierBuffer(InputGate inputGate, BufferBlocker bufferBlocker, long maxBufferedBytes)
+ throws IOException {
checkArgument(maxBufferedBytes == -1 || maxBufferedBytes > 0);
this.inputGate = inputGate;
@@ -147,8 +148,8 @@ public class BarrierBuffer implements CheckpointBarrierHandler {
this.totalNumberOfInputChannels = inputGate.getNumberOfInputChannels();
this.blockedChannels = new boolean[this.totalNumberOfInputChannels];
- this.bufferSpiller = new BufferSpiller(ioManager, inputGate.getPageSize());
- this.queuedBuffered = new ArrayDeque<BufferSpiller.SpilledBufferOrEventSequence>();
+ this.bufferBlocker = checkNotNull(bufferBlocker);
+ this.queuedBuffered = new ArrayDeque<BufferOrEventSequence>();
}
// ------------------------------------------------------------------------
@@ -185,10 +186,9 @@ public class BarrierBuffer implements CheckpointBarrierHandler {
}
BufferOrEvent bufferOrEvent = next.get();
-
if (isBlocked(bufferOrEvent.getChannelIndex())) {
// if the channel is blocked we, we just store the BufferOrEvent
- bufferSpiller.add(bufferOrEvent);
+ bufferBlocker.add(bufferOrEvent);
checkSizeLimit();
}
else if (bufferOrEvent.isBuffer()) {
@@ -399,7 +399,7 @@ public class BarrierBuffer implements CheckpointBarrierHandler {
}
private void checkSizeLimit() throws Exception {
- if (maxBufferedBytes > 0 && (numQueuedBytes + bufferSpiller.getBytesWritten()) > maxBufferedBytes) {
+ if (maxBufferedBytes > 0 && (numQueuedBytes + bufferBlocker.getBytesBlocked()) > maxBufferedBytes) {
// exceeded our limit - abort this checkpoint
LOG.info("Checkpoint {} aborted because alignment volume limit ({} bytes) exceeded",
currentCheckpointId, maxBufferedBytes);
@@ -426,11 +426,11 @@ public class BarrierBuffer implements CheckpointBarrierHandler {
@Override
public void cleanup() throws IOException {
- bufferSpiller.close();
+ bufferBlocker.close();
if (currentBuffered != null) {
currentBuffered.cleanup();
}
- for (BufferSpiller.SpilledBufferOrEventSequence seq : queuedBuffered) {
+ for (BufferOrEventSequence seq : queuedBuffered) {
seq.cleanup();
}
queuedBuffered.clear();
@@ -491,7 +491,7 @@ public class BarrierBuffer implements CheckpointBarrierHandler {
if (currentBuffered == null) {
// common case: no more buffered data
- currentBuffered = bufferSpiller.rollOver();
+ currentBuffered = bufferBlocker.rollOverReusingResources();
if (currentBuffered != null) {
currentBuffered.open();
}
@@ -503,7 +503,7 @@ public class BarrierBuffer implements CheckpointBarrierHandler {
"Pushing back current alignment buffers and feeding back new alignment data first.");
// since we did not fully drain the previous sequence, we need to allocate a new buffer for this one
- BufferSpiller.SpilledBufferOrEventSequence bufferedNow = bufferSpiller.rollOverWithNewBuffer();
+ BufferOrEventSequence bufferedNow = bufferBlocker.rollOverWithoutReusingResources();
if (bufferedNow != null) {
bufferedNow.open();
queuedBuffered.addFirst(currentBuffered);
http://git-wip-us.apache.org/repos/asf/flink/blob/3126bf52/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/BufferBlocker.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/BufferBlocker.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/BufferBlocker.java
new file mode 100644
index 0000000..4d0f66f
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/BufferBlocker.java
@@ -0,0 +1,68 @@
+/*
+ * 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.io;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.io.network.partition.consumer.BufferOrEvent;
+
+import java.io.IOException;
+
+/**
+ * The buffer blocker takes the buffers and events from a data stream and adds them in a sequence.
+ * After a number of elements have been added, the blocker can "roll over": It presents the added
+ * elements as a readable sequence, and creates a new sequence.
+ */
+@Internal
+public interface BufferBlocker {
+
+ /**
+ * Adds a buffer or event to the blocker.
+ *
+ * @param boe The buffer or event to be added into the blocker.
+ */
+ void add(BufferOrEvent boe) throws IOException;
+
+ /**
+ * Starts a new sequence of buffers and event without reusing the same resources and
+ * returns the current sequence of buffers for reading.
+ *
+ * @return The readable sequence of buffers and events, or 'null', if nothing was added.
+ */
+ BufferOrEventSequence rollOverWithoutReusingResources() throws IOException;
+
+ /**
+ * Starts a new sequence of buffers and event reusing the same resources and
+ * returns the current sequence of buffers for reading.
+ *
+ * @return The readable sequence of buffers and events, or 'null', if nothing was added.
+ */
+ BufferOrEventSequence rollOverReusingResources() throws IOException;
+
+ /**
+ * Cleans up all the resources in the current sequence.
+ */
+ void close() throws IOException;
+
+ /**
+ * Gets the number of bytes blocked in the current sequence.
+ *
+ * @return the number of bytes blocked in the current sequence.
+ */
+ long getBytesBlocked();
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/3126bf52/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/BufferOrEventSequence.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/BufferOrEventSequence.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/BufferOrEventSequence.java
new file mode 100644
index 0000000..c5bde1b
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/BufferOrEventSequence.java
@@ -0,0 +1,59 @@
+/*
+ * 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.io;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.io.network.partition.consumer.BufferOrEvent;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+/**
+ * This class represents a sequence of buffers and events which are blocked by
+ * {@link CheckpointBarrierHandler}. The sequence of buffers and events can be
+ * read back using the method {@link #getNext()}.
+ */
+@Internal
+public interface BufferOrEventSequence {
+
+ /**
+ * Initializes the sequence for reading.
+ */
+ void open();
+
+ /**
+ * Gets the next BufferOrEvent from the sequence, or {@code null}, if the
+ * sequence is exhausted.
+ *
+ * @return The next BufferOrEvent from the buffered sequence, or {@code null} (end of sequence).
+ */
+ @Nullable
+ BufferOrEvent getNext() throws IOException;
+
+ /**
+ * Cleans up all the resources held by the sequence.
+ */
+ void cleanup() throws IOException;
+
+ /**
+ * Gets the size of the sequence.
+ */
+ long size();
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/3126bf52/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/BufferSpiller.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/BufferSpiller.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/BufferSpiller.java
index 33aac7e..7a0be33 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/BufferSpiller.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/BufferSpiller.java
@@ -45,13 +45,14 @@ import java.util.concurrent.atomic.AtomicInteger;
*
* <p>This implementation buffers data effectively in the OS cache, which gracefully extends to the
* disk. Most data is written and re-read milliseconds later. The file is deleted after the read.
- * Consequently, in most cases, the data will never actually hit the physical disks.</p>
+ * Consequently, in most cases, the data will never actually hit the physical disks.
*
* <p>IMPORTANT: The SpilledBufferOrEventSequences created by this spiller all reuse the same
- * reading memory (to reduce overhead) and can consequently not be read concurrently.</p>
+ * reading memory (to reduce overhead) and can consequently not be read concurrently.
*/
@Internal
-public class BufferSpiller {
+@Deprecated
+public class BufferSpiller implements BufferBlocker {
/** Size of header in bytes (see add method). */
static final int HEADER_SIZE = 9;
@@ -127,6 +128,7 @@ public class BufferSpiller {
* @param boe The buffer or event to add and spill.
* @throws IOException Thrown, if the buffer of event could not be spilled.
*/
+ @Override
public void add(BufferOrEvent boe) throws IOException {
try {
ByteBuffer contents;
@@ -157,40 +159,35 @@ public class BufferSpiller {
}
/**
- * Starts a new sequence of spilled buffers and event and returns the current sequence of spilled buffers
- * for reading. This method returns {@code null}, if nothing was added since the creation of the spiller, or the
- * last call to this method.
+ * NOTE: The BufferOrEventSequences created by this method all reuse the same reading memory
+ * (to reduce overhead) and can consequently not be read concurrently with each other.
*
- * <p>NOTE: The SpilledBufferOrEventSequences created by this method all reuse the same
- * reading memory (to reduce overhead) and can consequently not be read concurrently with each other.
- * To create a sequence that can be read concurrently with the previous SpilledBufferOrEventSequence, use the
- * {@link #rollOverWithNewBuffer()} method.</p>
+ * <p>To create a sequence that can be read concurrently with the previous BufferOrEventSequence,
+ * use the {@link #rollOverWithoutReusingResources()} ()} method.
*
* @return The readable sequence of spilled buffers and events, or 'null', if nothing was added.
* @throws IOException Thrown, if the readable sequence could not be created, or no new spill
* file could be created.
*/
- public SpilledBufferOrEventSequence rollOver() throws IOException {
- return rollOverInternal(false);
+ @Override
+ public BufferOrEventSequence rollOverReusingResources() throws IOException {
+ return rollOver(false);
}
/**
- * Starts a new sequence of spilled buffers and event and returns the current sequence of spilled buffers
- * for reading. This method returns {@code null}, if nothing was added since the creation of the spiller, or the
- * last call to this method.
- *
- * <p>The SpilledBufferOrEventSequence returned by this method is safe for concurrent consumption with
- * any previously returned sequence.</p>
+ * The BufferOrEventSequence returned by this method is safe for concurrent consumption with
+ * any previously returned sequence.
*
* @return The readable sequence of spilled buffers and events, or 'null', if nothing was added.
* @throws IOException Thrown, if the readable sequence could not be created, or no new spill
* file could be created.
*/
- public SpilledBufferOrEventSequence rollOverWithNewBuffer() throws IOException {
- return rollOverInternal(true);
+ @Override
+ public BufferOrEventSequence rollOverWithoutReusingResources() throws IOException {
+ return rollOver(true);
}
- private SpilledBufferOrEventSequence rollOverInternal(boolean newBuffer) throws IOException {
+ private BufferOrEventSequence rollOver(boolean newBuffer) throws IOException {
if (bytesWritten == 0) {
return null;
}
@@ -219,10 +216,11 @@ public class BufferSpiller {
* Cleans up the current spilling channel and file.
*
* <p>Does not clean up the SpilledBufferOrEventSequences generated by calls to
- * {@link #rollOver()}.
+ * {@link #rollOver(boolean false)}.
*
* @throws IOException Thrown if channel closing or file deletion fail.
*/
+ @Override
public void close() throws IOException {
currentChannel.close();
if (!currentSpillFile.delete()) {
@@ -232,9 +230,11 @@ public class BufferSpiller {
/**
* Gets the number of bytes written in the current spill file.
+ *
* @return the number of bytes written in the current spill file
*/
- public long getBytesWritten() {
+ @Override
+ public long getBytesBlocked() {
return bytesWritten;
}
@@ -264,10 +264,10 @@ public class BufferSpiller {
/**
* This class represents a sequence of spilled buffers and events, created by the
- * {@link BufferSpiller}. The sequence of buffers and events can be read back using the
- * method {@link #getNext()}.
+ * {@link BufferSpiller}.
*/
- public static class SpilledBufferOrEventSequence {
+ @Deprecated
+ public static class SpilledBufferOrEventSequence implements BufferOrEventSequence {
/** Header is "channel index" (4 bytes) + length (4 bytes) + buffer/event (1 byte). */
private static final int HEADER_LENGTH = 9;
@@ -308,10 +308,10 @@ public class BufferSpiller {
}
/**
- * Initializes the sequence for reading.
- * This method needs to be called before the first call to {@link #getNext()}. Otherwise
- * the results of {@link #getNext()} are not predictable.
+ * This method needs to be called before the first call to {@link #getNext()}.
+ * Otherwise the results of {@link #getNext()} are not predictable.
*/
+ @Override
public void open() {
if (!opened) {
opened = true;
@@ -320,13 +320,7 @@ public class BufferSpiller {
}
}
- /**
- * Gets the next BufferOrEvent from the spilled sequence, or {@code null}, if the
- * sequence is exhausted.
- *
- * @return The next BufferOrEvent from the spilled sequence, or {@code null} (end of sequence).
- * @throws IOException Thrown, if the reads failed, of if the byte stream is corrupt.
- */
+ @Override
public BufferOrEvent getNext() throws IOException {
if (buffer.remaining() < HEADER_LENGTH) {
buffer.compact();
@@ -413,11 +407,7 @@ public class BufferSpiller {
}
}
- /**
- * Cleans up all file resources held by this spilled sequence.
- *
- * @throws IOException Thrown, if file channel closing or file deletion fail.
- */
+ @Override
public void cleanup() throws IOException {
fileChannel.close();
if (!file.delete()) {
@@ -425,10 +415,8 @@ public class BufferSpiller {
}
}
- /**
- * Gets the size of this spilled sequence.
- */
- public long size() throws IOException {
+ @Override
+ public long size() {
return size;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/3126bf52/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CachedBufferBlocker.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CachedBufferBlocker.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CachedBufferBlocker.java
new file mode 100644
index 0000000..f91e8cc
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CachedBufferBlocker.java
@@ -0,0 +1,147 @@
+/*
+ * 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.io;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.io.network.partition.consumer.BufferOrEvent;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayDeque;
+
+/**
+ * The cached buffer blocker takes the buffers and events from a data stream and adds them to a memory queue.
+ * After a number of elements have been cached, the blocker can "roll over": It presents the cached
+ * elements as a readable sequence, and creates a new memory queue.
+ *
+ * <p>This buffer blocked can be used in credit-based flow control for better barrier alignment in exactly-once mode.
+ */
+@Internal
+public class CachedBufferBlocker implements BufferBlocker {
+
+ /** The page size, to estimate the total cached data size. */
+ private final int pageSize;
+
+ /** The number of bytes cached since the last roll over. */
+ private long bytesBlocked;
+
+ /** The current memory queue for caching the buffers or events. */
+ private ArrayDeque<BufferOrEvent> currentBuffers;
+
+ /**
+ * Creates a new buffer blocker, caching the buffers or events in memory queue.
+ *
+ * @param pageSize The page size used to estimate the cached size.
+ */
+ public CachedBufferBlocker(int pageSize) {
+ this.pageSize = pageSize;
+ this.currentBuffers = new ArrayDeque<BufferOrEvent>();
+ }
+
+ @Override
+ public void add(BufferOrEvent boe) {
+ bytesBlocked += pageSize;
+
+ currentBuffers.add(boe);
+ }
+
+ /**
+ * It is never reusing resources and is defaulting to {@link #rollOverWithoutReusingResources()}.
+ */
+ @Override
+ public BufferOrEventSequence rollOverReusingResources() {
+ return rollOverWithoutReusingResources();
+ }
+
+ @Override
+ public BufferOrEventSequence rollOverWithoutReusingResources() {
+ if (bytesBlocked == 0) {
+ return null;
+ }
+
+ CachedBufferOrEventSequence currentSequence = new CachedBufferOrEventSequence(currentBuffers, bytesBlocked);
+ currentBuffers = new ArrayDeque<BufferOrEvent>();
+ bytesBlocked = 0L;
+
+ return currentSequence;
+ }
+
+ @Override
+ public void close() {
+ BufferOrEvent boe;
+ while ((boe = currentBuffers.poll()) != null) {
+ if (boe.isBuffer()) {
+ boe.getBuffer().recycleBuffer();
+ }
+ }
+ }
+
+ @Override
+ public long getBytesBlocked() {
+ return bytesBlocked;
+ }
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * This class represents a sequence of cached buffers and events, created by the
+ * {@link CachedBufferBlocker}.
+ */
+ public static class CachedBufferOrEventSequence implements BufferOrEventSequence {
+
+ /** The sequence of buffers and events to be consumed by {@link BarrierBuffer}.*/
+ private final ArrayDeque<BufferOrEvent> queuedBuffers;
+
+ /** The total size of the cached data. */
+ private final long size;
+
+ /**
+ * Creates a reader that reads a sequence of buffers and events.
+ *
+ * @param size The total size of cached data.
+ */
+ CachedBufferOrEventSequence(ArrayDeque<BufferOrEvent> buffers, long size) {
+ this.queuedBuffers = buffers;
+ this.size = size;
+ }
+
+ @Override
+ public void open() {}
+
+ @Override
+ @Nullable
+ public BufferOrEvent getNext() {
+ return queuedBuffers.poll();
+ }
+
+ @Override
+ public void cleanup() {
+ BufferOrEvent boe;
+ while ((boe = queuedBuffers.poll()) != null) {
+ if (boe.isBuffer()) {
+ boe.getBuffer().recycleBuffer();
+ }
+ }
+ }
+
+ @Override
+ public long size() {
+ return size;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/3126bf52/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/InputProcessorUtil.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/InputProcessorUtil.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/InputProcessorUtil.java
new file mode 100644
index 0000000..cb56eee
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/InputProcessorUtil.java
@@ -0,0 +1,71 @@
+/*
+ * 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.io;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.IllegalConfigurationException;
+import org.apache.flink.configuration.TaskManagerOptions;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
+import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+
+import java.io.IOException;
+
+/**
+ * Utility for creating {@link CheckpointBarrierHandler} based on checkpoint mode
+ * for {@link StreamInputProcessor} and {@link StreamTwoInputProcessor}.
+ */
+@Internal
+public class InputProcessorUtil {
+
+ public static CheckpointBarrierHandler createCheckpointBarrierHandler(
+ StreamTask<?, ?> checkpointedTask,
+ CheckpointingMode checkpointMode,
+ IOManager ioManager,
+ InputGate inputGate,
+ Configuration taskManagerConfig) throws IOException {
+
+ CheckpointBarrierHandler barrierHandler;
+ if (checkpointMode == CheckpointingMode.EXACTLY_ONCE) {
+ long maxAlign = taskManagerConfig.getLong(TaskManagerOptions.TASK_CHECKPOINT_ALIGNMENT_BYTES_LIMIT);
+ if (!(maxAlign == -1 || maxAlign > 0)) {
+ throw new IllegalConfigurationException(
+ TaskManagerOptions.TASK_CHECKPOINT_ALIGNMENT_BYTES_LIMIT.key()
+ + " must be positive or -1 (infinite)");
+ }
+
+ if (taskManagerConfig.getBoolean(TaskManagerOptions.EXACTLY_ONCE_BLOCKING_DATA_ENABLED)) {
+ barrierHandler = new BarrierBuffer(inputGate, new CachedBufferBlocker(inputGate.getPageSize()), maxAlign);
+ } else {
+ barrierHandler = new BarrierBuffer(inputGate, new BufferSpiller(ioManager, inputGate.getPageSize()), maxAlign);
+ }
+ } else if (checkpointMode == CheckpointingMode.AT_LEAST_ONCE) {
+ barrierHandler = new BarrierTracker(inputGate);
+ } else {
+ throw new IllegalArgumentException("Unrecognized Checkpointing Mode: " + checkpointMode);
+ }
+
+ if (checkpointedTask != null) {
+ barrierHandler.registerCheckpointEventHandler(checkpointedTask);
+ }
+
+ return barrierHandler;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/3126bf52/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamInputProcessor.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamInputProcessor.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamInputProcessor.java
index dc3dc5c..a9c64b5 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamInputProcessor.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamInputProcessor.java
@@ -21,8 +21,6 @@ package org.apache.flink.streaming.runtime.io;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.IllegalConfigurationException;
-import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.SimpleCounter;
import org.apache.flink.runtime.event.AbstractEvent;
@@ -127,25 +125,8 @@ public class StreamInputProcessor<IN> {
InputGate inputGate = InputGateUtil.createInputGate(inputGates);
- if (checkpointMode == CheckpointingMode.EXACTLY_ONCE) {
- long maxAlign = taskManagerConfig.getLong(TaskManagerOptions.TASK_CHECKPOINT_ALIGNMENT_BYTES_LIMIT);
- if (!(maxAlign == -1 || maxAlign > 0)) {
- throw new IllegalConfigurationException(
- TaskManagerOptions.TASK_CHECKPOINT_ALIGNMENT_BYTES_LIMIT.key()
- + " must be positive or -1 (infinite)");
- }
- this.barrierHandler = new BarrierBuffer(inputGate, ioManager, maxAlign);
- }
- else if (checkpointMode == CheckpointingMode.AT_LEAST_ONCE) {
- this.barrierHandler = new BarrierTracker(inputGate);
- }
- else {
- throw new IllegalArgumentException("Unrecognized Checkpointing Mode: " + checkpointMode);
- }
-
- if (checkpointedTask != null) {
- this.barrierHandler.registerCheckpointEventHandler(checkpointedTask);
- }
+ this.barrierHandler = InputProcessorUtil.createCheckpointBarrierHandler(
+ checkpointedTask, checkpointMode, ioManager, inputGate, taskManagerConfig);
this.lock = checkNotNull(lock);
@@ -157,7 +138,7 @@ public class StreamInputProcessor<IN> {
for (int i = 0; i < recordDeserializers.length; i++) {
recordDeserializers[i] = new SpillingAdaptiveSpanningRecordDeserializer<>(
- ioManager.getSpillingDirectoriesPaths());
+ ioManager.getSpillingDirectoriesPaths());
}
this.numInputChannels = inputGate.getNumberOfInputChannels();
http://git-wip-us.apache.org/repos/asf/flink/blob/3126bf52/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTwoInputProcessor.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTwoInputProcessor.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTwoInputProcessor.java
index 494a82a..ab4f90d 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTwoInputProcessor.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTwoInputProcessor.java
@@ -21,8 +21,6 @@ package org.apache.flink.streaming.runtime.io;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.IllegalConfigurationException;
-import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.SimpleCounter;
import org.apache.flink.runtime.event.AbstractEvent;
@@ -146,25 +144,8 @@ public class StreamTwoInputProcessor<IN1, IN2> {
final InputGate inputGate = InputGateUtil.createInputGate(inputGates1, inputGates2);
- if (checkpointMode == CheckpointingMode.EXACTLY_ONCE) {
- long maxAlign = taskManagerConfig.getLong(TaskManagerOptions.TASK_CHECKPOINT_ALIGNMENT_BYTES_LIMIT);
- if (!(maxAlign == -1 || maxAlign > 0)) {
- throw new IllegalConfigurationException(
- TaskManagerOptions.TASK_CHECKPOINT_ALIGNMENT_BYTES_LIMIT.key()
- + " must be positive or -1 (infinite)");
- }
- this.barrierHandler = new BarrierBuffer(inputGate, ioManager, maxAlign);
- }
- else if (checkpointMode == CheckpointingMode.AT_LEAST_ONCE) {
- this.barrierHandler = new BarrierTracker(inputGate);
- }
- else {
- throw new IllegalArgumentException("Unrecognized CheckpointingMode: " + checkpointMode);
- }
-
- if (checkpointedTask != null) {
- this.barrierHandler.registerCheckpointEventHandler(checkpointedTask);
- }
+ this.barrierHandler = InputProcessorUtil.createCheckpointBarrierHandler(
+ checkpointedTask, checkpointMode, ioManager, inputGate, taskManagerConfig);
this.lock = checkNotNull(lock);
@@ -179,7 +160,7 @@ public class StreamTwoInputProcessor<IN1, IN2> {
for (int i = 0; i < recordDeserializers.length; i++) {
recordDeserializers[i] = new SpillingAdaptiveSpanningRecordDeserializer<>(
- ioManager.getSpillingDirectoriesPaths());
+ ioManager.getSpillingDirectoriesPaths());
}
// determine which unioned channels belong to input 1 and which belong to input 2
http://git-wip-us.apache.org/repos/asf/flink/blob/3126bf52/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BarrierBufferAlignmentLimitTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BarrierBufferAlignmentLimitTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BarrierBufferAlignmentLimitTest.java
index c31d5ad..9f46ed7 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BarrierBufferAlignmentLimitTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BarrierBufferAlignmentLimitTest.java
@@ -115,7 +115,7 @@ public class BarrierBufferAlignmentLimitTest {
// the barrier buffer has a limit that only 1000 bytes may be spilled in alignment
MockInputGate gate = new MockInputGate(PAGE_SIZE, 3, Arrays.asList(sequence));
- BarrierBuffer buffer = new BarrierBuffer(gate, ioManager, 1000);
+ BarrierBuffer buffer = new BarrierBuffer(gate, new BufferSpiller(ioManager, gate.getPageSize()), 1000);
AbstractInvokable toNotify = mock(AbstractInvokable.class);
buffer.registerCheckpointEventHandler(toNotify);
@@ -209,7 +209,7 @@ public class BarrierBufferAlignmentLimitTest {
// the barrier buffer has a limit that only 1000 bytes may be spilled in alignment
MockInputGate gate = new MockInputGate(PAGE_SIZE, 3, Arrays.asList(sequence));
- BarrierBuffer buffer = new BarrierBuffer(gate, ioManager, 500);
+ BarrierBuffer buffer = new BarrierBuffer(gate, new BufferSpiller(ioManager, gate.getPageSize()), 500);
AbstractInvokable toNotify = mock(AbstractInvokable.class);
buffer.registerCheckpointEventHandler(toNotify);
http://git-wip-us.apache.org/repos/asf/flink/blob/3126bf52/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BarrierBufferMassiveRandomTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BarrierBufferMassiveRandomTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BarrierBufferMassiveRandomTest.java
index 39c41ef..6dd1e5e 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BarrierBufferMassiveRandomTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BarrierBufferMassiveRandomTest.java
@@ -63,7 +63,7 @@ public class BarrierBufferMassiveRandomTest {
new BufferPool[] { pool1, pool2 },
new BarrierGenerator[] { new CountBarrier(100000), new RandomBarrier(100000) });
- BarrierBuffer barrierBuffer = new BarrierBuffer(myIG, ioMan);
+ BarrierBuffer barrierBuffer = new BarrierBuffer(myIG, new BufferSpiller(ioMan, myIG.getPageSize()));
for (int i = 0; i < 2000000; i++) {
BufferOrEvent boe = barrierBuffer.getNextNonBlocked();