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:25 UTC
[2/4] flink git commit: [FLINK-8547][network] Implement
CheckpointBarrierHandler not to spill data for exactly-once
http://git-wip-us.apache.org/repos/asf/flink/blob/3126bf52/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BarrierBufferTestBase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BarrierBufferTestBase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BarrierBufferTestBase.java
new file mode 100644
index 0000000..b2ed781
--- /dev/null
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BarrierBufferTestBase.java
@@ -0,0 +1,1426 @@
+/*
+ * 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.core.memory.MemorySegment;
+import org.apache.flink.core.memory.MemorySegmentFactory;
+import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
+import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.checkpoint.decline.CheckpointDeclineOnCancellationBarrierException;
+import org.apache.flink.runtime.checkpoint.decline.CheckpointDeclineSubsumedException;
+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;
+import org.apache.flink.runtime.io.network.buffer.Buffer;
+import org.apache.flink.runtime.io.network.buffer.FreeingBufferRecycler;
+import org.apache.flink.runtime.io.network.buffer.NetworkBuffer;
+import org.apache.flink.runtime.io.network.partition.consumer.BufferOrEvent;
+import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
+import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
+import org.apache.flink.runtime.operators.testutils.DummyEnvironment;
+
+import org.hamcrest.BaseMatcher;
+import org.hamcrest.Description;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Random;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.argThat;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+/**
+ * Tests for the behavior of the {@link BarrierBuffer} with different {@link BufferBlocker} implements.
+ */
+public abstract class BarrierBufferTestBase {
+
+ protected static final int PAGE_SIZE = 512;
+
+ private static final Random RND = new Random();
+
+ private static int sizeCounter = 1;
+
+ abstract BarrierBuffer createBarrierHandler(InputGate gate) throws IOException;
+
+ abstract void validateAlignmentBuffered(long actualBytesBuffered, BufferOrEvent... sequence);
+
+ // ------------------------------------------------------------------------
+ // Tests
+ // ------------------------------------------------------------------------
+
+ /**
+ * Validates that the buffer behaves correctly if no checkpoint barriers come,
+ * for a single input channel.
+ */
+ @Test
+ public void testSingleChannelNoBarriers() throws Exception {
+ BufferOrEvent[] sequence = {
+ createBuffer(0, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBuffer(0, PAGE_SIZE), createEndOfPartition(0)
+ };
+
+ MockInputGate gate = new MockInputGate(PAGE_SIZE, 1, Arrays.asList(sequence));
+ BarrierBuffer buffer = createBarrierHandler(gate);
+
+ for (BufferOrEvent boe : sequence) {
+ assertEquals(boe, buffer.getNextNonBlocked());
+ }
+
+ assertEquals(0L, buffer.getAlignmentDurationNanos());
+
+ assertNull(buffer.getNextNonBlocked());
+ assertNull(buffer.getNextNonBlocked());
+
+ buffer.cleanup();
+ }
+
+ /**
+ * Validates that the buffer behaves correctly if no checkpoint barriers come,
+ * for an input with multiple input channels.
+ */
+ @Test
+ public void testMultiChannelNoBarriers() throws Exception {
+ BufferOrEvent[] sequence = {
+ createBuffer(2, PAGE_SIZE), createBuffer(2, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBuffer(1, PAGE_SIZE), createBuffer(0, PAGE_SIZE), createEndOfPartition(0),
+ createBuffer(3, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createEndOfPartition(3),
+ createBuffer(1, PAGE_SIZE), createEndOfPartition(1), createBuffer(2, PAGE_SIZE), createEndOfPartition(2)
+ };
+
+ MockInputGate gate = new MockInputGate(PAGE_SIZE, 4, Arrays.asList(sequence));
+ BarrierBuffer buffer = createBarrierHandler(gate);
+
+ for (BufferOrEvent boe : sequence) {
+ assertEquals(boe, buffer.getNextNonBlocked());
+ }
+
+ assertEquals(0L, buffer.getAlignmentDurationNanos());
+
+ assertNull(buffer.getNextNonBlocked());
+ assertNull(buffer.getNextNonBlocked());
+
+ buffer.cleanup();
+ }
+
+ /**
+ * Validates that the buffer preserved the order of elements for a
+ * input with a single input channel, and checkpoint events.
+ */
+ @Test
+ public void testSingleChannelWithBarriers() throws Exception {
+ BufferOrEvent[] sequence = {
+ createBuffer(0, PAGE_SIZE), createBuffer(0, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(1, 0),
+ createBuffer(0, PAGE_SIZE), createBuffer(0, PAGE_SIZE), createBuffer(0, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(2, 0), createBarrier(3, 0),
+ createBuffer(0, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(4, 0), createBarrier(5, 0), createBarrier(6, 0),
+ createBuffer(0, PAGE_SIZE), createEndOfPartition(0)
+ };
+
+ MockInputGate gate = new MockInputGate(PAGE_SIZE, 1, Arrays.asList(sequence));
+ BarrierBuffer buffer = createBarrierHandler(gate);
+
+ ValidatingCheckpointHandler handler = new ValidatingCheckpointHandler();
+ buffer.registerCheckpointEventHandler(handler);
+ handler.setNextExpectedCheckpointId(1L);
+
+ for (BufferOrEvent boe : sequence) {
+ if (boe.isBuffer() || boe.getEvent().getClass() != CheckpointBarrier.class) {
+ assertEquals(boe, buffer.getNextNonBlocked());
+ }
+ }
+
+ assertNull(buffer.getNextNonBlocked());
+ assertNull(buffer.getNextNonBlocked());
+
+ buffer.cleanup();
+ }
+
+ /**
+ * Validates that the buffer correctly aligns the streams for inputs with
+ * multiple input channels, by buffering and blocking certain inputs.
+ */
+ @Test
+ public void testMultiChannelWithBarriers() throws Exception {
+ BufferOrEvent[] sequence = {
+ // checkpoint with blocked data
+ createBuffer(0, PAGE_SIZE), createBuffer(2, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(1, 1), createBarrier(1, 2),
+ createBuffer(2, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(1, 0),
+
+ // checkpoint without blocked data
+ createBuffer(0, PAGE_SIZE), createBuffer(0, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(2, PAGE_SIZE),
+ createBarrier(2, 0), createBarrier(2, 1), createBarrier(2, 2),
+
+ // checkpoint with data only from one channel
+ createBuffer(2, PAGE_SIZE), createBuffer(2, PAGE_SIZE),
+ createBarrier(3, 2),
+ createBuffer(2, PAGE_SIZE), createBuffer(2, PAGE_SIZE),
+ createBarrier(3, 0), createBarrier(3, 1),
+
+ // empty checkpoint
+ createBarrier(4, 1), createBarrier(4, 2), createBarrier(4, 0),
+
+ // checkpoint with blocked data in mixed order
+ createBuffer(0, PAGE_SIZE), createBuffer(2, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(5, 1),
+ createBuffer(2, PAGE_SIZE), createBuffer(0, PAGE_SIZE), createBuffer(2, PAGE_SIZE), createBuffer(1, PAGE_SIZE),
+ createBarrier(5, 2),
+ createBuffer(1, PAGE_SIZE), createBuffer(0, PAGE_SIZE), createBuffer(2, PAGE_SIZE), createBuffer(1, PAGE_SIZE),
+ createBarrier(5, 0),
+
+ // some trailing data
+ createBuffer(0, PAGE_SIZE),
+ createEndOfPartition(0), createEndOfPartition(1), createEndOfPartition(2)
+ };
+
+ MockInputGate gate = new MockInputGate(PAGE_SIZE, 3, Arrays.asList(sequence));
+ BarrierBuffer buffer = createBarrierHandler(gate);
+
+ ValidatingCheckpointHandler handler = new ValidatingCheckpointHandler();
+ buffer.registerCheckpointEventHandler(handler);
+ handler.setNextExpectedCheckpointId(1L);
+
+ // pre checkpoint 1
+ check(sequence[0], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[1], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[2], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(1L, handler.getNextExpectedCheckpointId());
+
+ long startTs = System.nanoTime();
+
+ // blocking while aligning for checkpoint 1
+ check(sequence[7], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(1L, handler.getNextExpectedCheckpointId());
+
+ // checkpoint 1 done, returning buffered data
+ check(sequence[5], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(2L, handler.getNextExpectedCheckpointId());
+ validateAlignmentTime(startTs, buffer.getAlignmentDurationNanos());
+ validateAlignmentBuffered(handler.getLastReportedBytesBufferedInAlignment(), sequence[5], sequence[6]);
+
+ check(sequence[6], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // pre checkpoint 2
+ check(sequence[9], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[10], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[11], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[12], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[13], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(2L, handler.getNextExpectedCheckpointId());
+
+ // checkpoint 2 barriers come together
+ startTs = System.nanoTime();
+ check(sequence[17], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(3L, handler.getNextExpectedCheckpointId());
+ validateAlignmentTime(startTs, buffer.getAlignmentDurationNanos());
+ validateAlignmentBuffered(handler.getLastReportedBytesBufferedInAlignment());
+
+ check(sequence[18], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // checkpoint 3 starts, data buffered
+ check(sequence[20], buffer.getNextNonBlocked(), PAGE_SIZE);
+ validateAlignmentBuffered(handler.getLastReportedBytesBufferedInAlignment(), sequence[20], sequence[21]);
+ assertEquals(4L, handler.getNextExpectedCheckpointId());
+ check(sequence[21], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // checkpoint 4 happens without extra data
+
+ // pre checkpoint 5
+ check(sequence[27], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ validateAlignmentBuffered(handler.getLastReportedBytesBufferedInAlignment());
+ assertEquals(5L, handler.getNextExpectedCheckpointId());
+
+ check(sequence[28], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[29], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // checkpoint 5 aligning
+ check(sequence[31], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[32], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[33], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[37], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // buffered data from checkpoint 5 alignment
+ check(sequence[34], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[36], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[38], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[39], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // remaining data
+ check(sequence[41], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[42], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[43], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[44], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ assertNull(buffer.getNextNonBlocked());
+ assertNull(buffer.getNextNonBlocked());
+
+ validateAlignmentBuffered(handler.getLastReportedBytesBufferedInAlignment(),
+ sequence[34], sequence[36], sequence[38], sequence[39]);
+
+ buffer.cleanup();
+ }
+
+ @Test
+ public void testMultiChannelTrailingBlockedData() throws Exception {
+ BufferOrEvent[] sequence = {
+ createBuffer(0, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(2, PAGE_SIZE),
+ createBarrier(1, 1), createBarrier(1, 2), createBarrier(1, 0),
+
+ createBuffer(2, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(2, 1),
+ createBuffer(1, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createEndOfPartition(1), createBuffer(0, PAGE_SIZE), createBuffer(2, PAGE_SIZE),
+ createBarrier(2, 2),
+ createBuffer(2, PAGE_SIZE), createEndOfPartition(2), createBuffer(0, PAGE_SIZE), createEndOfPartition(0)
+ };
+
+ MockInputGate gate = new MockInputGate(PAGE_SIZE, 3, Arrays.asList(sequence));
+ BarrierBuffer buffer = createBarrierHandler(gate);
+
+ ValidatingCheckpointHandler handler = new ValidatingCheckpointHandler();
+ buffer.registerCheckpointEventHandler(handler);
+ handler.setNextExpectedCheckpointId(1L);
+
+ // pre-checkpoint 1
+ check(sequence[0], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[1], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[2], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(1L, handler.getNextExpectedCheckpointId());
+
+ // pre-checkpoint 2
+ check(sequence[6], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(2L, handler.getNextExpectedCheckpointId());
+ check(sequence[7], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[8], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // checkpoint 2 alignment
+ long startTs = System.nanoTime();
+ check(sequence[13], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[14], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[18], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[19], buffer.getNextNonBlocked(), PAGE_SIZE);
+ validateAlignmentTime(startTs, buffer.getAlignmentDurationNanos());
+
+ // end of stream: remaining buffered contents
+ check(sequence[10], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[11], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[12], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[16], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[17], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ assertNull(buffer.getNextNonBlocked());
+ assertNull(buffer.getNextNonBlocked());
+
+ buffer.cleanup();
+ }
+
+ /**
+ * Validates that the buffer correctly aligns the streams in cases
+ * where some channels receive barriers from multiple successive checkpoints
+ * before the pending checkpoint is complete.
+ */
+ @Test
+ public void testMultiChannelWithQueuedFutureBarriers() throws Exception{
+ BufferOrEvent[] sequence = {
+ // checkpoint 1 - with blocked data
+ createBuffer(0, PAGE_SIZE), createBuffer(2, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(1, 1), createBarrier(1, 2),
+ createBuffer(2, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(1, 0),
+ createBuffer(1, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+
+ // checkpoint 2 - where future checkpoint barriers come before
+ // the current checkpoint is complete
+ createBarrier(2, 1),
+ createBuffer(1, PAGE_SIZE), createBuffer(2, PAGE_SIZE), createBarrier(2, 0),
+ createBarrier(3, 0), createBuffer(0, PAGE_SIZE),
+ createBarrier(3, 1), createBuffer(0, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(2, PAGE_SIZE),
+ createBarrier(4, 1), createBuffer(1, PAGE_SIZE), createBuffer(2, PAGE_SIZE),
+
+ // complete checkpoint 2, send a barrier for checkpoints 4 and 5
+ createBarrier(2, 2),
+ createBuffer(2, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(2, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(4, 0),
+ createBuffer(2, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(2, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(5, 1),
+
+ // complete checkpoint 3
+ createBarrier(3, 2),
+ createBuffer(2, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(2, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(6, 1),
+
+ // complete checkpoint 4, checkpoint 5 remains not fully triggered
+ createBarrier(4, 2),
+ createBuffer(2, PAGE_SIZE),
+ createBuffer(1, PAGE_SIZE), createEndOfPartition(1),
+ createBuffer(2, PAGE_SIZE), createEndOfPartition(2),
+ createBuffer(0, PAGE_SIZE), createEndOfPartition(0)
+ };
+
+ MockInputGate gate = new MockInputGate(PAGE_SIZE, 3, Arrays.asList(sequence));
+ BarrierBuffer buffer = createBarrierHandler(gate);
+
+ ValidatingCheckpointHandler handler = new ValidatingCheckpointHandler();
+ buffer.registerCheckpointEventHandler(handler);
+ handler.setNextExpectedCheckpointId(1L);
+
+ // around checkpoint 1
+ check(sequence[0], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[1], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[2], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[7], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ check(sequence[5], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(2L, handler.getNextExpectedCheckpointId());
+ check(sequence[6], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[9], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[10], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // alignment of checkpoint 2 - buffering also some barriers for
+ // checkpoints 3 and 4
+ long startTs = System.nanoTime();
+ check(sequence[13], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[20], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[23], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // checkpoint 2 completed
+ check(sequence[12], buffer.getNextNonBlocked(), PAGE_SIZE);
+ validateAlignmentTime(startTs, buffer.getAlignmentDurationNanos());
+ check(sequence[25], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[27], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[30], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[32], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // checkpoint 3 completed (emit buffered)
+ check(sequence[16], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[18], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[19], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[28], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // past checkpoint 3
+ check(sequence[36], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[38], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // checkpoint 4 completed (emit buffered)
+ check(sequence[22], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[26], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[31], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[33], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[39], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // past checkpoint 4, alignment for checkpoint 5
+ check(sequence[42], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[45], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[46], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // abort checkpoint 5 (end of partition)
+ check(sequence[37], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // start checkpoint 6 alignment
+ check(sequence[47], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[48], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // end of input, emit remainder
+ check(sequence[43], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[44], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ assertNull(buffer.getNextNonBlocked());
+ assertNull(buffer.getNextNonBlocked());
+
+ buffer.cleanup();
+ }
+
+ /**
+ * Validates that the buffer skips over the current checkpoint if it
+ * receives a barrier from a later checkpoint on a non-blocked input.
+ */
+ @Test
+ public void testMultiChannelSkippingCheckpoints() throws Exception {
+ BufferOrEvent[] sequence = {
+ // checkpoint 1 - with blocked data
+ createBuffer(0, PAGE_SIZE), createBuffer(2, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(1, 1), createBarrier(1, 2),
+ createBuffer(2, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(1, 0),
+ createBuffer(1, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+
+ // checkpoint 2 will not complete: pre-mature barrier from checkpoint 3
+ createBarrier(2, 1),
+ createBuffer(1, PAGE_SIZE), createBuffer(2, PAGE_SIZE),
+ createBarrier(2, 0),
+ createBuffer(2, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(3, 2),
+
+ createBuffer(2, PAGE_SIZE),
+ createBuffer(1, PAGE_SIZE), createEndOfPartition(1),
+ createBuffer(2, PAGE_SIZE), createEndOfPartition(2),
+ createBuffer(0, PAGE_SIZE), createEndOfPartition(0)
+ };
+
+ MockInputGate gate = new MockInputGate(PAGE_SIZE, 3, Arrays.asList(sequence));
+ BarrierBuffer buffer = createBarrierHandler(gate);
+
+ AbstractInvokable toNotify = mock(AbstractInvokable.class);
+ buffer.registerCheckpointEventHandler(toNotify);
+
+ long startTs;
+
+ // initial data
+ check(sequence[0], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[1], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[2], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // align checkpoint 1
+ startTs = System.nanoTime();
+ check(sequence[7], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(1L, buffer.getCurrentCheckpointId());
+
+ // checkpoint done - replay buffered
+ check(sequence[5], buffer.getNextNonBlocked(), PAGE_SIZE);
+ validateAlignmentTime(startTs, buffer.getAlignmentDurationNanos());
+ verify(toNotify).triggerCheckpointOnBarrier(argThat(new CheckpointMatcher(1L)), any(CheckpointOptions.class), any(CheckpointMetrics.class));
+ check(sequence[6], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ check(sequence[9], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[10], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // alignment of checkpoint 2
+ startTs = System.nanoTime();
+ check(sequence[13], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[15], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // checkpoint 2 aborted, checkpoint 3 started
+ check(sequence[12], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(3L, buffer.getCurrentCheckpointId());
+ validateAlignmentTime(startTs, buffer.getAlignmentDurationNanos());
+ verify(toNotify).abortCheckpointOnBarrier(eq(2L), any(CheckpointDeclineSubsumedException.class));
+ check(sequence[16], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // checkpoint 3 alignment in progress
+ check(sequence[19], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // checkpoint 3 aborted (end of partition)
+ check(sequence[20], buffer.getNextNonBlocked(), PAGE_SIZE);
+ verify(toNotify).abortCheckpointOnBarrier(eq(3L), any(CheckpointDeclineSubsumedException.class));
+
+ // replay buffered data from checkpoint 3
+ check(sequence[18], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // all the remaining messages
+ check(sequence[21], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[22], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[23], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[24], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ assertNull(buffer.getNextNonBlocked());
+ assertNull(buffer.getNextNonBlocked());
+
+ buffer.cleanup();
+ }
+
+ /**
+ * Validates that the buffer skips over the current checkpoint if it
+ * receives a barrier from a later checkpoint on a non-blocked input.
+ */
+ @Test
+ public void testMultiChannelJumpingOverCheckpoint() throws Exception {
+ BufferOrEvent[] sequence = {
+ // checkpoint 1 - with blocked data
+ createBuffer(0, PAGE_SIZE), createBuffer(2, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(1, 1), createBarrier(1, 2),
+ createBuffer(2, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(1, 0),
+ createBuffer(1, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+
+ // checkpoint 2 will not complete: pre-mature barrier from checkpoint 3
+ createBarrier(2, 1),
+ createBuffer(1, PAGE_SIZE), createBuffer(2, PAGE_SIZE),
+ createBarrier(2, 0),
+ createBuffer(2, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(3, 1),
+ createBuffer(1, PAGE_SIZE), createBuffer(2, PAGE_SIZE),
+ createBarrier(3, 0),
+ createBuffer(2, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(4, 2),
+
+ createBuffer(2, PAGE_SIZE),
+ createBuffer(1, PAGE_SIZE), createEndOfPartition(1),
+ createBuffer(2, PAGE_SIZE), createEndOfPartition(2),
+ createBuffer(0, PAGE_SIZE), createEndOfPartition(0)
+ };
+
+ MockInputGate gate = new MockInputGate(PAGE_SIZE, 3, Arrays.asList(sequence));
+ BarrierBuffer buffer = createBarrierHandler(gate);
+
+ ValidatingCheckpointHandler handler = new ValidatingCheckpointHandler();
+ buffer.registerCheckpointEventHandler(handler);
+ handler.setNextExpectedCheckpointId(1L);
+
+ // checkpoint 1
+ check(sequence[0], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[1], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[2], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[7], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(1L, buffer.getCurrentCheckpointId());
+
+ check(sequence[5], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[6], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[9], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[10], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // alignment of checkpoint 2
+ check(sequence[13], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(2L, buffer.getCurrentCheckpointId());
+ check(sequence[15], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[19], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[21], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ long startTs = System.nanoTime();
+
+ // checkpoint 2 aborted, checkpoint 4 started. replay buffered
+ check(sequence[12], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(4L, buffer.getCurrentCheckpointId());
+ check(sequence[16], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[18], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[22], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // align checkpoint 4 remainder
+ check(sequence[25], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[26], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ validateAlignmentTime(startTs, buffer.getAlignmentDurationNanos());
+
+ // checkpoint 4 aborted (due to end of partition)
+ check(sequence[24], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[27], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[28], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[29], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[30], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ assertNull(buffer.getNextNonBlocked());
+ assertNull(buffer.getNextNonBlocked());
+
+ buffer.cleanup();
+ }
+
+ /**
+ * Validates that the buffer skips over a later checkpoint if it
+ * receives a barrier from an even later checkpoint on a blocked input.
+ */
+ @Test
+ public void testMultiChannelSkippingCheckpointsViaBlockedInputs() throws Exception {
+ BufferOrEvent[] sequence = {
+ // checkpoint 1 - with blocked data
+ createBuffer(0, PAGE_SIZE), createBuffer(2, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(1, 1), createBarrier(1, 2),
+ createBuffer(2, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(1, 0),
+ createBuffer(1, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+
+ // checkpoint 2 will not complete: pre-mature barrier from checkpoint 3
+ createBarrier(2, 1),
+ createBuffer(1, PAGE_SIZE), createBuffer(2, PAGE_SIZE),
+ createBarrier(2, 0),
+ createBuffer(1, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+
+ createBarrier(3, 0), // queued barrier on blocked input
+ createBuffer(0, PAGE_SIZE),
+
+ createBarrier(4, 1), // pre-mature barrier on blocked input
+ createBuffer(1, PAGE_SIZE),
+ createBuffer(0, PAGE_SIZE),
+ createBuffer(2, PAGE_SIZE),
+
+ // complete checkpoint 2
+ createBarrier(2, 2),
+ createBuffer(0, PAGE_SIZE),
+
+ createBarrier(3, 2), // should be ignored
+ createBuffer(2, PAGE_SIZE),
+ createBarrier(4, 0),
+ createBuffer(0, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(2, PAGE_SIZE),
+ createBarrier(4, 2),
+
+ createBuffer(1, PAGE_SIZE), createEndOfPartition(1),
+ createBuffer(2, PAGE_SIZE), createEndOfPartition(2),
+ createBuffer(0, PAGE_SIZE), createEndOfPartition(0)
+ };
+
+ MockInputGate gate = new MockInputGate(PAGE_SIZE, 3, Arrays.asList(sequence));
+ BarrierBuffer buffer = createBarrierHandler(gate);
+
+ // checkpoint 1
+ check(sequence[0], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[1], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[2], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[7], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(1L, buffer.getCurrentCheckpointId());
+ check(sequence[5], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[6], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[9], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[10], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // alignment of checkpoint 2
+ check(sequence[13], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[22], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(2L, buffer.getCurrentCheckpointId());
+
+ // checkpoint 2 completed
+ check(sequence[12], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[15], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[16], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // checkpoint 3 skipped, alignment for 4 started
+ check(sequence[18], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(4L, buffer.getCurrentCheckpointId());
+ check(sequence[21], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[24], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[26], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[30], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // checkpoint 4 completed
+ check(sequence[20], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[28], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[29], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ check(sequence[32], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[33], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[34], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[35], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[36], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[37], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ assertNull(buffer.getNextNonBlocked());
+ assertNull(buffer.getNextNonBlocked());
+
+ buffer.cleanup();
+ }
+
+ @Test
+ public void testEarlyCleanup() throws Exception {
+ BufferOrEvent[] sequence = {
+ createBuffer(0, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(2, PAGE_SIZE),
+ createBarrier(1, 1), createBarrier(1, 2), createBarrier(1, 0),
+
+ createBuffer(2, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(2, 1),
+ createBuffer(1, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createEndOfPartition(1), createBuffer(0, PAGE_SIZE), createBuffer(2, PAGE_SIZE),
+ createBarrier(2, 2),
+ createBuffer(2, PAGE_SIZE), createEndOfPartition(2), createBuffer(0, PAGE_SIZE), createEndOfPartition(0)
+ };
+
+ MockInputGate gate = new MockInputGate(PAGE_SIZE, 3, Arrays.asList(sequence));
+ BarrierBuffer buffer = createBarrierHandler(gate);
+
+ ValidatingCheckpointHandler handler = new ValidatingCheckpointHandler();
+ buffer.registerCheckpointEventHandler(handler);
+ handler.setNextExpectedCheckpointId(1L);
+
+ // pre-checkpoint 1
+ check(sequence[0], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[1], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[2], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(1L, handler.getNextExpectedCheckpointId());
+
+ // pre-checkpoint 2
+ check(sequence[6], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(2L, handler.getNextExpectedCheckpointId());
+ check(sequence[7], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[8], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // checkpoint 2 alignment
+ check(sequence[13], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[14], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[18], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[19], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // end of stream: remaining buffered contents
+ buffer.getNextNonBlocked();
+ buffer.cleanup();
+ }
+
+ @Test
+ public void testStartAlignmentWithClosedChannels() throws Exception {
+ BufferOrEvent[] sequence = {
+ // close some channels immediately
+ createEndOfPartition(2), createEndOfPartition(1),
+
+ // checkpoint without blocked data
+ createBuffer(0, PAGE_SIZE), createBuffer(0, PAGE_SIZE), createBuffer(3, PAGE_SIZE),
+ createBarrier(2, 3), createBarrier(2, 0),
+
+ // checkpoint with blocked data
+ createBuffer(3, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(3, 3),
+ createBuffer(3, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ createBarrier(3, 0),
+
+ // empty checkpoint
+ createBarrier(4, 0), createBarrier(4, 3),
+
+ // some data, one channel closes
+ createBuffer(0, PAGE_SIZE), createBuffer(0, PAGE_SIZE), createBuffer(3, PAGE_SIZE),
+ createEndOfPartition(0),
+
+ // checkpoint on last remaining channel
+ createBuffer(3, PAGE_SIZE),
+ createBarrier(5, 3),
+ createBuffer(3, PAGE_SIZE),
+ createEndOfPartition(3)
+ };
+
+ MockInputGate gate = new MockInputGate(PAGE_SIZE, 4, Arrays.asList(sequence));
+
+ BarrierBuffer buffer = createBarrierHandler(gate);
+
+ // pre checkpoint 2
+ check(sequence[0], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[1], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[2], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[3], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[4], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // checkpoint 3 alignment
+ check(sequence[7], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(2L, buffer.getCurrentCheckpointId());
+ check(sequence[8], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[11], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // checkpoint 3 buffered
+ check(sequence[10], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(3L, buffer.getCurrentCheckpointId());
+
+ // after checkpoint 4
+ check(sequence[15], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(4L, buffer.getCurrentCheckpointId());
+ check(sequence[16], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[17], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[18], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ check(sequence[19], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[21], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(5L, buffer.getCurrentCheckpointId());
+ check(sequence[22], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ assertNull(buffer.getNextNonBlocked());
+ assertNull(buffer.getNextNonBlocked());
+
+ buffer.cleanup();
+ }
+
+ @Test
+ public void testEndOfStreamWhileCheckpoint() throws Exception {
+ BufferOrEvent[] sequence = {
+ // one checkpoint
+ createBarrier(1, 0), createBarrier(1, 1), createBarrier(1, 2),
+
+ // some buffers
+ createBuffer(0, PAGE_SIZE), createBuffer(0, PAGE_SIZE), createBuffer(2, PAGE_SIZE),
+
+ // start the checkpoint that will be incomplete
+ createBarrier(2, 2), createBarrier(2, 0),
+ createBuffer(0, PAGE_SIZE), createBuffer(2, PAGE_SIZE), createBuffer(1, PAGE_SIZE),
+
+ // close one after the barrier one before the barrier
+ createEndOfPartition(2), createEndOfPartition(1),
+ createBuffer(0, PAGE_SIZE),
+
+ // final end of stream
+ createEndOfPartition(0)
+ };
+
+ MockInputGate gate = new MockInputGate(PAGE_SIZE, 3, Arrays.asList(sequence));
+ BarrierBuffer buffer = createBarrierHandler(gate);
+
+ // data after first checkpoint
+ check(sequence[3], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[4], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[5], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(1L, buffer.getCurrentCheckpointId());
+
+ // alignment of second checkpoint
+ check(sequence[10], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(2L, buffer.getCurrentCheckpointId());
+
+ // first end-of-partition encountered: checkpoint will not be completed
+ check(sequence[12], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[8], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[9], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[11], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[13], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[14], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // all done
+ assertNull(buffer.getNextNonBlocked());
+ assertNull(buffer.getNextNonBlocked());
+
+ buffer.cleanup();
+ }
+
+ @Test
+ public void testSingleChannelAbortCheckpoint() throws Exception {
+ BufferOrEvent[] sequence = {
+ createBuffer(0, PAGE_SIZE),
+ createBarrier(1, 0),
+ createBuffer(0, PAGE_SIZE),
+ createBarrier(2, 0),
+ createCancellationBarrier(4, 0),
+ createBarrier(5, 0),
+ createBuffer(0, PAGE_SIZE),
+ createCancellationBarrier(6, 0),
+ createBuffer(0, PAGE_SIZE)
+ };
+
+ MockInputGate gate = new MockInputGate(PAGE_SIZE, 1, Arrays.asList(sequence));
+ BarrierBuffer buffer = createBarrierHandler(gate);
+
+ AbstractInvokable toNotify = mock(AbstractInvokable.class);
+ buffer.registerCheckpointEventHandler(toNotify);
+
+ check(sequence[0], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[2], buffer.getNextNonBlocked(), PAGE_SIZE);
+ verify(toNotify, times(1)).triggerCheckpointOnBarrier(argThat(new CheckpointMatcher(1L)), any(CheckpointOptions.class), any(CheckpointMetrics.class));
+ assertEquals(0L, buffer.getAlignmentDurationNanos());
+
+ check(sequence[6], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(5L, buffer.getCurrentCheckpointId());
+ verify(toNotify, times(1)).triggerCheckpointOnBarrier(argThat(new CheckpointMatcher(2L)), any(CheckpointOptions.class), any(CheckpointMetrics.class));
+ verify(toNotify, times(1)).abortCheckpointOnBarrier(eq(4L), any(CheckpointDeclineOnCancellationBarrierException.class));
+ verify(toNotify, times(1)).triggerCheckpointOnBarrier(argThat(new CheckpointMatcher(5L)), any(CheckpointOptions.class), any(CheckpointMetrics.class));
+ assertEquals(0L, buffer.getAlignmentDurationNanos());
+
+ check(sequence[8], buffer.getNextNonBlocked(), PAGE_SIZE);
+ assertEquals(6L, buffer.getCurrentCheckpointId());
+ verify(toNotify, times(1)).abortCheckpointOnBarrier(eq(6L), any(CheckpointDeclineOnCancellationBarrierException.class));
+ assertEquals(0L, buffer.getAlignmentDurationNanos());
+
+ buffer.cleanup();
+ }
+
+ @Test
+ public void testMultiChannelAbortCheckpoint() throws Exception {
+ BufferOrEvent[] sequence = {
+ // some buffers and a successful checkpoint
+ /* 0 */ createBuffer(0, PAGE_SIZE), createBuffer(2, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+ /* 3 */ createBarrier(1, 1), createBarrier(1, 2),
+ /* 5 */ createBuffer(2, PAGE_SIZE), createBuffer(1, PAGE_SIZE),
+ /* 7 */ createBarrier(1, 0),
+ /* 8 */ createBuffer(0, PAGE_SIZE), createBuffer(2, PAGE_SIZE),
+
+ // aborted on last barrier
+ /* 10 */ createBarrier(2, 0), createBarrier(2, 2),
+ /* 12 */ createBuffer(0, PAGE_SIZE), createBuffer(2, PAGE_SIZE),
+ /* 14 */ createCancellationBarrier(2, 1),
+
+ // successful checkpoint
+ /* 15 */ createBuffer(2, PAGE_SIZE), createBuffer(1, PAGE_SIZE),
+ /* 17 */ createBarrier(3, 1), createBarrier(3, 2), createBarrier(3, 0),
+
+ // abort on first barrier
+ /* 20 */ createBuffer(0, PAGE_SIZE), createBuffer(1, PAGE_SIZE),
+ /* 22 */ createCancellationBarrier(4, 1), createBarrier(4, 2),
+ /* 24 */ createBuffer(0, PAGE_SIZE),
+ /* 25 */ createBarrier(4, 0),
+
+ // another successful checkpoint
+ /* 26 */ createBuffer(0, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(2, PAGE_SIZE),
+ /* 29 */ createBarrier(5, 2), createBarrier(5, 1), createBarrier(5, 0),
+ /* 32 */ createBuffer(0, PAGE_SIZE), createBuffer(1, PAGE_SIZE),
+
+ // abort multiple cancellations and a barrier after the cancellations
+ /* 34 */ createCancellationBarrier(6, 1), createCancellationBarrier(6, 2),
+ /* 36 */ createBarrier(6, 0),
+
+ /* 37 */ createBuffer(0, PAGE_SIZE)
+ };
+
+ MockInputGate gate = new MockInputGate(PAGE_SIZE, 3, Arrays.asList(sequence));
+ BarrierBuffer buffer = createBarrierHandler(gate);
+
+ AbstractInvokable toNotify = mock(AbstractInvokable.class);
+ buffer.registerCheckpointEventHandler(toNotify);
+
+ long startTs;
+
+ // successful first checkpoint, with some aligned buffers
+ check(sequence[0], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[1], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[2], buffer.getNextNonBlocked(), PAGE_SIZE);
+ startTs = System.nanoTime();
+ check(sequence[5], buffer.getNextNonBlocked(), PAGE_SIZE);
+ verify(toNotify, times(1)).triggerCheckpointOnBarrier(argThat(new CheckpointMatcher(1L)), any(CheckpointOptions.class), any(CheckpointMetrics.class));
+ validateAlignmentTime(startTs, buffer.getAlignmentDurationNanos());
+
+ check(sequence[6], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[8], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[9], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // canceled checkpoint on last barrier
+ startTs = System.nanoTime();
+ check(sequence[12], buffer.getNextNonBlocked(), PAGE_SIZE);
+ verify(toNotify, times(1)).abortCheckpointOnBarrier(eq(2L), any(CheckpointDeclineOnCancellationBarrierException.class));
+ validateAlignmentTime(startTs, buffer.getAlignmentDurationNanos());
+ check(sequence[13], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // one more successful checkpoint
+ check(sequence[15], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[16], buffer.getNextNonBlocked(), PAGE_SIZE);
+ startTs = System.nanoTime();
+ check(sequence[20], buffer.getNextNonBlocked(), PAGE_SIZE);
+ verify(toNotify, times(1)).triggerCheckpointOnBarrier(argThat(new CheckpointMatcher(3L)), any(CheckpointOptions.class), any(CheckpointMetrics.class));
+ validateAlignmentTime(startTs, buffer.getAlignmentDurationNanos());
+ check(sequence[21], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // this checkpoint gets immediately canceled
+ check(sequence[24], buffer.getNextNonBlocked(), PAGE_SIZE);
+ verify(toNotify, times(1)).abortCheckpointOnBarrier(eq(4L), any(CheckpointDeclineOnCancellationBarrierException.class));
+ assertEquals(0L, buffer.getAlignmentDurationNanos());
+
+ // some buffers
+ check(sequence[26], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[27], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[28], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // a simple successful checkpoint
+ startTs = System.nanoTime();
+ check(sequence[32], buffer.getNextNonBlocked(), PAGE_SIZE);
+ verify(toNotify, times(1)).triggerCheckpointOnBarrier(argThat(new CheckpointMatcher(5L)), any(CheckpointOptions.class), any(CheckpointMetrics.class));
+ validateAlignmentTime(startTs, buffer.getAlignmentDurationNanos());
+ check(sequence[33], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ check(sequence[37], buffer.getNextNonBlocked(), PAGE_SIZE);
+ verify(toNotify, times(1)).abortCheckpointOnBarrier(eq(6L), any(CheckpointDeclineOnCancellationBarrierException.class));
+ assertEquals(0L, buffer.getAlignmentDurationNanos());
+
+ // all done
+ assertNull(buffer.getNextNonBlocked());
+ assertNull(buffer.getNextNonBlocked());
+
+ buffer.cleanup();
+ }
+
+ @Test
+ public void testAbortViaQueuedBarriers() throws Exception {
+ BufferOrEvent[] sequence = {
+ // starting a checkpoint
+ /* 0 */ createBuffer(1, PAGE_SIZE),
+ /* 1 */ createBarrier(1, 1), createBarrier(1, 2),
+ /* 3 */ createBuffer(2, PAGE_SIZE), createBuffer(0, PAGE_SIZE), createBuffer(1, PAGE_SIZE),
+
+ // queued barrier and cancellation barrier
+ /* 6 */ createCancellationBarrier(2, 2),
+ /* 7 */ createBarrier(2, 1),
+
+ // some intermediate buffers (some queued)
+ /* 8 */ createBuffer(0, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(2, PAGE_SIZE),
+
+ // complete initial checkpoint
+ /* 11 */ createBarrier(1, 0),
+
+ // some buffers (none queued, since checkpoint is aborted)
+ /* 12 */ createBuffer(2, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+
+ // final barrier of aborted checkpoint
+ /* 15 */ createBarrier(2, 0),
+
+ // some more buffers
+ /* 16 */ createBuffer(0, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(2, PAGE_SIZE)
+ };
+
+ MockInputGate gate = new MockInputGate(PAGE_SIZE, 3, Arrays.asList(sequence));
+ BarrierBuffer buffer = createBarrierHandler(gate);
+
+ AbstractInvokable toNotify = mock(AbstractInvokable.class);
+ buffer.registerCheckpointEventHandler(toNotify);
+
+ long startTs;
+
+ check(sequence[0], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // starting first checkpoint
+ startTs = System.nanoTime();
+ check(sequence[4], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[8], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // finished first checkpoint
+ check(sequence[3], buffer.getNextNonBlocked(), PAGE_SIZE);
+ verify(toNotify, times(1)).triggerCheckpointOnBarrier(argThat(new CheckpointMatcher(1L)), any(CheckpointOptions.class), any(CheckpointMetrics.class));
+ validateAlignmentTime(startTs, buffer.getAlignmentDurationNanos());
+
+ check(sequence[5], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // re-read the queued cancellation barriers
+ check(sequence[9], buffer.getNextNonBlocked(), PAGE_SIZE);
+ verify(toNotify, times(1)).abortCheckpointOnBarrier(eq(2L), any(CheckpointDeclineOnCancellationBarrierException.class));
+ assertEquals(0L, buffer.getAlignmentDurationNanos());
+
+ check(sequence[10], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[12], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[13], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[14], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ check(sequence[16], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[17], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[18], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // no further alignment should have happened
+ assertEquals(0L, buffer.getAlignmentDurationNanos());
+
+ // no further checkpoint (abort) notifications
+ verify(toNotify, times(1)).triggerCheckpointOnBarrier(any(CheckpointMetaData.class), any(CheckpointOptions.class), any(CheckpointMetrics.class));
+ verify(toNotify, times(1)).abortCheckpointOnBarrier(anyLong(), any(CheckpointDeclineOnCancellationBarrierException.class));
+
+ // all done
+ assertNull(buffer.getNextNonBlocked());
+ assertNull(buffer.getNextNonBlocked());
+
+ buffer.cleanup();
+ }
+
+ /**
+ * This tests the where a replay of queued checkpoint barriers meets
+ * a canceled checkpoint.
+ *
+ * <p>The replayed newer checkpoint barrier must not try to cancel the
+ * already canceled checkpoint.
+ */
+ @Test
+ public void testAbortWhileHavingQueuedBarriers() throws Exception {
+ BufferOrEvent[] sequence = {
+ // starting a checkpoint
+ /* 0 */ createBuffer(1, PAGE_SIZE),
+ /* 1 */ createBarrier(1, 1),
+ /* 2 */ createBuffer(2, PAGE_SIZE), createBuffer(0, PAGE_SIZE), createBuffer(1, PAGE_SIZE),
+
+ // queued barrier and cancellation barrier
+ /* 5 */ createBarrier(2, 1),
+
+ // some queued buffers
+ /* 6 */ createBuffer(2, PAGE_SIZE), createBuffer(1, PAGE_SIZE),
+
+ // cancel the initial checkpoint
+ /* 8 */ createCancellationBarrier(1, 0),
+
+ // some more buffers
+ /* 9 */ createBuffer(2, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+
+ // ignored barrier - already canceled and moved to next checkpoint
+ /* 12 */ createBarrier(1, 2),
+
+ // some more buffers
+ /* 13 */ createBuffer(0, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(2, PAGE_SIZE),
+
+ // complete next checkpoint regularly
+ /* 16 */ createBarrier(2, 0), createBarrier(2, 2),
+
+ // some more buffers
+ /* 18 */ createBuffer(0, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(2, PAGE_SIZE)
+ };
+
+ MockInputGate gate = new MockInputGate(PAGE_SIZE, 3, Arrays.asList(sequence));
+ BarrierBuffer buffer = createBarrierHandler(gate);
+
+ AbstractInvokable toNotify = mock(AbstractInvokable.class);
+ buffer.registerCheckpointEventHandler(toNotify);
+
+ long startTs;
+
+ check(sequence[0], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // starting first checkpoint
+ startTs = System.nanoTime();
+ check(sequence[2], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[3], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[6], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // cancelled by cancellation barrier
+ check(sequence[4], buffer.getNextNonBlocked(), PAGE_SIZE);
+ validateAlignmentTime(startTs, buffer.getAlignmentDurationNanos());
+ verify(toNotify).abortCheckpointOnBarrier(eq(1L), any(CheckpointDeclineOnCancellationBarrierException.class));
+
+ // the next checkpoint alignment starts now
+ startTs = System.nanoTime();
+ check(sequence[9], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[11], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[13], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[15], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // checkpoint done
+ check(sequence[7], buffer.getNextNonBlocked(), PAGE_SIZE);
+ validateAlignmentTime(startTs, buffer.getAlignmentDurationNanos());
+ verify(toNotify).triggerCheckpointOnBarrier(argThat(new CheckpointMatcher(2L)), any(CheckpointOptions.class), any(CheckpointMetrics.class));
+
+ // queued data
+ check(sequence[10], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[14], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // trailing data
+ check(sequence[18], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[19], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[20], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // all done
+ assertNull(buffer.getNextNonBlocked());
+ assertNull(buffer.getNextNonBlocked());
+
+ buffer.cleanup();
+
+ // check overall notifications
+ verify(toNotify, times(1)).triggerCheckpointOnBarrier(any(CheckpointMetaData.class), any(CheckpointOptions.class), any(CheckpointMetrics.class));
+ verify(toNotify, times(1)).abortCheckpointOnBarrier(anyLong(), any(Throwable.class));
+ }
+
+ /**
+ * This tests the where a cancellation barrier is received for a checkpoint already
+ * canceled due to receiving a newer checkpoint barrier.
+ */
+ @Test
+ public void testIgnoreCancelBarrierIfCheckpointSubsumed() throws Exception {
+ BufferOrEvent[] sequence = {
+ // starting a checkpoint
+ /* 0 */ createBuffer(2, PAGE_SIZE),
+ /* 1 */ createBarrier(3, 1), createBarrier(3, 0),
+ /* 3 */ createBuffer(0, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(2, PAGE_SIZE),
+
+ // newer checkpoint barrier cancels/subsumes pending checkpoint
+ /* 6 */ createBarrier(5, 2),
+
+ // some queued buffers
+ /* 7 */ createBuffer(2, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(0, PAGE_SIZE),
+
+ // cancel barrier the initial checkpoint /it is already canceled)
+ /* 10 */ createCancellationBarrier(3, 2),
+
+ // some more buffers
+ /* 11 */ createBuffer(2, PAGE_SIZE), createBuffer(0, PAGE_SIZE), createBuffer(1, PAGE_SIZE),
+
+ // complete next checkpoint regularly
+ /* 14 */ createBarrier(5, 0), createBarrier(5, 1),
+
+ // some more buffers
+ /* 16 */ createBuffer(0, PAGE_SIZE), createBuffer(1, PAGE_SIZE), createBuffer(2, PAGE_SIZE)
+ };
+
+ MockInputGate gate = new MockInputGate(PAGE_SIZE, 3, Arrays.asList(sequence));
+ BarrierBuffer buffer = createBarrierHandler(gate);
+
+ AbstractInvokable toNotify = mock(AbstractInvokable.class);
+ buffer.registerCheckpointEventHandler(toNotify);
+
+ long startTs;
+
+ // validate the sequence
+
+ check(sequence[0], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // beginning of first checkpoint
+ check(sequence[5], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // future barrier aborts checkpoint
+ startTs = System.nanoTime();
+ check(sequence[3], buffer.getNextNonBlocked(), PAGE_SIZE);
+ verify(toNotify, times(1)).abortCheckpointOnBarrier(eq(3L), any(CheckpointDeclineSubsumedException.class));
+ check(sequence[4], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // alignment of next checkpoint
+ check(sequence[8], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[9], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[12], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[13], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // checkpoint finished
+ check(sequence[7], buffer.getNextNonBlocked(), PAGE_SIZE);
+ validateAlignmentTime(startTs, buffer.getAlignmentDurationNanos());
+ verify(toNotify, times(1)).triggerCheckpointOnBarrier(argThat(new CheckpointMatcher(5L)), any(CheckpointOptions.class), any(CheckpointMetrics.class));
+ check(sequence[11], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // remaining data
+ check(sequence[16], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[17], buffer.getNextNonBlocked(), PAGE_SIZE);
+ check(sequence[18], buffer.getNextNonBlocked(), PAGE_SIZE);
+
+ // all done
+ assertNull(buffer.getNextNonBlocked());
+ assertNull(buffer.getNextNonBlocked());
+
+ buffer.cleanup();
+
+ // check overall notifications
+ verify(toNotify, times(1)).triggerCheckpointOnBarrier(any(CheckpointMetaData.class), any(CheckpointOptions.class), any(CheckpointMetrics.class));
+ verify(toNotify, times(1)).abortCheckpointOnBarrier(anyLong(), any(Throwable.class));
+ }
+
+ // ------------------------------------------------------------------------
+ // Utils
+ // ------------------------------------------------------------------------
+
+ private static BufferOrEvent createBarrier(long checkpointId, int channel) {
+ return new BufferOrEvent(new CheckpointBarrier(
+ checkpointId, System.currentTimeMillis(), CheckpointOptions.forCheckpointWithDefaultLocation()), channel);
+ }
+
+ private static BufferOrEvent createCancellationBarrier(long checkpointId, int channel) {
+ return new BufferOrEvent(new CancelCheckpointMarker(checkpointId), channel);
+ }
+
+ private static BufferOrEvent createBuffer(int channel, int pageSize) {
+ final int size = sizeCounter++;
+ byte[] bytes = new byte[size];
+ RND.nextBytes(bytes);
+
+ MemorySegment memory = MemorySegmentFactory.allocateUnpooledSegment(pageSize);
+ memory.put(0, bytes);
+
+ Buffer buf = new NetworkBuffer(memory, FreeingBufferRecycler.INSTANCE);
+ buf.setSize(size);
+
+ // retain an additional time so it does not get disposed after being read by the input gate
+ buf.retainBuffer();
+
+ return new BufferOrEvent(buf, channel);
+ }
+
+ private static BufferOrEvent createEndOfPartition(int channel) {
+ return new BufferOrEvent(EndOfPartitionEvent.INSTANCE, channel);
+ }
+
+ private static void check(BufferOrEvent expected, BufferOrEvent present, int pageSize) {
+ assertNotNull(expected);
+ assertNotNull(present);
+ assertEquals(expected.isBuffer(), present.isBuffer());
+
+ if (expected.isBuffer()) {
+ assertEquals(expected.getBuffer().getMaxCapacity(), present.getBuffer().getMaxCapacity());
+ assertEquals(expected.getBuffer().getSize(), present.getBuffer().getSize());
+ MemorySegment expectedMem = expected.getBuffer().getMemorySegment();
+ MemorySegment presentMem = present.getBuffer().getMemorySegment();
+ assertTrue("memory contents differs", expectedMem.compare(presentMem, 0, 0, pageSize) == 0);
+ } else {
+ assertEquals(expected.getEvent(), present.getEvent());
+ }
+ }
+
+ private static void validateAlignmentTime(long startTimestamp, long alignmentDuration) {
+ final long elapsed = System.nanoTime() - startTimestamp;
+ assertTrue("wrong alignment time", alignmentDuration <= elapsed);
+ }
+
+ // ------------------------------------------------------------------------
+ // Testing Mocks
+ // ------------------------------------------------------------------------
+
+ /**
+ * The invokable handler used for triggering checkpoint and validation.
+ */
+ private static class ValidatingCheckpointHandler extends AbstractInvokable {
+
+ private long nextExpectedCheckpointId = -1L;
+ private long lastReportedBytesBufferedInAlignment = -1;
+
+ public ValidatingCheckpointHandler() {
+ super(new DummyEnvironment("test", 1, 0));
+ }
+
+ public void setNextExpectedCheckpointId(long nextExpectedCheckpointId) {
+ this.nextExpectedCheckpointId = nextExpectedCheckpointId;
+ }
+
+ public long getNextExpectedCheckpointId() {
+ return nextExpectedCheckpointId;
+ }
+
+ long getLastReportedBytesBufferedInAlignment() {
+ return lastReportedBytesBufferedInAlignment;
+ }
+
+ @Override
+ public void invoke() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean triggerCheckpoint(
+ CheckpointMetaData checkpointMetaData,
+ CheckpointOptions checkpointOptions) throws Exception {
+ throw new UnsupportedOperationException("should never be called");
+ }
+
+ @Override
+ public void triggerCheckpointOnBarrier(
+ CheckpointMetaData checkpointMetaData,
+ CheckpointOptions checkpointOptions,
+ CheckpointMetrics checkpointMetrics) throws Exception {
+ assertTrue("wrong checkpoint id", nextExpectedCheckpointId == -1L ||
+ nextExpectedCheckpointId == checkpointMetaData.getCheckpointId());
+
+ assertTrue(checkpointMetaData.getTimestamp() > 0);
+ assertTrue(checkpointMetrics.getBytesBufferedInAlignment() >= 0);
+ assertTrue(checkpointMetrics.getAlignmentDurationNanos() >= 0);
+
+ nextExpectedCheckpointId++;
+ lastReportedBytesBufferedInAlignment = checkpointMetrics.getBytesBufferedInAlignment();
+ }
+
+ @Override
+ public void abortCheckpointOnBarrier(long checkpointId, Throwable cause) {}
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) throws Exception {
+ throw new UnsupportedOperationException("should never be called");
+ }
+ }
+
+ /**
+ * The matcher used for verifying checkpoint equality.
+ */
+ private static class CheckpointMatcher extends BaseMatcher<CheckpointMetaData> {
+
+ private final long checkpointId;
+
+ CheckpointMatcher(long checkpointId) {
+ this.checkpointId = checkpointId;
+ }
+
+ @Override
+ public boolean matches(Object o) {
+ return o != null &&
+ o.getClass() == CheckpointMetaData.class &&
+ ((CheckpointMetaData) o).getCheckpointId() == checkpointId;
+ }
+
+ @Override
+ public void describeTo(Description description) {
+ description.appendText("CheckpointMetaData - id = " + checkpointId);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/3126bf52/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BufferBlockerTestBase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BufferBlockerTestBase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BufferBlockerTestBase.java
new file mode 100644
index 0000000..4448edd
--- /dev/null
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BufferBlockerTestBase.java
@@ -0,0 +1,325 @@
+/*
+ * 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.core.memory.MemorySegment;
+import org.apache.flink.core.memory.MemorySegmentFactory;
+import org.apache.flink.runtime.io.network.buffer.Buffer;
+import org.apache.flink.runtime.io.network.buffer.FreeingBufferRecycler;
+import org.apache.flink.runtime.io.network.buffer.NetworkBuffer;
+import org.apache.flink.runtime.io.network.partition.consumer.BufferOrEvent;
+
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Random;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * Tests for {@link BufferBlocker}.
+ */
+public abstract class BufferBlockerTestBase {
+
+ protected static final int PAGE_SIZE = 4096;
+
+ abstract BufferBlocker createBufferBlocker();
+
+ @Test
+ public void testRollOverEmptySequences() throws IOException {
+ BufferBlocker bufferBlocker = createBufferBlocker();
+ assertNull(bufferBlocker.rollOverReusingResources());
+ assertNull(bufferBlocker.rollOverReusingResources());
+ assertNull(bufferBlocker.rollOverReusingResources());
+ }
+
+ @Test
+ public void testSpillAndRollOverSimple() throws IOException {
+ final Random rnd = new Random();
+ final Random bufferRnd = new Random();
+
+ final int maxNumEventsAndBuffers = 3000;
+ final int maxNumChannels = 1656;
+
+ BufferBlocker bufferBlocker = createBufferBlocker();
+
+ // do multiple spilling / rolling over rounds
+ for (int round = 0; round < 5; round++) {
+
+ final long bufferSeed = rnd.nextLong();
+ bufferRnd.setSeed(bufferSeed);
+
+ final int numEventsAndBuffers = rnd.nextInt(maxNumEventsAndBuffers) + 1;
+ final int numChannels = rnd.nextInt(maxNumChannels) + 1;
+
+ final ArrayList<BufferOrEvent> events = new ArrayList<BufferOrEvent>(128);
+
+ // generate sequence
+ for (int i = 0; i < numEventsAndBuffers; i++) {
+ boolean isEvent = rnd.nextDouble() < 0.05d;
+ BufferOrEvent evt;
+ if (isEvent) {
+ evt = generateRandomEvent(rnd, numChannels);
+ events.add(evt);
+ } else {
+ evt = generateRandomBuffer(bufferRnd.nextInt(PAGE_SIZE) + 1, bufferRnd.nextInt(numChannels));
+ }
+ bufferBlocker.add(evt);
+ }
+
+ // reset and create reader
+ bufferRnd.setSeed(bufferSeed);
+
+ BufferOrEventSequence seq = bufferBlocker.rollOverReusingResources();
+ seq.open();
+
+ // read and validate the sequence
+
+ int numEvent = 0;
+ for (int i = 0; i < numEventsAndBuffers; i++) {
+ BufferOrEvent next = seq.getNext();
+ assertNotNull(next);
+ if (next.isEvent()) {
+ BufferOrEvent expected = events.get(numEvent++);
+ assertEquals(expected.getEvent(), next.getEvent());
+ assertEquals(expected.getChannelIndex(), next.getChannelIndex());
+ } else {
+ validateBuffer(next, bufferRnd.nextInt(PAGE_SIZE) + 1, bufferRnd.nextInt(numChannels));
+ }
+ }
+
+ // no further data
+ assertNull(seq.getNext());
+
+ // all events need to be consumed
+ assertEquals(events.size(), numEvent);
+
+ seq.cleanup();
+ }
+ }
+
+ @Test
+ public void testSpillWhileReading() throws IOException {
+ final int sequences = 10;
+
+ final Random rnd = new Random();
+
+ final int maxNumEventsAndBuffers = 30000;
+ final int maxNumChannels = 1656;
+
+ int sequencesConsumed = 0;
+
+ ArrayDeque<SequenceToConsume> pendingSequences = new ArrayDeque<SequenceToConsume>();
+ SequenceToConsume currentSequence = null;
+ int currentNumEvents = 0;
+ int currentNumRecordAndEvents = 0;
+
+ BufferBlocker bufferBlocker = createBufferBlocker();
+
+ // do multiple spilling / rolling over rounds
+ for (int round = 0; round < 2 * sequences; round++) {
+
+ if (round % 2 == 1) {
+ // make this an empty sequence
+ assertNull(bufferBlocker.rollOverReusingResources());
+ } else {
+ // proper spilled sequence
+ final long bufferSeed = rnd.nextLong();
+ final Random bufferRnd = new Random(bufferSeed);
+
+ final int numEventsAndBuffers = rnd.nextInt(maxNumEventsAndBuffers) + 1;
+ final int numChannels = rnd.nextInt(maxNumChannels) + 1;
+
+ final ArrayList<BufferOrEvent> events = new ArrayList<BufferOrEvent>(128);
+
+ int generated = 0;
+ while (generated < numEventsAndBuffers) {
+
+ if (currentSequence == null || rnd.nextDouble() < 0.5) {
+ // add a new record
+ boolean isEvent = rnd.nextDouble() < 0.05;
+ BufferOrEvent evt;
+ if (isEvent) {
+ evt = generateRandomEvent(rnd, numChannels);
+ events.add(evt);
+ } else {
+ evt = generateRandomBuffer(bufferRnd.nextInt(PAGE_SIZE) + 1, bufferRnd.nextInt(numChannels));
+ }
+ bufferBlocker.add(evt);
+ generated++;
+ } else {
+ // consume a record
+ BufferOrEvent next = currentSequence.sequence.getNext();
+ assertNotNull(next);
+ if (next.isEvent()) {
+ BufferOrEvent expected = currentSequence.events.get(currentNumEvents++);
+ assertEquals(expected.getEvent(), next.getEvent());
+ assertEquals(expected.getChannelIndex(), next.getChannelIndex());
+ } else {
+ Random validationRnd = currentSequence.bufferRnd;
+ validateBuffer(next, validationRnd.nextInt(PAGE_SIZE) + 1, validationRnd.nextInt(currentSequence.numChannels));
+ }
+
+ currentNumRecordAndEvents++;
+ if (currentNumRecordAndEvents == currentSequence.numBuffersAndEvents) {
+ // done with the sequence
+ currentSequence.sequence.cleanup();
+ sequencesConsumed++;
+
+ // validate we had all events
+ assertEquals(currentSequence.events.size(), currentNumEvents);
+
+ // reset
+ currentSequence = pendingSequences.pollFirst();
+ if (currentSequence != null) {
+ currentSequence.sequence.open();
+ }
+
+ currentNumRecordAndEvents = 0;
+ currentNumEvents = 0;
+ }
+ }
+ }
+
+ // done generating a sequence. queue it for consumption
+ bufferRnd.setSeed(bufferSeed);
+ BufferOrEventSequence seq = bufferBlocker.rollOverReusingResources();
+
+ SequenceToConsume stc = new SequenceToConsume(bufferRnd, events, seq, numEventsAndBuffers, numChannels);
+
+ if (currentSequence == null) {
+ currentSequence = stc;
+ stc.sequence.open();
+ } else {
+ pendingSequences.addLast(stc);
+ }
+ }
+ }
+
+ // consume all the remainder
+ while (currentSequence != null) {
+ // consume a record
+ BufferOrEvent next = currentSequence.sequence.getNext();
+ assertNotNull(next);
+ if (next.isEvent()) {
+ BufferOrEvent expected = currentSequence.events.get(currentNumEvents++);
+ assertEquals(expected.getEvent(), next.getEvent());
+ assertEquals(expected.getChannelIndex(), next.getChannelIndex());
+ } else {
+ Random validationRnd = currentSequence.bufferRnd;
+ validateBuffer(next, validationRnd.nextInt(PAGE_SIZE) + 1, validationRnd.nextInt(currentSequence.numChannels));
+ }
+
+ currentNumRecordAndEvents++;
+ if (currentNumRecordAndEvents == currentSequence.numBuffersAndEvents) {
+ // done with the sequence
+ currentSequence.sequence.cleanup();
+ sequencesConsumed++;
+
+ // validate we had all events
+ assertEquals(currentSequence.events.size(), currentNumEvents);
+
+ // reset
+ currentSequence = pendingSequences.pollFirst();
+ if (currentSequence != null) {
+ currentSequence.sequence.open();
+ }
+
+ currentNumRecordAndEvents = 0;
+ currentNumEvents = 0;
+ }
+ }
+
+ assertEquals(sequences, sequencesConsumed);
+ }
+
+ // ------------------------------------------------------------------------
+ // Utils
+ // ------------------------------------------------------------------------
+
+ private static BufferOrEvent generateRandomEvent(Random rnd, int numChannels) {
+ long magicNumber = rnd.nextLong();
+ byte[] data = new byte[rnd.nextInt(1000)];
+ rnd.nextBytes(data);
+ TestEvent evt = new TestEvent(magicNumber, data);
+
+ int channelIndex = rnd.nextInt(numChannels);
+
+ return new BufferOrEvent(evt, channelIndex);
+ }
+
+ public static BufferOrEvent generateRandomBuffer(int size, int channelIndex) {
+ MemorySegment seg = MemorySegmentFactory.allocateUnpooledSegment(PAGE_SIZE);
+ for (int i = 0; i < size; i++) {
+ seg.put(i, (byte) i);
+ }
+
+ Buffer buf = new NetworkBuffer(seg, FreeingBufferRecycler.INSTANCE);
+ buf.setSize(size);
+ return new BufferOrEvent(buf, channelIndex);
+ }
+
+ private static void validateBuffer(BufferOrEvent boe, int expectedSize, int expectedChannelIndex) {
+ assertEquals("wrong channel index", expectedChannelIndex, boe.getChannelIndex());
+ assertTrue("is not buffer", boe.isBuffer());
+
+ Buffer buf = boe.getBuffer();
+ assertEquals("wrong buffer size", expectedSize, buf.getSize());
+
+ MemorySegment seg = buf.getMemorySegment();
+ for (int i = 0; i < expectedSize; i++) {
+ byte expected = (byte) i;
+ if (expected != seg.get(i)) {
+ fail(String.format(
+ "wrong buffer contents at position %s : expected=%d , found=%d", i, expected, seg.get(i)));
+ }
+ }
+ }
+
+ /**
+ * Wrappers the buffered sequence and related elements for consuming and validation.
+ */
+ private static class SequenceToConsume {
+
+ final BufferOrEventSequence sequence;
+ final ArrayList<BufferOrEvent> events;
+ final Random bufferRnd;
+ final int numBuffersAndEvents;
+ final int numChannels;
+
+ private SequenceToConsume(
+ Random bufferRnd,
+ ArrayList<BufferOrEvent> events,
+ BufferOrEventSequence sequence,
+ int numBuffersAndEvents,
+ int numChannels) {
+ this.bufferRnd = bufferRnd;
+ this.events = events;
+ this.sequence = sequence;
+ this.numBuffersAndEvents = numBuffersAndEvents;
+ this.numChannels = numChannels;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/3126bf52/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BufferSpillerTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BufferSpillerTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BufferSpillerTest.java
index ee58052..b70ba24 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BufferSpillerTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BufferSpillerTest.java
@@ -18,13 +18,8 @@
package org.apache.flink.streaming.runtime.io;
-import org.apache.flink.core.memory.MemorySegment;
-import org.apache.flink.core.memory.MemorySegmentFactory;
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.buffer.Buffer;
-import org.apache.flink.runtime.io.network.buffer.FreeingBufferRecycler;
-import org.apache.flink.runtime.io.network.buffer.NetworkBuffer;
import org.apache.flink.runtime.io.network.partition.consumer.BufferOrEvent;
import org.junit.After;
@@ -32,35 +27,23 @@ import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.io.File;
-import java.util.ArrayDeque;
-import java.util.ArrayList;
-import java.util.Random;
+import java.io.IOException;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
/**
* Tests for {@link BufferSpiller}.
*/
-public class BufferSpillerTest {
-
- private static final Logger LOG = LoggerFactory.getLogger(BufferSpillerTest.class);
-
- private static final int PAGE_SIZE = 4096;
+public class BufferSpillerTest extends BufferBlockerTestBase {
private static IOManager ioManager;
private BufferSpiller spiller;
-
// ------------------------------------------------------------------------
// Setup / Cleanup
// ------------------------------------------------------------------------
@@ -76,26 +59,14 @@ public class BufferSpillerTest {
}
@Before
- public void createSpiller() {
- try {
- spiller = new BufferSpiller(ioManager, PAGE_SIZE);
- }
- catch (Exception e) {
- e.printStackTrace();
- fail("Cannot create BufferSpiller: " + e.getMessage());
- }
+ public void createSpiller() throws IOException {
+ spiller = new BufferSpiller(ioManager, PAGE_SIZE);
}
@After
- public void cleanupSpiller() {
+ public void cleanupSpiller() throws IOException {
if (spiller != null) {
- try {
- spiller.close();
- }
- catch (Exception e) {
- e.printStackTrace();
- fail("Cannot properly close the BufferSpiller: " + e.getMessage());
- }
+ spiller.close();
assertFalse(spiller.getCurrentChannel().isOpen());
assertFalse(spiller.getCurrentSpillFile().exists());
@@ -104,239 +75,9 @@ public class BufferSpillerTest {
checkNoTempFilesRemain();
}
- // ------------------------------------------------------------------------
- // Tests
- // ------------------------------------------------------------------------
-
- @Test
- public void testRollOverEmptySequences() {
- try {
- assertNull(spiller.rollOver());
- assertNull(spiller.rollOver());
- assertNull(spiller.rollOver());
- }
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
- }
-
- @Test
- public void testSpillAndRollOverSimple() {
- try {
- final Random rnd = new Random();
- final Random bufferRnd = new Random();
-
- final int maxNumEventsAndBuffers = 3000;
- final int maxNumChannels = 1656;
-
- // do multiple spilling / rolling over rounds
- for (int round = 0; round < 5; round++) {
-
- final long bufferSeed = rnd.nextLong();
- bufferRnd.setSeed(bufferSeed);
-
- final int numEventsAndBuffers = rnd.nextInt(maxNumEventsAndBuffers) + 1;
- final int numChannels = rnd.nextInt(maxNumChannels) + 1;
-
- final ArrayList<BufferOrEvent> events = new ArrayList<BufferOrEvent>(128);
-
- // generate sequence
- for (int i = 0; i < numEventsAndBuffers; i++) {
- boolean isEvent = rnd.nextDouble() < 0.05d;
- if (isEvent) {
- BufferOrEvent evt = generateRandomEvent(rnd, numChannels);
- events.add(evt);
- spiller.add(evt);
- }
- else {
- BufferOrEvent evt = generateRandomBuffer(bufferRnd.nextInt(PAGE_SIZE) + 1, bufferRnd.nextInt(numChannels));
- spiller.add(evt);
- }
- }
-
- // reset and create reader
- bufferRnd.setSeed(bufferSeed);
-
- BufferSpiller.SpilledBufferOrEventSequence seq = spiller.rollOver();
- seq.open();
-
- // read and validate the sequence
-
- int numEvent = 0;
- for (int i = 0; i < numEventsAndBuffers; i++) {
- BufferOrEvent next = seq.getNext();
- assertNotNull(next);
- if (next.isEvent()) {
- BufferOrEvent expected = events.get(numEvent++);
- assertEquals(expected.getEvent(), next.getEvent());
- assertEquals(expected.getChannelIndex(), next.getChannelIndex());
- }
- else {
- validateBuffer(next, bufferRnd.nextInt(PAGE_SIZE) + 1, bufferRnd.nextInt(numChannels));
- }
- }
-
- // no further data
- assertNull(seq.getNext());
-
- // all events need to be consumed
- assertEquals(events.size(), numEvent);
-
- seq.cleanup();
- }
- }
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
- }
-
- @Test
- public void testSpillWhileReading() {
- LOG.info("Starting SpillWhileReading test");
-
- try {
- final int sequences = 10;
-
- final Random rnd = new Random();
-
- final int maxNumEventsAndBuffers = 30000;
- final int maxNumChannels = 1656;
-
- int sequencesConsumed = 0;
-
- ArrayDeque<SequenceToConsume> pendingSequences = new ArrayDeque<SequenceToConsume>();
- SequenceToConsume currentSequence = null;
- int currentNumEvents = 0;
- int currentNumRecordAndEvents = 0;
-
- // do multiple spilling / rolling over rounds
- for (int round = 0; round < 2 * sequences; round++) {
-
- if (round % 2 == 1) {
- // make this an empty sequence
- assertNull(spiller.rollOver());
- }
- else {
- // proper spilled sequence
- final long bufferSeed = rnd.nextLong();
- final Random bufferRnd = new Random(bufferSeed);
-
- final int numEventsAndBuffers = rnd.nextInt(maxNumEventsAndBuffers) + 1;
- final int numChannels = rnd.nextInt(maxNumChannels) + 1;
-
- final ArrayList<BufferOrEvent> events = new ArrayList<BufferOrEvent>(128);
-
- int generated = 0;
- while (generated < numEventsAndBuffers) {
-
- if (currentSequence == null || rnd.nextDouble() < 0.5) {
- // add a new record
- boolean isEvent = rnd.nextDouble() < 0.05;
- if (isEvent) {
- BufferOrEvent evt = generateRandomEvent(rnd, numChannels);
- events.add(evt);
- spiller.add(evt);
- }
- else {
- BufferOrEvent evt = generateRandomBuffer(bufferRnd.nextInt(PAGE_SIZE) + 1, bufferRnd.nextInt(numChannels));
- spiller.add(evt);
- }
- generated++;
- }
- else {
- // consume a record
- BufferOrEvent next = currentSequence.sequence.getNext();
- assertNotNull(next);
- if (next.isEvent()) {
- BufferOrEvent expected = currentSequence.events.get(currentNumEvents++);
- assertEquals(expected.getEvent(), next.getEvent());
- assertEquals(expected.getChannelIndex(), next.getChannelIndex());
- }
- else {
- Random validationRnd = currentSequence.bufferRnd;
- validateBuffer(next, validationRnd.nextInt(PAGE_SIZE) + 1, validationRnd.nextInt(currentSequence.numChannels));
- }
-
- currentNumRecordAndEvents++;
- if (currentNumRecordAndEvents == currentSequence.numBuffersAndEvents) {
- // done with the sequence
- currentSequence.sequence.cleanup();
- sequencesConsumed++;
-
- // validate we had all events
- assertEquals(currentSequence.events.size(), currentNumEvents);
-
- // reset
- currentSequence = pendingSequences.pollFirst();
- if (currentSequence != null) {
- currentSequence.sequence.open();
- }
-
- currentNumRecordAndEvents = 0;
- currentNumEvents = 0;
- }
- }
- }
-
- // done generating a sequence. queue it for consumption
- bufferRnd.setSeed(bufferSeed);
- BufferSpiller.SpilledBufferOrEventSequence seq = spiller.rollOver();
-
- SequenceToConsume stc = new SequenceToConsume(bufferRnd, events, seq, numEventsAndBuffers, numChannels);
-
- if (currentSequence == null) {
- currentSequence = stc;
- stc.sequence.open();
- }
- else {
- pendingSequences.addLast(stc);
- }
- }
- }
-
- // consume all the remainder
- while (currentSequence != null) {
- // consume a record
- BufferOrEvent next = currentSequence.sequence.getNext();
- assertNotNull(next);
- if (next.isEvent()) {
- BufferOrEvent expected = currentSequence.events.get(currentNumEvents++);
- assertEquals(expected.getEvent(), next.getEvent());
- assertEquals(expected.getChannelIndex(), next.getChannelIndex());
- }
- else {
- Random validationRnd = currentSequence.bufferRnd;
- validateBuffer(next, validationRnd.nextInt(PAGE_SIZE) + 1, validationRnd.nextInt(currentSequence.numChannels));
- }
-
- currentNumRecordAndEvents++;
- if (currentNumRecordAndEvents == currentSequence.numBuffersAndEvents) {
- // done with the sequence
- currentSequence.sequence.cleanup();
- sequencesConsumed++;
-
- // validate we had all events
- assertEquals(currentSequence.events.size(), currentNumEvents);
-
- // reset
- currentSequence = pendingSequences.pollFirst();
- if (currentSequence != null) {
- currentSequence.sequence.open();
- }
-
- currentNumRecordAndEvents = 0;
- currentNumEvents = 0;
- }
- }
-
- assertEquals(sequences, sequencesConsumed);
- }
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
+ @Override
+ public BufferBlocker createBufferBlocker() {
+ return spiller;
}
/**
@@ -351,50 +92,7 @@ public class BufferSpillerTest {
assertEquals(
"Changed the header format, but did not adjust the HEADER_SIZE field",
BufferSpiller.HEADER_SIZE + size,
- spiller.getBytesWritten());
- }
-
- // ------------------------------------------------------------------------
- // Utils
- // ------------------------------------------------------------------------
-
- private static BufferOrEvent generateRandomEvent(Random rnd, int numChannels) {
- long magicNumber = rnd.nextLong();
- byte[] data = new byte[rnd.nextInt(1000)];
- rnd.nextBytes(data);
- TestEvent evt = new TestEvent(magicNumber, data);
-
- int channelIndex = rnd.nextInt(numChannels);
-
- return new BufferOrEvent(evt, channelIndex);
- }
-
- private static BufferOrEvent generateRandomBuffer(int size, int channelIndex) {
- MemorySegment seg = MemorySegmentFactory.allocateUnpooledSegment(PAGE_SIZE);
- for (int i = 0; i < size; i++) {
- seg.put(i, (byte) i);
- }
-
- Buffer buf = new NetworkBuffer(seg, FreeingBufferRecycler.INSTANCE);
- buf.setSize(size);
- return new BufferOrEvent(buf, channelIndex);
- }
-
- private static void validateBuffer(BufferOrEvent boe, int expectedSize, int expectedChannelIndex) {
- assertEquals("wrong channel index", expectedChannelIndex, boe.getChannelIndex());
- assertTrue("is not buffer", boe.isBuffer());
-
- Buffer buf = boe.getBuffer();
- assertEquals("wrong buffer size", expectedSize, buf.getSize());
-
- MemorySegment seg = buf.getMemorySegment();
- for (int i = 0; i < expectedSize; i++) {
- byte expected = (byte) i;
- if (expected != seg.get(i)) {
- fail(String.format(
- "wrong buffer contents at position %s : expected=%d , found=%d", i, expected, seg.get(i)));
- }
- }
+ spiller.getBytesBlocked());
}
private static void checkNoTempFilesRemain() {
@@ -407,23 +105,4 @@ public class BufferSpillerTest {
}
}
}
-
- private static class SequenceToConsume {
-
- final BufferSpiller.SpilledBufferOrEventSequence sequence;
- final ArrayList<BufferOrEvent> events;
- final Random bufferRnd;
- final int numBuffersAndEvents;
- final int numChannels;
-
- private SequenceToConsume(Random bufferRnd, ArrayList<BufferOrEvent> events,
- BufferSpiller.SpilledBufferOrEventSequence sequence,
- int numBuffersAndEvents, int numChannels) {
- this.bufferRnd = bufferRnd;
- this.events = events;
- this.sequence = sequence;
- this.numBuffersAndEvents = numBuffersAndEvents;
- this.numChannels = numChannels;
- }
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/3126bf52/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/CachedBufferBlockerTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/CachedBufferBlockerTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/CachedBufferBlockerTest.java
new file mode 100644
index 0000000..e7bf128
--- /dev/null
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/CachedBufferBlockerTest.java
@@ -0,0 +1,53 @@
+/*
+ * 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.junit.After;
+import org.junit.Before;
+
+import java.io.IOException;
+
+/**
+ * Tests for {@link CachedBufferBlocker}.
+ */
+public class CachedBufferBlockerTest extends BufferBlockerTestBase {
+
+ private CachedBufferBlocker bufferBlocker;
+
+ // ------------------------------------------------------------------------
+ // Setup / Cleanup
+ // ------------------------------------------------------------------------
+
+ @Before
+ public void createBlocker() {
+ bufferBlocker = new CachedBufferBlocker(PAGE_SIZE);
+ }
+
+ @After
+ public void cleanupBlocker() throws IOException {
+ if (bufferBlocker != null) {
+ bufferBlocker.close();
+ }
+ }
+
+ @Override
+ public BufferBlocker createBufferBlocker() {
+ return bufferBlocker;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/3126bf52/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/CreditBasedBarrierBufferTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/CreditBasedBarrierBufferTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/CreditBasedBarrierBufferTest.java
new file mode 100644
index 0000000..a5f9469
--- /dev/null
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/CreditBasedBarrierBufferTest.java
@@ -0,0 +1,49 @@
+/*
+ * 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.runtime.io.network.partition.consumer.BufferOrEvent;
+import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
+
+import java.io.IOException;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for the behaviors of the {@link BarrierBuffer} with {@link CachedBufferBlocker}.
+ */
+public class CreditBasedBarrierBufferTest extends BarrierBufferTestBase {
+
+ @Override
+ public BarrierBuffer createBarrierHandler(InputGate gate) throws IOException {
+ return new BarrierBuffer(gate, new CachedBufferBlocker(PAGE_SIZE));
+ }
+
+ @Override
+ public void validateAlignmentBuffered(long actualBytesBuffered, BufferOrEvent... sequence) {
+ long expectedBuffered = 0;
+ for (BufferOrEvent boe : sequence) {
+ if (boe.isBuffer()) {
+ expectedBuffered += PAGE_SIZE;
+ }
+ }
+
+ assertEquals("Wrong alignment buffered bytes", actualBytesBuffered, expectedBuffered);
+ }
+}