[GitHub] pnowojski closed pull request #6538: [hotfix][streaming] Fix and simplify PrintSinkFunctionTest

[GitBox <gi...@apache.org>]

pnowojski closed pull request #6538: [hotfix][streaming] Fix and simplify PrintSinkFunctionTest
URL: https://github.com/apache/flink/pull/6538
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java
index c9b52415a3e..dbe76302ab0 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java
@@ -18,7 +18,6 @@
 
 package org.apache.flink.streaming.connectors.kafka;
 
-import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.state.BroadcastState;
 import org.apache.flink.api.common.state.KeyedStateStore;
 import org.apache.flink.api.common.state.ListState;
@@ -32,16 +31,12 @@
 import org.apache.flink.metrics.MetricGroup;
 import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
 import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
-import org.apache.flink.runtime.jobgraph.OperatorID;
-import org.apache.flink.runtime.memory.MemoryManager;
-import org.apache.flink.runtime.operators.testutils.MockEnvironmentBuilder;
 import org.apache.flink.runtime.state.FunctionInitializationContext;
 import org.apache.flink.runtime.state.StateSnapshotContextSynchronousImpl;
 import org.apache.flink.streaming.api.TimeCharacteristic;
 import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
 import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
 import org.apache.flink.streaming.api.functions.source.SourceFunction;
-import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
 import org.apache.flink.streaming.api.operators.StreamSource;
 import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
 import org.apache.flink.streaming.connectors.kafka.config.OffsetCommitMode;
@@ -54,6 +49,7 @@
 import org.apache.flink.streaming.connectors.kafka.testutils.TestSourceContext;
 import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
 import org.apache.flink.streaming.util.AbstractStreamOperatorTestHarness;
+import org.apache.flink.streaming.util.MockStreamingRuntimeContext;
 import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
 import org.apache.flink.util.Preconditions;
 import org.apache.flink.util.SerializedValue;
@@ -743,7 +739,7 @@ public void addAll(List<T> values) throws Exception {
 			int totalNumSubtasks) throws Exception {
 
 		// run setup procedure in operator life cycle
-		consumer.setRuntimeContext(new MockRuntimeContext(isCheckpointingEnabled, totalNumSubtasks, subtaskIndex));
+		consumer.setRuntimeContext(new MockStreamingRuntimeContext(isCheckpointingEnabled, totalNumSubtasks, subtaskIndex));
 		consumer.initializeState(new MockFunctionInitializationContext(isRestored, new MockOperatorStateStore(restoredListState)));
 		consumer.open(new Configuration());
 	}
@@ -820,68 +816,6 @@ private int getCommitCount() {
 		}
 	}
 
-	private static class MockRuntimeContext extends StreamingRuntimeContext {
-
-		private final boolean isCheckpointingEnabled;
-
-		private final int numParallelSubtasks;
-		private final int subtaskIndex;
-
-		private MockRuntimeContext(
-				boolean isCheckpointingEnabled,
-				int numParallelSubtasks,
-				int subtaskIndex) {
-
-			super(
-				new MockStreamOperator(),
-				new MockEnvironmentBuilder()
-					.setTaskName("mockTask")
-					.setMemorySize(4 * MemoryManager.DEFAULT_PAGE_SIZE)
-					.build(),
-				Collections.emptyMap());
-
-			this.isCheckpointingEnabled = isCheckpointingEnabled;
-			this.numParallelSubtasks = numParallelSubtasks;
-			this.subtaskIndex = subtaskIndex;
-		}
-
-		@Override
-		public MetricGroup getMetricGroup() {
-			return new UnregisteredMetricsGroup();
-		}
-
-		@Override
-		public boolean isCheckpointingEnabled() {
-			return isCheckpointingEnabled;
-		}
-
-		@Override
-		public int getIndexOfThisSubtask() {
-			return subtaskIndex;
-		}
-
-		@Override
-		public int getNumberOfParallelSubtasks() {
-			return numParallelSubtasks;
-		}
-
-		// ------------------------------------------------------------------------
-
-		private static class MockStreamOperator extends AbstractStreamOperator<Integer> {
-			private static final long serialVersionUID = -1153976702711944427L;
-
-			@Override
-			public ExecutionConfig getExecutionConfig() {
-				return new ExecutionConfig();
-			}
-
-			@Override
-			public OperatorID getOperatorID() {
-				return new OperatorID();
-			}
-		}
-	}
-
 	private static class MockOperatorStateStore implements OperatorStateStore {
 
 		private final ListState<?> mockRestoredUnionListState;
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/PrintSinkFunctionTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/PrintSinkFunctionTest.java
index 8c303d16282..4a163387b9e 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/PrintSinkFunctionTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/PrintSinkFunctionTest.java
@@ -20,12 +20,11 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;
 import org.apache.flink.streaming.api.functions.sink.SinkContextUtil;
-import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
+import org.apache.flink.streaming.util.MockStreamingRuntimeContext;
 
 import org.junit.After;
-import org.junit.Assert;
+import org.junit.Before;
 import org.junit.Test;
-import org.mockito.Mockito;
 
 import java.io.ByteArrayOutputStream;
 import java.io.PrintStream;
@@ -33,93 +32,75 @@
 import static org.junit.Assert.assertEquals;
 
 /**
- * Tests for the {@link org.apache.flink.streaming.api.functions.sink.PrintSinkFunction}.
+ * Tests for the {@link PrintSinkFunction}.
  */
 public class PrintSinkFunctionTest {
 
-	public PrintStream printStreamOriginal = System.out;
-	private String line = System.lineSeparator();
+	private final PrintStream originalSystemOut = System.out;
+	private final PrintStream originalSystemErr = System.err;
 
-	@Test
-	public void testPrintSinkStdOut() throws Exception {
-		ByteArrayOutputStream baos = new ByteArrayOutputStream();
-		PrintStream stream = new PrintStream(baos);
-		System.setOut(stream);
+	private final ByteArrayOutputStream arrayOutputStream = new ByteArrayOutputStream();
+	private final ByteArrayOutputStream arrayErrorStream = new ByteArrayOutputStream();
 
-		final StreamingRuntimeContext ctx = Mockito.mock(StreamingRuntimeContext.class);
+	private final String line = System.lineSeparator();
 
-		PrintSinkFunction<String> printSink = new PrintSinkFunction<>();
-		printSink.setRuntimeContext(ctx);
-		try {
-			printSink.open(new Configuration());
-		} catch (Exception e) {
-			Assert.fail();
+	@Before
+	public void setUp() {
+		System.setOut(new PrintStream(arrayOutputStream));
+		System.setErr(new PrintStream(arrayErrorStream));
+	}
+
+	@After
+	public void tearDown() {
+		if (System.out != originalSystemOut) {
+			System.out.close();
 		}
+		if (System.err != originalSystemErr) {
+			System.err.close();
+		}
+		System.setOut(originalSystemOut);
+		System.setErr(originalSystemErr);
+	}
+
+	@Test
+	public void testPrintSinkStdOut() throws Exception {
+		PrintSinkFunction<String> printSink = new PrintSinkFunction<>();
+		printSink.setRuntimeContext(new MockStreamingRuntimeContext(false, 1, 0));
 		printSink.setTargetToStandardOut();
+		printSink.open(new Configuration());
+
 		printSink.invoke("hello world!", SinkContextUtil.forTimestamp(0));
 
 		assertEquals("Print to System.out", printSink.toString());
-		assertEquals("hello world!" + line, baos.toString());
-
+		assertEquals("hello world!" + line, arrayOutputStream.toString());
 		printSink.close();
-		stream.close();
 	}
 
 	@Test
 	public void testPrintSinkStdErr() throws Exception {
-		ByteArrayOutputStream baos = new ByteArrayOutputStream();
-		PrintStream stream = new PrintStream(baos);
-		System.setOut(stream);
-
-		final StreamingRuntimeContext ctx = Mockito.mock(StreamingRuntimeContext.class);
-
 		PrintSinkFunction<String> printSink = new PrintSinkFunction<>();
-		printSink.setRuntimeContext(ctx);
-		try {
-			printSink.open(new Configuration());
-		} catch (Exception e) {
-			Assert.fail();
-		}
+		printSink.setRuntimeContext(new MockStreamingRuntimeContext(false, 1, 0));
 		printSink.setTargetToStandardErr();
+		printSink.open(new Configuration());
+
 		printSink.invoke("hello world!", SinkContextUtil.forTimestamp(0));
 
 		assertEquals("Print to System.err", printSink.toString());
-		assertEquals("hello world!" + line, baos.toString());
-
+		assertEquals("hello world!" + line, arrayErrorStream.toString());
 		printSink.close();
-		stream.close();
 	}
 
 	@Test
 	public void testPrintSinkWithPrefix() throws Exception {
-		ByteArrayOutputStream baos = new ByteArrayOutputStream();
-		PrintStream stream = new PrintStream(baos);
-		System.setOut(stream);
-
-		final StreamingRuntimeContext ctx = Mockito.mock(StreamingRuntimeContext.class);
-		Mockito.when(ctx.getNumberOfParallelSubtasks()).thenReturn(2);
-		Mockito.when(ctx.getIndexOfThisSubtask()).thenReturn(1);
-
 		PrintSinkFunction<String> printSink = new PrintSinkFunction<>();
-		printSink.setRuntimeContext(ctx);
-		try {
-			printSink.open(new Configuration());
-		} catch (Exception e) {
-			Assert.fail();
-		}
-		printSink.setTargetToStandardErr();
-		printSink.invoke("hello world!", SinkContextUtil.forTimestamp(0));
+		printSink.setRuntimeContext(new MockStreamingRuntimeContext(false, 2, 1));
+		printSink.setTargetToStandardOut();
+		printSink.open(new Configuration());
 
-		assertEquals("Print to System.err", printSink.toString());
-		assertEquals("2> hello world!" + line, baos.toString());
+		printSink.invoke("hello world!", SinkContextUtil.forTimestamp(0));
 
+		assertEquals("Print to System.out", printSink.toString());
+		assertEquals("2> hello world!" + line, arrayOutputStream.toString());
 		printSink.close();
-		stream.close();
 	}
-
-	@After
-	public void restoreSystemOut() {
-		System.setOut(printStreamOriginal);
-	}
-
 }
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/MockStreamingRuntimeContext.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/MockStreamingRuntimeContext.java
new file mode 100644
index 00000000000..d024c04f5e0
--- /dev/null
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/MockStreamingRuntimeContext.java
@@ -0,0 +1,92 @@
+/*
+ * 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.util;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.operators.testutils.MockEnvironmentBuilder;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
+
+import java.util.Collections;
+
+/**
+ * Mock {@link StreamingRuntimeContext} to use in tests.
+ */
+public class MockStreamingRuntimeContext extends StreamingRuntimeContext {
+
+	private final boolean isCheckpointingEnabled;
+
+	private final int numParallelSubtasks;
+	private final int subtaskIndex;
+
+	public MockStreamingRuntimeContext(
+		boolean isCheckpointingEnabled,
+		int numParallelSubtasks,
+		int subtaskIndex) {
+
+		super(
+			new MockStreamOperator(),
+			new MockEnvironmentBuilder()
+				.setTaskName("mockTask")
+				.setMemorySize(4 * MemoryManager.DEFAULT_PAGE_SIZE)
+				.build(),
+			Collections.emptyMap());
+
+		this.isCheckpointingEnabled = isCheckpointingEnabled;
+		this.numParallelSubtasks = numParallelSubtasks;
+		this.subtaskIndex = subtaskIndex;
+	}
+
+	@Override
+	public MetricGroup getMetricGroup() {
+		return new UnregisteredMetricsGroup();
+	}
+
+	@Override
+	public boolean isCheckpointingEnabled() {
+		return isCheckpointingEnabled;
+	}
+
+	@Override
+	public int getIndexOfThisSubtask() {
+		return subtaskIndex;
+	}
+
+	@Override
+	public int getNumberOfParallelSubtasks() {
+		return numParallelSubtasks;
+	}
+
+	private static class MockStreamOperator extends AbstractStreamOperator<Integer> {
+		private static final long serialVersionUID = -1153976702711944427L;
+
+		@Override
+		public ExecutionConfig getExecutionConfig() {
+			return new ExecutionConfig();
+		}
+
+		@Override
+		public OperatorID getOperatorID() {
+			return new OperatorID();
+		}
+	}
+}


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services