[FLINK-6769] Replace usage of deprecated FileSystem#create(Path, bool…

…ean)

This closes apache#4116.

flink-core/src/test/java/org/apache/flink/core/fs/InitOutputPathTest.java

@@ -191,7 +191,7 @@ private static class FileCreator extends CheckedThread {
  @Override
  public void go() throws Exception {
  fs.initOutPathLocalFS(path.getParent(), WriteMode.OVERWRITE, true);
- try (FSDataOutputStream out = fs.create(path, true)) {
+ try (FSDataOutputStream out = fs.create(path, WriteMode.OVERWRITE)) {
  out.write(11);
  }
  }

flink-core/src/test/java/org/apache/flink/core/fs/SafetyNetCloseableRegistryTest.java

@@ -84,7 +84,7 @@ public void go() throws Exception {
 
  Path tmp = new Path(tmpFolder.newFolder().toURI().toString(), "test_file");
 
- try (FSDataOutputStream stream = fs1.create(tmp, false)) {
+ try (FSDataOutputStream stream = fs1.create(tmp, FileSystem.WriteMode.NO_OVERWRITE)) {
  CheckedThread t2 = new CheckedThread() {
  @Override
  public void go() {
@@ -285,4 +285,4 @@ public void close() throws IOException {
  }
  }
  }
-}
+}

flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/distcp/DistCp.java

@@ -129,7 +129,7 @@ public void flatMap(FileCopyTask task, Collector<Object> out) throws Exception {
  FSDataOutputStream outputStream = null;
  FSDataInputStream inputStream = null;
  try {
- outputStream = targetFs.create(outPath, true);
+ outputStream = targetFs.create(outPath, FileSystem.WriteMode.OVERWRITE);
  inputStream = task.getPath().getFileSystem().open(task.getPath());
  int bytes = IOUtils.copy(inputStream, outputStream);
  bytesCounter.add(bytes);

flink-fs-tests/src/test/java/org/apache/flink/hdfstests/HDFSTest.java

@@ -199,7 +199,7 @@ public void testDeletePathIfEmpty() throws IOException {
  byte[] data = "HDFSTest#testDeletePathIfEmpty".getBytes(ConfigConstants.DEFAULT_CHARSET);
 
  for (Path file: Arrays.asList(singleFile, directoryFile)) {
- org.apache.flink.core.fs.FSDataOutputStream outputStream = fs.create(file, true);
+ org.apache.flink.core.fs.FSDataOutputStream outputStream = fs.create(file, FileSystem.WriteMode.OVERWRITE);
  outputStream.write(data);
  outputStream.close();
  }

flink-runtime/src/main/java/org/apache/flink/runtime/blob/FileSystemBlobStore.java

@@ -74,7 +74,7 @@ public void put(File localFile, JobID jobId, String key) throws IOException {
  }
 
  private void put(File fromFile, String toBlobPath) throws IOException {
- try (OutputStream os = fileSystem.create(new Path(toBlobPath), true)) {
+ try (OutputStream os = fileSystem.create(new Path(toBlobPath), FileSystem.WriteMode.OVERWRITE)) {
  LOG.debug("Copying from {} to {}.", fromFile, toBlobPath);
  Files.copy(fromFile, os);
  }

flink-runtime/src/main/java/org/apache/flink/runtime/filecache/FileCache.java

@@ -252,7 +252,7 @@ public static void copy(Path sourcePath, Path targetPath, boolean executable) th
  copy(content.getPath(), new Path(localPath), executable);
  }
  } else {
- try (FSDataOutputStream lfsOutput = tFS.create(targetPath, false); FSDataInputStream fsInput = sFS.open(sourcePath)) {
+ try (FSDataOutputStream lfsOutput = tFS.create(targetPath, FileSystem.WriteMode.NO_OVERWRITE); FSDataInputStream fsInput = sFS.open(sourcePath)) {
  IOUtils.copyBytes(fsInput, lfsOutput);
  //noinspection ResultOfMethodCallIgnored
  new File(targetPath.toString()).setExecutable(executable);

flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java

@@ -359,7 +359,7 @@ private void createStream() throws IOException {
  for (int attempt = 0; attempt < 10; attempt++) {
  try {
  statePath = createStatePath();
- outStream = fs.create(statePath, false);
+ outStream = fs.create(statePath, FileSystem.WriteMode.NO_OVERWRITE);
  break;
  }
  catch (Exception e) {

flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/filesystem/FileSystemStateStorageHelper.java

@@ -62,7 +62,7 @@ public RetrievableStateHandle<T> store(T state) throws Exception {
  for (int attempt = 0; attempt < 10; attempt++) {
  Path filePath = getNewFilePath();
 
- try (FSDataOutputStream outStream = fs.create(filePath, false)) {
+ try (FSDataOutputStream outStream = fs.create(filePath, FileSystem.WriteMode.NO_OVERWRITE)) {
  InstantiationUtil.serializeObject(outStream, state);
  return new RetrievableStreamStateHandle<T>(filePath, outStream.getPos());
  }

flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/MigrationV0ToV1Test.java

@@ -109,7 +109,7 @@ public void testSavepointMigrationV0ToV1() throws Exception {
  }
 
  try {
- fdos = fs.create(path, false);
+ fdos = fs.create(path, FileSystem.WriteMode.NO_OVERWRITE);
  break;
  } catch (Exception e) {
  latestException = e;

flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointStoreTest.java

@@ -104,7 +104,7 @@ public void testStoreLoadDispose() throws Exception {
  public void testUnexpectedSavepoint() throws Exception {
  // Random file
  Path filePath = new Path(tmp.getRoot().getPath(), UUID.randomUUID().toString());
- FSDataOutputStream fdos = FileSystem.get(filePath.toUri()).create(filePath, false);
+ FSDataOutputStream fdos = FileSystem.get(filePath.toUri()).create(filePath, FileSystem.WriteMode.NO_OVERWRITE);
  DataOutputStream dos = new DataOutputStream(fdos);
  for (int i = 0; i < 10; i++) {
  dos.writeLong(ThreadLocalRandom.current().nextLong());

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

@@ -130,7 +130,7 @@ public void testCleanupWhenClosingStream() throws IOException {
 
  final ArgumentCaptor<Path> pathCaptor = ArgumentCaptor.forClass(Path.class);
 
- when(fs.create(pathCaptor.capture(), anyBoolean())).thenReturn(outputStream);
+ when(fs.create(pathCaptor.capture(), any(FileSystem.WriteMode.class))).thenReturn(outputStream);
 
  CheckpointStreamFactory.CheckpointStateOutputStream stream = new FsCheckpointStreamFactory.FsCheckpointStateOutputStream(
  TEMP_DIR_PATH,
@@ -141,7 +141,7 @@ public void testCleanupWhenClosingStream() throws IOException {
  // this should create the underlying file stream
  stream.write(new byte[]{1,2,3,4,5});
 
- verify(fs).create(any(Path.class), anyBoolean());
+ verify(fs).create(any(Path.class), any(FileSystem.WriteMode.class));
 
  stream.close();
 
@@ -158,7 +158,7 @@ public void testCleanupWhenFailingCloseAndGetHandle() throws IOException {
 
  final ArgumentCaptor<Path> pathCaptor = ArgumentCaptor.forClass(Path.class);
 
- when(fs.create(pathCaptor.capture(), anyBoolean())).thenReturn(outputStream);
+ when(fs.create(pathCaptor.capture(), any(FileSystem.WriteMode.class))).thenReturn(outputStream);
  doThrow(new IOException("Test IOException.")).when(outputStream).close();
 
  CheckpointStreamFactory.CheckpointStateOutputStream stream = new FsCheckpointStreamFactory.FsCheckpointStateOutputStream(
@@ -170,7 +170,7 @@ public void testCleanupWhenFailingCloseAndGetHandle() throws IOException {
  // this should create the underlying file stream
  stream.write(new byte[]{1,2,3,4,5});
 
- verify(fs).create(any(Path.class), anyBoolean());
+ verify(fs).create(any(Path.class), any(FileSystem.WriteMode.class));
 
  try {
  stream.closeAndGetHandle();