[flink] 04/08: [hotfix][test] Do not hide original exception in BarrierTrackerTest

[pn...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit ce729d757e7272471fd75f3174badfbb2ab839b8
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Thu May 16 15:39:54 2019 +0200

    [hotfix][test] Do not hide original exception in BarrierTrackerTest
---
 .../streaming/runtime/io/BarrierTrackerTest.java   | 446 ++++++++++-----------
 1 file changed, 202 insertions(+), 244 deletions(-)

diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BarrierTrackerTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BarrierTrackerTest.java
index 2ba9d57..c292ddd 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BarrierTrackerTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/BarrierTrackerTest.java
@@ -52,210 +52,174 @@ public class BarrierTrackerTest {
 	private static final int PAGE_SIZE = 512;
 
 	@Test
-	public void testSingleChannelNoBarriers() {
-		try {
-			BufferOrEvent[] sequence = { createBuffer(0), createBuffer(0), createBuffer(0) };
+	public void testSingleChannelNoBarriers() throws Exception {
+		BufferOrEvent[] sequence = { createBuffer(0), createBuffer(0), createBuffer(0) };
 
-			MockInputGate gate = new MockInputGate(PAGE_SIZE, 1, Arrays.asList(sequence));
-			BarrierTracker tracker = new BarrierTracker(gate);
-
-			for (BufferOrEvent boe : sequence) {
-				assertEquals(boe, tracker.getNextNonBlocked());
-			}
+		MockInputGate gate = new MockInputGate(PAGE_SIZE, 1, Arrays.asList(sequence));
+		BarrierTracker tracker = new BarrierTracker(gate);
 
-			assertNull(tracker.getNextNonBlocked());
-			assertNull(tracker.getNextNonBlocked());
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getMessage());
+		for (BufferOrEvent boe : sequence) {
+			assertEquals(boe, tracker.getNextNonBlocked());
 		}
+
+		assertNull(tracker.getNextNonBlocked());
+		assertNull(tracker.getNextNonBlocked());
 	}
 
 	@Test
-	public void testMultiChannelNoBarriers() {
-		try {
-			BufferOrEvent[] sequence = { createBuffer(2), createBuffer(2), createBuffer(0),
-					createBuffer(1), createBuffer(0), createBuffer(3),
-					createBuffer(1), createBuffer(1), createBuffer(2)
-			};
-
-			MockInputGate gate = new MockInputGate(PAGE_SIZE, 4, Arrays.asList(sequence));
-			BarrierTracker tracker = new BarrierTracker(gate);
+	public void testMultiChannelNoBarriers() throws Exception {
+		BufferOrEvent[] sequence = { createBuffer(2), createBuffer(2), createBuffer(0),
+				createBuffer(1), createBuffer(0), createBuffer(3),
+				createBuffer(1), createBuffer(1), createBuffer(2)
+		};
 
-			for (BufferOrEvent boe : sequence) {
-				assertEquals(boe, tracker.getNextNonBlocked());
-			}
+		MockInputGate gate = new MockInputGate(PAGE_SIZE, 4, Arrays.asList(sequence));
+		BarrierTracker tracker = new BarrierTracker(gate);
 
-			assertNull(tracker.getNextNonBlocked());
-			assertNull(tracker.getNextNonBlocked());
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getMessage());
+		for (BufferOrEvent boe : sequence) {
+			assertEquals(boe, tracker.getNextNonBlocked());
 		}
+
+		assertNull(tracker.getNextNonBlocked());
+		assertNull(tracker.getNextNonBlocked());
 	}
 
 	@Test
-	public void testSingleChannelWithBarriers() {
-		try {
-			BufferOrEvent[] sequence = {
-					createBuffer(0), createBuffer(0), createBuffer(0),
-					createBarrier(1, 0),
-					createBuffer(0), createBuffer(0), createBuffer(0), createBuffer(0),
-					createBarrier(2, 0), createBarrier(3, 0),
-					createBuffer(0), createBuffer(0),
-					createBarrier(4, 0), createBarrier(5, 0), createBarrier(6, 0),
-					createBuffer(0)
-			};
-
-			MockInputGate gate = new MockInputGate(PAGE_SIZE, 1, Arrays.asList(sequence));
-			BarrierTracker tracker = new BarrierTracker(gate);
-
-			CheckpointSequenceValidator validator =
-					new CheckpointSequenceValidator(1, 2, 3, 4, 5, 6);
-			tracker.registerCheckpointEventHandler(validator);
-
-			for (BufferOrEvent boe : sequence) {
-				if (boe.isBuffer() || boe.getEvent().getClass() != CheckpointBarrier.class) {
-					assertEquals(boe, tracker.getNextNonBlocked());
-				}
-			}
+	public void testSingleChannelWithBarriers() throws Exception {
+		BufferOrEvent[] sequence = {
+				createBuffer(0), createBuffer(0), createBuffer(0),
+				createBarrier(1, 0),
+				createBuffer(0), createBuffer(0), createBuffer(0), createBuffer(0),
+				createBarrier(2, 0), createBarrier(3, 0),
+				createBuffer(0), createBuffer(0),
+				createBarrier(4, 0), createBarrier(5, 0), createBarrier(6, 0),
+				createBuffer(0)
+		};
 
-			assertNull(tracker.getNextNonBlocked());
-			assertNull(tracker.getNextNonBlocked());
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getMessage());
+		MockInputGate gate = new MockInputGate(PAGE_SIZE, 1, Arrays.asList(sequence));
+		BarrierTracker tracker = new BarrierTracker(gate);
+
+		CheckpointSequenceValidator validator =
+				new CheckpointSequenceValidator(1, 2, 3, 4, 5, 6);
+		tracker.registerCheckpointEventHandler(validator);
+
+		for (BufferOrEvent boe : sequence) {
+			if (boe.isBuffer() || boe.getEvent().getClass() != CheckpointBarrier.class) {
+				assertEquals(boe, tracker.getNextNonBlocked());
+			}
 		}
+
+		assertNull(tracker.getNextNonBlocked());
+		assertNull(tracker.getNextNonBlocked());
 	}
 
 	@Test
-	public void testSingleChannelWithSkippedBarriers() {
-		try {
-			BufferOrEvent[] sequence = {
-					createBuffer(0),
-					createBarrier(1, 0),
-					createBuffer(0), createBuffer(0),
-					createBarrier(3, 0), createBuffer(0),
-					createBarrier(4, 0), createBarrier(6, 0), createBuffer(0),
-					createBarrier(7, 0), createBuffer(0), createBarrier(10, 0),
-					createBuffer(0)
-			};
-
-			MockInputGate gate = new MockInputGate(PAGE_SIZE, 1, Arrays.asList(sequence));
-			BarrierTracker tracker = new BarrierTracker(gate);
-
-			CheckpointSequenceValidator validator =
-					new CheckpointSequenceValidator(1, 3, 4, 6, 7, 10);
-			tracker.registerCheckpointEventHandler(validator);
-
-			for (BufferOrEvent boe : sequence) {
-				if (boe.isBuffer() || boe.getEvent().getClass() != CheckpointBarrier.class) {
-					assertEquals(boe, tracker.getNextNonBlocked());
-				}
-			}
+	public void testSingleChannelWithSkippedBarriers() throws Exception {
+		BufferOrEvent[] sequence = {
+				createBuffer(0),
+				createBarrier(1, 0),
+				createBuffer(0), createBuffer(0),
+				createBarrier(3, 0), createBuffer(0),
+				createBarrier(4, 0), createBarrier(6, 0), createBuffer(0),
+				createBarrier(7, 0), createBuffer(0), createBarrier(10, 0),
+				createBuffer(0)
+		};
 
-			assertNull(tracker.getNextNonBlocked());
-			assertNull(tracker.getNextNonBlocked());
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getMessage());
+		MockInputGate gate = new MockInputGate(PAGE_SIZE, 1, Arrays.asList(sequence));
+		BarrierTracker tracker = new BarrierTracker(gate);
+
+		CheckpointSequenceValidator validator =
+				new CheckpointSequenceValidator(1, 3, 4, 6, 7, 10);
+		tracker.registerCheckpointEventHandler(validator);
+
+		for (BufferOrEvent boe : sequence) {
+			if (boe.isBuffer() || boe.getEvent().getClass() != CheckpointBarrier.class) {
+				assertEquals(boe, tracker.getNextNonBlocked());
+			}
 		}
+
+		assertNull(tracker.getNextNonBlocked());
+		assertNull(tracker.getNextNonBlocked());
 	}
 
 	@Test
-	public void testMultiChannelWithBarriers() {
-		try {
-			BufferOrEvent[] sequence = {
-					createBuffer(0), createBuffer(2), createBuffer(0),
-					createBarrier(1, 1), createBarrier(1, 2),
-					createBuffer(2), createBuffer(1),
-					createBarrier(1, 0),
-
-					createBuffer(0), createBuffer(0), createBuffer(1), createBuffer(1), createBuffer(2),
-					createBarrier(2, 0), createBarrier(2, 1), createBarrier(2, 2),
-
-					createBuffer(2), createBuffer(2),
-					createBarrier(3, 2),
-					createBuffer(2), createBuffer(2),
-					createBarrier(3, 0), createBarrier(3, 1),
-
-					createBarrier(4, 1), createBarrier(4, 2), createBarrier(4, 0),
-
-					createBuffer(0)
-			};
-
-			MockInputGate gate = new MockInputGate(PAGE_SIZE, 3, Arrays.asList(sequence));
-			BarrierTracker tracker = new BarrierTracker(gate);
-
-			CheckpointSequenceValidator validator =
-					new CheckpointSequenceValidator(1, 2, 3, 4);
-			tracker.registerCheckpointEventHandler(validator);
-
-			for (BufferOrEvent boe : sequence) {
-				if (boe.isBuffer() || boe.getEvent().getClass() != CheckpointBarrier.class) {
-					assertEquals(boe, tracker.getNextNonBlocked());
-				}
-			}
+	public void testMultiChannelWithBarriers() throws Exception {
+		BufferOrEvent[] sequence = {
+				createBuffer(0), createBuffer(2), createBuffer(0),
+				createBarrier(1, 1), createBarrier(1, 2),
+				createBuffer(2), createBuffer(1),
+				createBarrier(1, 0),
 
-			assertNull(tracker.getNextNonBlocked());
-			assertNull(tracker.getNextNonBlocked());
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getMessage());
+				createBuffer(0), createBuffer(0), createBuffer(1), createBuffer(1), createBuffer(2),
+				createBarrier(2, 0), createBarrier(2, 1), createBarrier(2, 2),
+
+				createBuffer(2), createBuffer(2),
+				createBarrier(3, 2),
+				createBuffer(2), createBuffer(2),
+				createBarrier(3, 0), createBarrier(3, 1),
+
+				createBarrier(4, 1), createBarrier(4, 2), createBarrier(4, 0),
+
+				createBuffer(0)
+		};
+
+		MockInputGate gate = new MockInputGate(PAGE_SIZE, 3, Arrays.asList(sequence));
+		BarrierTracker tracker = new BarrierTracker(gate);
+
+		CheckpointSequenceValidator validator =
+				new CheckpointSequenceValidator(1, 2, 3, 4);
+		tracker.registerCheckpointEventHandler(validator);
+
+		for (BufferOrEvent boe : sequence) {
+			if (boe.isBuffer() || boe.getEvent().getClass() != CheckpointBarrier.class) {
+				assertEquals(boe, tracker.getNextNonBlocked());
+			}
 		}
+
+		assertNull(tracker.getNextNonBlocked());
+		assertNull(tracker.getNextNonBlocked());
 	}
 
 	@Test
-	public void testMultiChannelSkippingCheckpoints() {
-		try {
-			BufferOrEvent[] sequence = {
-					createBuffer(0), createBuffer(2), createBuffer(0),
-					createBarrier(1, 1), createBarrier(1, 2),
-					createBuffer(2), createBuffer(1),
-					createBarrier(1, 0),
-
-					createBuffer(0), createBuffer(0), createBuffer(1), createBuffer(1), createBuffer(2),
-					createBarrier(2, 0), createBarrier(2, 1), createBarrier(2, 2),
-
-					createBuffer(2), createBuffer(2),
-					createBarrier(3, 2),
-					createBuffer(2), createBuffer(2),
-
-					// jump to checkpoint 4
-					createBarrier(4, 0),
-					createBuffer(0), createBuffer(1), createBuffer(2),
-					createBarrier(4, 1),
-					createBuffer(1),
-					createBarrier(4, 2),
-
-					createBuffer(0)
-			};
-
-			MockInputGate gate = new MockInputGate(PAGE_SIZE, 3, Arrays.asList(sequence));
-			BarrierTracker tracker = new BarrierTracker(gate);
-
-			CheckpointSequenceValidator validator =
-					new CheckpointSequenceValidator(1, 2, 4);
-			tracker.registerCheckpointEventHandler(validator);
-
-			for (BufferOrEvent boe : sequence) {
-				if (boe.isBuffer() || boe.getEvent().getClass() != CheckpointBarrier.class) {
-					assertEquals(boe, tracker.getNextNonBlocked());
-				}
-			}
+	public void testMultiChannelSkippingCheckpoints() throws Exception {
+		BufferOrEvent[] sequence = {
+				createBuffer(0), createBuffer(2), createBuffer(0),
+				createBarrier(1, 1), createBarrier(1, 2),
+				createBuffer(2), createBuffer(1),
+				createBarrier(1, 0),
 
-			assertNull(tracker.getNextNonBlocked());
-			assertNull(tracker.getNextNonBlocked());
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getMessage());
+				createBuffer(0), createBuffer(0), createBuffer(1), createBuffer(1), createBuffer(2),
+				createBarrier(2, 0), createBarrier(2, 1), createBarrier(2, 2),
+
+				createBuffer(2), createBuffer(2),
+				createBarrier(3, 2),
+				createBuffer(2), createBuffer(2),
+
+				// jump to checkpoint 4
+				createBarrier(4, 0),
+				createBuffer(0), createBuffer(1), createBuffer(2),
+				createBarrier(4, 1),
+				createBuffer(1),
+				createBarrier(4, 2),
+
+				createBuffer(0)
+		};
+
+		MockInputGate gate = new MockInputGate(PAGE_SIZE, 3, Arrays.asList(sequence));
+		BarrierTracker tracker = new BarrierTracker(gate);
+
+		CheckpointSequenceValidator validator =
+				new CheckpointSequenceValidator(1, 2, 4);
+		tracker.registerCheckpointEventHandler(validator);
+
+		for (BufferOrEvent boe : sequence) {
+			if (boe.isBuffer() || boe.getEvent().getClass() != CheckpointBarrier.class) {
+				assertEquals(boe, tracker.getNextNonBlocked());
+			}
 		}
+
+		assertNull(tracker.getNextNonBlocked());
+		assertNull(tracker.getNextNonBlocked());
 	}
 
 	/**
@@ -269,77 +233,71 @@ public class BarrierTrackerTest {
 	 * checkpoint all together.
 	 */
 	@Test
-	public void testCompleteCheckpointsOnLateBarriers() {
-		try {
-			BufferOrEvent[] sequence = {
-					// checkpoint 2
-					createBuffer(1), createBuffer(1), createBuffer(0), createBuffer(2),
-					createBarrier(2, 1), createBarrier(2, 0), createBarrier(2, 2),
-
-					// incomplete checkpoint 3
-					createBuffer(1), createBuffer(0),
-					createBarrier(3, 1), createBarrier(3, 2),
-
-					// some barriers from checkpoint 4
-					createBuffer(1), createBuffer(0),
-					createBarrier(4, 2), createBarrier(4, 1),
-					createBuffer(1), createBuffer(2),
-
-					// last barrier from checkpoint 3
-					createBarrier(3, 0),
-
-					// complete checkpoint 4
-					createBuffer(0), createBarrier(4, 0),
-
-					// regular checkpoint 5
-					createBuffer(1), createBuffer(2), createBarrier(5, 1),
-					createBuffer(0), createBarrier(5, 0),
-					createBuffer(1), createBarrier(5, 2),
-
-					// checkpoint 6 (incomplete),
-					createBuffer(1), createBarrier(6, 1),
-					createBuffer(0), createBarrier(6, 0),
-
-					// checkpoint 7, with early barriers for checkpoints 8 and 9
-					createBuffer(1), createBarrier(7, 1),
-					createBuffer(0), createBarrier(7, 2),
-					createBuffer(2), createBarrier(8, 2),
-					createBuffer(0), createBarrier(8, 1),
-					createBuffer(1), createBarrier(9, 1),
-
-					// complete checkpoint 7, first barriers from checkpoint 10
-					createBarrier(7, 0),
-					createBuffer(0), createBarrier(9, 2),
-					createBuffer(2), createBarrier(10, 2),
-
-					// complete checkpoint 8 and 9
-					createBarrier(8, 0),
-					createBuffer(1), createBuffer(2), createBarrier(9, 0),
-
-					// trailing data
-					createBuffer(1), createBuffer(0), createBuffer(2)
-			};
-
-			MockInputGate gate = new MockInputGate(PAGE_SIZE, 3, Arrays.asList(sequence));
-			BarrierTracker tracker = new BarrierTracker(gate);
-
-			CheckpointSequenceValidator validator =
-					new CheckpointSequenceValidator(2, 3, 4, 5, 7, 8, 9);
-			tracker.registerCheckpointEventHandler(validator);
-
-			for (BufferOrEvent boe : sequence) {
-				if (boe.isBuffer() || boe.getEvent().getClass() != CheckpointBarrier.class) {
-					assertEquals(boe, tracker.getNextNonBlocked());
-				}
-			}
+	public void testCompleteCheckpointsOnLateBarriers() throws Exception {
+		BufferOrEvent[] sequence = {
+				// checkpoint 2
+				createBuffer(1), createBuffer(1), createBuffer(0), createBuffer(2),
+				createBarrier(2, 1), createBarrier(2, 0), createBarrier(2, 2),
+
+				// incomplete checkpoint 3
+				createBuffer(1), createBuffer(0),
+				createBarrier(3, 1), createBarrier(3, 2),
+
+				// some barriers from checkpoint 4
+				createBuffer(1), createBuffer(0),
+				createBarrier(4, 2), createBarrier(4, 1),
+				createBuffer(1), createBuffer(2),
+
+				// last barrier from checkpoint 3
+				createBarrier(3, 0),
+
+				// complete checkpoint 4
+				createBuffer(0), createBarrier(4, 0),
+
+				// regular checkpoint 5
+				createBuffer(1), createBuffer(2), createBarrier(5, 1),
+				createBuffer(0), createBarrier(5, 0),
+				createBuffer(1), createBarrier(5, 2),
+
+				// checkpoint 6 (incomplete),
+				createBuffer(1), createBarrier(6, 1),
+				createBuffer(0), createBarrier(6, 0),
+
+				// checkpoint 7, with early barriers for checkpoints 8 and 9
+				createBuffer(1), createBarrier(7, 1),
+				createBuffer(0), createBarrier(7, 2),
+				createBuffer(2), createBarrier(8, 2),
+				createBuffer(0), createBarrier(8, 1),
+				createBuffer(1), createBarrier(9, 1),
+
+				// complete checkpoint 7, first barriers from checkpoint 10
+				createBarrier(7, 0),
+				createBuffer(0), createBarrier(9, 2),
+				createBuffer(2), createBarrier(10, 2),
+
+				// complete checkpoint 8 and 9
+				createBarrier(8, 0),
+				createBuffer(1), createBuffer(2), createBarrier(9, 0),
+
+				// trailing data
+				createBuffer(1), createBuffer(0), createBuffer(2)
+		};
 
-			assertNull(tracker.getNextNonBlocked());
-			assertNull(tracker.getNextNonBlocked());
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getMessage());
+		MockInputGate gate = new MockInputGate(PAGE_SIZE, 3, Arrays.asList(sequence));
+		BarrierTracker tracker = new BarrierTracker(gate);
+
+		CheckpointSequenceValidator validator =
+				new CheckpointSequenceValidator(2, 3, 4, 5, 7, 8, 9);
+		tracker.registerCheckpointEventHandler(validator);
+
+		for (BufferOrEvent boe : sequence) {
+			if (boe.isBuffer() || boe.getEvent().getClass() != CheckpointBarrier.class) {
+				assertEquals(boe, tracker.getNextNonBlocked());
+			}
 		}
+
+		assertNull(tracker.getNextNonBlocked());
+		assertNull(tracker.getNextNonBlocked());
 	}
 
 	@Test