[FLINK-11704][tests] Improve AbstractCheckpointStateOutputStreamTestBase by...

[FLINK-11704][tests] Improve AbstractCheckpointStateOutputStreamTestBase by using JUnit's Parameterized

This closes #7784.

flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/AbstractCheckpointStateOutputStreamTestBase.java → flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/CheckpointStateOutputStreamTest.java

@@ -28,14 +28,19 @@ import org.apache.flink.core.testutils.CheckedThread;
 import org.apache.flink.core.testutils.OneShotLatch;
 import org.apache.flink.util.TestLogger;
 import org.apache.flink.util.function.FunctionWithException;
+
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
 
 import java.io.EOFException;
 import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.Arrays;
+import java.util.Collection;
 import java.util.Random;
 import java.util.UUID;
 
@@ -51,11 +56,24 @@ import static org.mockito.Mockito.verify;
 /**
  * Abstract base class for tests against checkpointing streams.
  */
-public abstract class AbstractCheckpointStateOutputStreamTestBase extends TestLogger {
+@RunWith(Parameterized.class)
+public class CheckpointStateOutputStreamTest extends TestLogger {
 
 	@Rule
 	public final TemporaryFolder tmp = new TemporaryFolder();
 
+	public enum CheckpointStateOutputStreamType {
+		FileBasedState, FsCheckpointMetaData
+	}
+
+	@Parameterized.Parameters
+	public static Collection<CheckpointStateOutputStreamType> getCheckpointStateOutputStreamType() {
+		return Arrays.asList(CheckpointStateOutputStreamType.values());
+	}
+
+	@Parameterized.Parameter
+	public CheckpointStateOutputStreamType stateOutputStreamType;
+
 	// ------------------------------------------------------------------------
 	//  Tests
 	// ------------------------------------------------------------------------
@@ -220,15 +238,34 @@ public abstract class AbstractCheckpointStateOutputStreamTestBase extends TestLo
 	/**
 	 * Creates a new test stream instance.
 	 */
-	protected abstract FSDataOutputStream createTestStream(
+	private FSDataOutputStream createTestStream(
 		FileSystem fs,
 		Path dir,
-		String fileName) throws IOException;
+		String fileName) throws IOException {
+		switch (stateOutputStreamType) {
+			case FileBasedState:
+				return new FileBasedStateOutputStream(fs, new Path(dir, fileName));
+			case FsCheckpointMetaData:
+				Path fullPath = new Path(dir, fileName);
+				return new FsCheckpointMetadataOutputStream(fs, fullPath, dir);
+			default:
+				throw new IllegalStateException("Unsupported checkpoint stream output type.");
+		}
+	}
 
 	/**
 	 * Closes the stream successfully and returns a FileStateHandle to the result.
 	 */
-	protected abstract FileStateHandle closeAndGetResult(FSDataOutputStream stream) throws IOException;
+	private FileStateHandle closeAndGetResult(FSDataOutputStream stream) throws IOException {
+		switch (stateOutputStreamType) {
+			case FileBasedState:
+				return ((FileBasedStateOutputStream) stream).closeAndGetHandle();
+			case FsCheckpointMetaData:
+				return ((FsCheckpointMetadataOutputStream) stream).closeAndFinalizeCheckpoint().getMetadataHandle();
+			default:
+				throw new IllegalStateException("Unsupported checkpoint stream output type.");
+		}
+	}
 
 	// ------------------------------------------------------------------------
 	//  utilities

flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FileBasedStateOutputStreamTest.java

-/*
- * 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.runtime.state.filesystem;
-
-import org.apache.flink.core.fs.FSDataOutputStream;
-import org.apache.flink.core.fs.FileSystem;
-import org.apache.flink.core.fs.Path;
-
-import java.io.IOException;
-
-/**
- * Tests for the {@link FileBasedStateOutputStream}.
- */
-public class FileBasedStateOutputStreamTest extends AbstractCheckpointStateOutputStreamTestBase {
-
-	@Override
-	protected FSDataOutputStream createTestStream(FileSystem fs, Path dir, String fileName) throws IOException {
-		return new FileBasedStateOutputStream(fs, new Path(dir, fileName));
-	}
-
-	@Override
-	protected FileStateHandle closeAndGetResult(FSDataOutputStream stream) throws IOException {
-		return ((FileBasedStateOutputStream) stream).closeAndGetHandle();
-	}
-}

flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FsCheckpointMetadataOutputStreamTest.java

-/*
- * 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.runtime.state.filesystem;
-
-import org.apache.flink.core.fs.FSDataOutputStream;
-import org.apache.flink.core.fs.FileSystem;
-import org.apache.flink.core.fs.Path;
-
-import java.io.IOException;
-
-/**
- * Tests for the {@link FsCheckpointMetadataOutputStream}.
- */
-public class FsCheckpointMetadataOutputStreamTest extends AbstractCheckpointStateOutputStreamTestBase {
-
-	@Override
-	protected FSDataOutputStream createTestStream(FileSystem fs, Path dir, String fileName) throws IOException {
-		Path fullPath = new Path(dir, fileName);
-		return new FsCheckpointMetadataOutputStream(fs, fullPath, dir);
-	}
-
-	@Override
-	protected FileStateHandle closeAndGetResult(FSDataOutputStream stream) throws IOException {
-		return ((FsCheckpointMetadataOutputStream) stream).closeAndFinalizeCheckpoint().getMetadataHandle();
-	}
-}