pnowojski commented on a change in pull request #8495: [FLINK-12556][e2e] Extend some end-to-end tests to run with custom (input) File System implementation URL: https://github.com/apache/flink/pull/8495#discussion_r286446648
########## File path: flink-end-to-end-tests/flink-plugins-test/src/main/java/org/apache/flink/fs/dummy/DummyFs.java ########## @@ -0,0 +1,166 @@ +/* + * 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.fs.dummy; + +import org.apache.flink.core.fs.BlockLocation; +import org.apache.flink.core.fs.FSDataInputStream; +import org.apache.flink.core.fs.FSDataOutputStream; +import org.apache.flink.core.fs.FileStatus; +import org.apache.flink.core.fs.FileSystem; +import org.apache.flink.core.fs.FileSystemKind; +import org.apache.flink.core.fs.Path; +import org.apache.flink.core.fs.local.LocalBlockLocation; + +import javax.annotation.Nullable; + +import java.io.File; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.net.URI; +import java.nio.charset.Charset; +import java.util.HashMap; +import java.util.Map; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +class DummyFs extends FileSystem { + + static final URI FS_URI = URI.create("dummy:///"); + + private static final String HOSTNAME = "localhost"; + + private final URI workingDir; + + private final URI homeDir; + + private final Map<String, byte[]> contents; + + DummyFs(Map<String, String> contents) { + this.workingDir = new File(System.getProperty("user.dir")).toURI(); + this.homeDir = new File(System.getProperty("user.home")).toURI(); + this.contents = convertToByteArrayMap(contents); + } + + // ------------------------------------------------------------------------ + + @Override + public URI getUri() { + return FS_URI; + } + + @Override + public Path getWorkingDirectory() { + return new Path(workingDir); + } + + @Override + public Path getHomeDirectory() { + return new Path(homeDir); + } + + @Override + public boolean exists(Path f) throws IOException { + return getDataByPath(f) != null; + } + + @Override + public FileStatus[] listStatus(final Path f) throws IOException { + byte[] data = getDataByPath(f); + if (data == null) { + return null; + } + return new FileStatus[] { new DummyFsFileStatus(f, data.length) }; + } + + @Override + public BlockLocation[] getFileBlockLocations(FileStatus file, long start, long len) throws IOException { + return new BlockLocation[] { + new LocalBlockLocation(HOSTNAME, file.getLen()) + }; + } + + @Override + public FileStatus getFileStatus(Path f) throws IOException { + byte[] data = getDataByPath(f); + if (data == null) { + throw new FileNotFoundException("File " + f + " does not exist or the user running " + + "Flink ('" + System.getProperty("user.name") + "') has insufficient permissions to access it."); + } + return new DummyFsFileStatus(f, data.length); + } + + @Override + public FSDataInputStream open(final Path f, final int bufferSize) throws IOException { + return open(f); + } + + @Override + public FSDataInputStream open(final Path f) throws IOException { + return DummyFsInputStream.create(getDataByPath(f)); + } + + @Override + public boolean delete(final Path path, final boolean recursive) throws IOException { + checkNotNull(path, "path is null"); + throw new UnsupportedOperationException("Test FS doesn't support delete operation"); + } + + @Override + public boolean mkdirs(final Path path) throws IOException { + checkNotNull(path, "path is null"); + throw new UnsupportedOperationException("Test FS doesn't support mkdirs operation"); + } + + @Override + public FSDataOutputStream create(final Path path, final WriteMode overwrite) throws IOException { + checkNotNull(path, "path"); + throw new UnsupportedOperationException("Test FS doesn't support create operation"); + } + + @Override + public boolean rename(final Path src, final Path dst) throws IOException { + throw new UnsupportedOperationException("Test FS doesn't support rename operation"); + } + + @Override + public boolean isDistributedFS() { + return false; Review comment: shouldn't this be `return true`? All of the processes will see the same dummy files, right? ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
