From 4d465307cb7b3101ee90476acd8f34f86c1b989e Mon Sep 17 00:00:00 2001 From: visu83 <60400453+visu83@users.noreply.github.com> Date: Fri, 24 Jul 2020 07:45:47 -0700 Subject: [PATCH 1/7] Create DemoFile.java --- DemoFile.java | 886 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 886 insertions(+) create mode 100644 DemoFile.java diff --git a/DemoFile.java b/DemoFile.java new file mode 100644 index 0000000..dab6664 --- /dev/null +++ b/DemoFile.java @@ -0,0 +1,886 @@ +/* + * 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.hudi.common.fs; + +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.exception.HoodieIOException; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.BlockLocation; +import org.apache.hadoop.fs.ContentSummary; +import org.apache.hadoop.fs.CreateFlag; +import org.apache.hadoop.fs.FSDataInputStream; +import org.apache.hadoop.fs.FSDataOutputStream; +import org.apache.hadoop.fs.FileChecksum; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.FsServerDefaults; +import org.apache.hadoop.fs.FsStatus; +import org.apache.hadoop.fs.LocatedFileStatus; +import org.apache.hadoop.fs.Options; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.PathFilter; +import org.apache.hadoop.fs.RemoteIterator; +import org.apache.hadoop.fs.XAttrSetFlag; +import org.apache.hadoop.fs.permission.AclEntry; +import org.apache.hadoop.fs.permission.AclStatus; +import org.apache.hadoop.fs.permission.FsAction; +import org.apache.hadoop.fs.permission.FsPermission; +import org.apache.hadoop.security.Credentials; +import org.apache.hadoop.security.token.Token; +import org.apache.hadoop.util.Progressable; + +import java.io.IOException; +import java.net.URI; +import java.net.URISyntaxException; +import java.util.EnumSet; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.TimeoutException; + +/** + * HoodieWrapperFileSystem wraps the default file system. It holds state about the open streams in the file system to + * support getting the written size to each of the open streams. + */ +public class HoodieWrapperFileSystem extends FileSystem { + + public static final String HOODIE_SCHEME_PREFIX = "hoodie-"; + + private ConcurrentMap openStreams = new ConcurrentHashMap<>(); + private FileSystem fileSystem; + private URI uri; + private ConsistencyGuard consistencyGuard = new NoOpConsistencyGuard(); + + public HoodieWrapperFileSystem() {} + + public HoodieWrapperFileSystem(FileSystem fileSystem, ConsistencyGuard consistencyGuard) { + this.fileSystem = fileSystem; + this.uri = fileSystem.getUri(); + this.consistencyGuard = consistencyGuard; + } + + public static Path convertToHoodiePath(Path file, Configuration conf) { + try { + String scheme = FSUtils.getFs(file.toString(), conf).getScheme(); + return convertPathWithScheme(file, getHoodieScheme(scheme)); + } catch (HoodieIOException e) { + throw e; + } + } + + private static Path convertPathWithScheme(Path oldPath, String newScheme) { + URI oldURI = oldPath.toUri(); + URI newURI; + try { + newURI = new URI(newScheme, oldURI.getUserInfo(), oldURI.getHost(), oldURI.getPort(), oldURI.getPath(), + oldURI.getQuery(), oldURI.getFragment()); + return new Path(newURI); + } catch (URISyntaxException e) { + // TODO - Better Exception handling + throw new RuntimeException(e); + } + } + + public static String getHoodieScheme(String scheme) { + String newScheme; + if (StorageSchemes.isSchemeSupported(scheme)) { + newScheme = HOODIE_SCHEME_PREFIX + scheme; + } else { + throw new IllegalArgumentException("BlockAlignedAvroParquetWriter does not support scheme " + scheme); + } + return newScheme; + } + + @Override + public void initialize(URI uri, Configuration conf) { + // Get the default filesystem to decorate + Path path = new Path(uri); + // Remove 'hoodie-' prefix from path + if (path.toString().startsWith(HOODIE_SCHEME_PREFIX)) { + path = new Path(path.toString().replace(HOODIE_SCHEME_PREFIX, "")); + this.uri = path.toUri(); + } else { + this.uri = uri; + } + this.fileSystem = FSUtils.getFs(path.toString(), conf); + // Do not need to explicitly initialize the default filesystem, its done already in the above + // FileSystem.get + // fileSystem.initialize(FileSystem.getDefaultUri(conf), conf); + // fileSystem.setConf(conf); + } + + @Override + public URI getUri() { + return uri; + } + + @Override + public FSDataInputStream open(Path f, int bufferSize) throws IOException { + return fileSystem.open(convertToDefaultPath(f), bufferSize); + } + + @Override + public FSDataOutputStream create(Path f, FsPermission permission, boolean overwrite, int bufferSize, + short replication, long blockSize, Progressable progress) throws IOException { + final Path translatedPath = convertToDefaultPath(f); + return wrapOutputStream(f, + fileSystem.create(translatedPath, permission, overwrite, bufferSize, replication, blockSize, progress)); + } + + private FSDataOutputStream wrapOutputStream(final Path path, FSDataOutputStream fsDataOutputStream) + throws IOException { + if (fsDataOutputStream instanceof SizeAwareFSDataOutputStream) { + return fsDataOutputStream; + } + + SizeAwareFSDataOutputStream os = new SizeAwareFSDataOutputStream(path, fsDataOutputStream, consistencyGuard, + () -> openStreams.remove(path.getName())); + openStreams.put(path.getName(), os); + return os; + } + + @Override + public FSDataOutputStream create(Path f, boolean overwrite) throws IOException { + return wrapOutputStream(f, fileSystem.create(convertToDefaultPath(f), overwrite)); + } + + @Override + public FSDataOutputStream create(Path f) throws IOException { + return wrapOutputStream(f, fileSystem.create(convertToDefaultPath(f))); + } + + @Override + public FSDataOutputStream create(Path f, Progressable progress) throws IOException { + return wrapOutputStream(f, fileSystem.create(convertToDefaultPath(f), progress)); + } + + @Override + public FSDataOutputStream create(Path f, short replication) throws IOException { + return wrapOutputStream(f, fileSystem.create(convertToDefaultPath(f), replication)); + } + + @Override + public FSDataOutputStream create(Path f, short replication, Progressable progress) throws IOException { + return wrapOutputStream(f, fileSystem.create(convertToDefaultPath(f), replication, progress)); + } + + @Override + public FSDataOutputStream create(Path f, boolean overwrite, int bufferSize) throws IOException { + return wrapOutputStream(f, fileSystem.create(convertToDefaultPath(f), overwrite, bufferSize)); + } + + @Override + public FSDataOutputStream create(Path f, boolean overwrite, int bufferSize, Progressable progress) + throws IOException { + return wrapOutputStream(f, fileSystem.create(convertToDefaultPath(f), overwrite, bufferSize, progress)); + } + + @Override + public FSDataOutputStream create(Path f, boolean overwrite, int bufferSize, short replication, long blockSize, + Progressable progress) throws IOException { + return wrapOutputStream(f, + fileSystem.create(convertToDefaultPath(f), overwrite, bufferSize, replication, blockSize, progress)); + } + + @Override + public FSDataOutputStream create(Path f, FsPermission permission, EnumSet flags, int bufferSize, + short replication, long blockSize, Progressable progress) throws IOException { + return wrapOutputStream(f, + fileSystem.create(convertToDefaultPath(f), permission, flags, bufferSize, replication, blockSize, progress)); + } + + @Override + public FSDataOutputStream create(Path f, FsPermission permission, EnumSet flags, int bufferSize, + short replication, long blockSize, Progressable progress, Options.ChecksumOpt checksumOpt) throws IOException { + return wrapOutputStream(f, fileSystem.create(convertToDefaultPath(f), permission, flags, bufferSize, replication, + blockSize, progress, checksumOpt)); + } + + @Override + public FSDataOutputStream create(Path f, boolean overwrite, int bufferSize, short replication, long blockSize) + throws IOException { + return wrapOutputStream(f, + fileSystem.create(convertToDefaultPath(f), overwrite, bufferSize, replication, blockSize)); + } + + @Override + public FSDataOutputStream append(Path f, int bufferSize, Progressable progress) throws IOException { + return wrapOutputStream(f, fileSystem.append(convertToDefaultPath(f), bufferSize, progress)); + } + + @Override + public boolean rename(Path src, Path dst) throws IOException { + try { + consistencyGuard.waitTillFileAppears(convertToDefaultPath(src)); + } catch (TimeoutException e) { + throw new HoodieException("Timed out waiting for " + src + " to appear", e); + } + + boolean success = fileSystem.rename(convertToDefaultPath(src), convertToDefaultPath(dst)); + + if (success) { + try { + consistencyGuard.waitTillFileAppears(convertToDefaultPath(dst)); + } catch (TimeoutException e) { + throw new HoodieException("Timed out waiting for " + dst + " to appear", e); + } + + try { + consistencyGuard.waitTillFileDisappears(convertToDefaultPath(src)); + } catch (TimeoutException e) { + throw new HoodieException("Timed out waiting for " + src + " to disappear", e); + } + } + return success; + } + + @Override + public boolean delete(Path f, boolean recursive) throws IOException { + boolean success = fileSystem.delete(convertToDefaultPath(f), recursive); + + if (success) { + try { + consistencyGuard.waitTillFileDisappears(f); + } catch (TimeoutException e) { + throw new HoodieException("Timed out waiting for " + f + " to disappear", e); + } + } + return success; + } + + @Override + public FileStatus[] listStatus(Path f) throws IOException { + return fileSystem.listStatus(convertToDefaultPath(f)); + } + + @Override + public Path getWorkingDirectory() { + return convertToHoodiePath(fileSystem.getWorkingDirectory()); + } + + @Override + public void setWorkingDirectory(Path newDir) { + fileSystem.setWorkingDirectory(convertToDefaultPath(newDir)); + } + + @Override + public boolean mkdirs(Path f, FsPermission permission) throws IOException { + boolean success = fileSystem.mkdirs(convertToDefaultPath(f), permission); + if (success) { + try { + consistencyGuard.waitTillFileAppears(convertToDefaultPath(f)); + } catch (TimeoutException e) { + throw new HoodieException("Timed out waiting for directory " + f + " to appear", e); + } + } + return success; + } + + @Override + public FileStatus getFileStatus(Path f) throws IOException { + try { + consistencyGuard.waitTillFileAppears(convertToDefaultPath(f)); + } catch (TimeoutException e) { + // pass + } + return fileSystem.getFileStatus(convertToDefaultPath(f)); + } + + @Override + public String getScheme() { + return uri.getScheme(); + } + + @Override + public String getCanonicalServiceName() { + return fileSystem.getCanonicalServiceName(); + } + + @Override + public String getName() { + return fileSystem.getName(); + } + + @Override + public Path makeQualified(Path path) { + return convertToHoodiePath(fileSystem.makeQualified(convertToDefaultPath(path))); + } + + @Override + public Token getDelegationToken(String renewer) throws IOException { + return fileSystem.getDelegationToken(renewer); + } + + @Override + public Token[] addDelegationTokens(String renewer, Credentials credentials) throws IOException { + return fileSystem.addDelegationTokens(renewer, credentials); + } + + @Override + public FileSystem[] getChildFileSystems() { + return fileSystem.getChildFileSystems(); + } + + @Override + public BlockLocation[] getFileBlockLocations(FileStatus file, long start, long len) throws IOException { + return fileSystem.getFileBlockLocations(file, start, len); + } + + @Override + public BlockLocation[] getFileBlockLocations(Path p, long start, long len) throws IOException { + return fileSystem.getFileBlockLocations(convertToDefaultPath(p), start, len); + } + + @Override + public FsServerDefaults getServerDefaults() throws IOException { + return fileSystem.getServerDefaults(); + } + + @Override + public FsServerDefaults getServerDefaults(Path p) throws IOException { + return fileSystem.getServerDefaults(convertToDefaultPath(p)); + } + + @Override + public Path resolvePath(Path p) throws IOException { + return convertToHoodiePath(fileSystem.resolvePath(convertToDefaultPath(p))); + } + + @Override + public FSDataInputStream open(Path f) throws IOException { + return fileSystem.open(convertToDefaultPath(f)); + } + + @Override + public FSDataOutputStream createNonRecursive(Path f, boolean overwrite, int bufferSize, short replication, + long blockSize, Progressable progress) throws IOException { + Path p = convertToDefaultPath(f); + return wrapOutputStream(p, + fileSystem.createNonRecursive(p, overwrite, bufferSize, replication, blockSize, progress)); + } + + @Override + public FSDataOutputStream createNonRecursive(Path f, FsPermission permission, boolean overwrite, int bufferSize, + short replication, long blockSize, Progressable progress) throws IOException { + Path p = convertToDefaultPath(f); + return wrapOutputStream(p, + fileSystem.createNonRecursive(p, permission, overwrite, bufferSize, replication, blockSize, progress)); + } + + @Override + public FSDataOutputStream createNonRecursive(Path f, FsPermission permission, EnumSet flags, + int bufferSize, short replication, long blockSize, Progressable progress) throws IOException { + Path p = convertToDefaultPath(f); + return wrapOutputStream(p, + fileSystem.createNonRecursive(p, permission, flags, bufferSize, replication, blockSize, progress)); + } + + @Override + public boolean createNewFile(Path f) throws IOException { + boolean newFile = fileSystem.createNewFile(convertToDefaultPath(f)); + if (newFile) { + try { + consistencyGuard.waitTillFileAppears(convertToDefaultPath(f)); + } catch (TimeoutException e) { + throw new HoodieException("Timed out waiting for " + f + " to appear", e); + } + } + return newFile; + } + + @Override + public FSDataOutputStream append(Path f) throws IOException { + return wrapOutputStream(f, fileSystem.append(convertToDefaultPath(f))); + } + + @Override + public FSDataOutputStream append(Path f, int bufferSize) throws IOException { + return wrapOutputStream(f, fileSystem.append(convertToDefaultPath(f), bufferSize)); + } + + @Override + public void concat(Path trg, Path[] psrcs) throws IOException { + Path[] psrcsNew = convertDefaults(psrcs); + fileSystem.concat(convertToDefaultPath(trg), psrcsNew); + try { + consistencyGuard.waitTillFileAppears(convertToDefaultPath(trg)); + } catch (TimeoutException e) { + throw new HoodieException("Timed out waiting for " + trg + " to appear", e); + } + } + + @Override + public short getReplication(Path src) throws IOException { + return fileSystem.getReplication(convertToDefaultPath(src)); + } + + @Override + public boolean setReplication(Path src, short replication) throws IOException { + return fileSystem.setReplication(convertToDefaultPath(src), replication); + } + + @Override + public boolean delete(Path f) throws IOException { + return delete(f, true); + } + + @Override + public boolean deleteOnExit(Path f) throws IOException { + return fileSystem.deleteOnExit(convertToDefaultPath(f)); + } + + @Override + public boolean cancelDeleteOnExit(Path f) { + return fileSystem.cancelDeleteOnExit(convertToDefaultPath(f)); + } + + @Override + public boolean exists(Path f) throws IOException { + return fileSystem.exists(convertToDefaultPath(f)); + } + + @Override + public boolean isDirectory(Path f) throws IOException { + return fileSystem.isDirectory(convertToDefaultPath(f)); + } + + @Override + public boolean isFile(Path f) throws IOException { + return fileSystem.isFile(convertToDefaultPath(f)); + } + + @Override + public long getLength(Path f) throws IOException { + return fileSystem.getLength(convertToDefaultPath(f)); + } + + @Override + public ContentSummary getContentSummary(Path f) throws IOException { + return fileSystem.getContentSummary(convertToDefaultPath(f)); + } + + @Override + public RemoteIterator listCorruptFileBlocks(Path path) throws IOException { + return fileSystem.listCorruptFileBlocks(convertToDefaultPath(path)); + } + + @Override + public FileStatus[] listStatus(Path f, PathFilter filter) throws IOException { + return fileSystem.listStatus(convertToDefaultPath(f), filter); + } + + @Override + public FileStatus[] listStatus(Path[] files) throws IOException { + return fileSystem.listStatus(convertDefaults(files)); + } + + @Override + public FileStatus[] listStatus(Path[] files, PathFilter filter) throws IOException { + return fileSystem.listStatus(convertDefaults(files), filter); + } + + @Override + public FileStatus[] globStatus(Path pathPattern) throws IOException { + return fileSystem.globStatus(convertToDefaultPath(pathPattern)); + } + + @Override + public FileStatus[] globStatus(Path pathPattern, PathFilter filter) throws IOException { + return fileSystem.globStatus(convertToDefaultPath(pathPattern), filter); + } + + @Override + public RemoteIterator listLocatedStatus(Path f) throws IOException { + return fileSystem.listLocatedStatus(convertToDefaultPath(f)); + } + + @Override + public RemoteIterator listFiles(Path f, boolean recursive) throws IOException { + return fileSystem.listFiles(convertToDefaultPath(f), recursive); + } + + @Override + public Path getHomeDirectory() { + return convertToHoodiePath(fileSystem.getHomeDirectory()); + } + + @Override + public boolean mkdirs(Path f) throws IOException { + boolean success = fileSystem.mkdirs(convertToDefaultPath(f)); + if (success) { + try { + consistencyGuard.waitTillFileAppears(convertToDefaultPath(f)); + } catch (TimeoutException e) { + throw new HoodieException("Timed out waiting for directory " + f + " to appear", e); + } + } + return success; + } + + @Override + public void copyFromLocalFile(Path src, Path dst) throws IOException { + fileSystem.copyFromLocalFile(convertToLocalPath(src), convertToDefaultPath(dst)); + try { + consistencyGuard.waitTillFileAppears(convertToDefaultPath(dst)); + } catch (TimeoutException e) { + throw new HoodieException("Timed out waiting for destination " + dst + " to appear", e); + } + } + + @Override + public void moveFromLocalFile(Path[] srcs, Path dst) throws IOException { + fileSystem.moveFromLocalFile(convertLocalPaths(srcs), convertToDefaultPath(dst)); + try { + consistencyGuard.waitTillFileAppears(convertToDefaultPath(dst)); + } catch (TimeoutException e) { + throw new HoodieException("Timed out waiting for destination " + dst + " to appear", e); + } + } + + @Override + public void moveFromLocalFile(Path src, Path dst) throws IOException { + fileSystem.moveFromLocalFile(convertToLocalPath(src), convertToDefaultPath(dst)); + try { + consistencyGuard.waitTillFileAppears(convertToDefaultPath(dst)); + } catch (TimeoutException e) { + throw new HoodieException("Timed out waiting for destination " + dst + " to appear", e); + } + } + + @Override + public void copyFromLocalFile(boolean delSrc, Path src, Path dst) throws IOException { + fileSystem.copyFromLocalFile(delSrc, convertToLocalPath(src), convertToDefaultPath(dst)); + try { + consistencyGuard.waitTillFileAppears(convertToDefaultPath(dst)); + } catch (TimeoutException e) { + throw new HoodieException("Timed out waiting for destination " + dst + " to appear", e); + } + } + + @Override + public void copyFromLocalFile(boolean delSrc, boolean overwrite, Path[] srcs, Path dst) throws IOException { + fileSystem.copyFromLocalFile(delSrc, overwrite, convertLocalPaths(srcs), convertToDefaultPath(dst)); + try { + consistencyGuard.waitTillFileAppears(convertToDefaultPath(dst)); + } catch (TimeoutException e) { + throw new HoodieException("Timed out waiting for destination " + dst + " to appear", e); + } + } + + @Override + public void copyFromLocalFile(boolean delSrc, boolean overwrite, Path src, Path dst) throws IOException { + fileSystem.copyFromLocalFile(delSrc, overwrite, convertToLocalPath(src), convertToDefaultPath(dst)); + try { + consistencyGuard.waitTillFileAppears(convertToDefaultPath(dst)); + } catch (TimeoutException e) { + throw new HoodieException("Timed out waiting for destination " + dst + " to appear", e); + } + } + + @Override + public void copyToLocalFile(Path src, Path dst) throws IOException { + fileSystem.copyToLocalFile(convertToDefaultPath(src), convertToLocalPath(dst)); + } + + @Override + public void moveToLocalFile(Path src, Path dst) throws IOException { + fileSystem.moveToLocalFile(convertToDefaultPath(src), convertToLocalPath(dst)); + } + + @Override + public void copyToLocalFile(boolean delSrc, Path src, Path dst) throws IOException { + fileSystem.copyToLocalFile(delSrc, convertToDefaultPath(src), convertToLocalPath(dst)); + } + + @Override + public void copyToLocalFile(boolean delSrc, Path src, Path dst, boolean useRawLocalFileSystem) throws IOException { + fileSystem.copyToLocalFile(delSrc, convertToDefaultPath(src), convertToLocalPath(dst), useRawLocalFileSystem); + } + + @Override + public Path startLocalOutput(Path fsOutputFile, Path tmpLocalFile) throws IOException { + return convertToHoodiePath( + fileSystem.startLocalOutput(convertToDefaultPath(fsOutputFile), convertToDefaultPath(tmpLocalFile))); + } + + @Override + public void completeLocalOutput(Path fsOutputFile, Path tmpLocalFile) throws IOException { + fileSystem.completeLocalOutput(convertToDefaultPath(fsOutputFile), convertToDefaultPath(tmpLocalFile)); + } + + @Override + public void close() throws IOException { + // Don't close the wrapped `fileSystem` object. This will end up closing it for every thread since it + // could be cached across jvm. We don't own that object anyway. + super.close(); + } + + @Override + public long getUsed() throws IOException { + return fileSystem.getUsed(); + } + + @Override + public long getBlockSize(Path f) throws IOException { + return fileSystem.getBlockSize(convertToDefaultPath(f)); + } + + @Override + public long getDefaultBlockSize() { + return fileSystem.getDefaultBlockSize(); + } + + @Override + public long getDefaultBlockSize(Path f) { + return fileSystem.getDefaultBlockSize(convertToDefaultPath(f)); + } + + @Override + public short getDefaultReplication() { + return fileSystem.getDefaultReplication(); + } + + @Override + public short getDefaultReplication(Path path) { + return fileSystem.getDefaultReplication(convertToDefaultPath(path)); + } + + @Override + public void access(Path path, FsAction mode) throws IOException { + fileSystem.access(convertToDefaultPath(path), mode); + } + + @Override + public void createSymlink(Path target, Path link, boolean createParent) throws IOException { + fileSystem.createSymlink(convertToDefaultPath(target), convertToDefaultPath(link), createParent); + } + + @Override + public FileStatus getFileLinkStatus(Path f) throws IOException { + return fileSystem.getFileLinkStatus(convertToDefaultPath(f)); + } + + @Override + public boolean supportsSymlinks() { + return fileSystem.supportsSymlinks(); + } + + @Override + public Path getLinkTarget(Path f) throws IOException { + return convertToHoodiePath(fileSystem.getLinkTarget(convertToDefaultPath(f))); + } + + @Override + public FileChecksum getFileChecksum(Path f) throws IOException { + return fileSystem.getFileChecksum(convertToDefaultPath(f)); + } + + @Override + public FileChecksum getFileChecksum(Path f, long length) throws IOException { + return fileSystem.getFileChecksum(convertToDefaultPath(f), length); + } + + @Override + public void setVerifyChecksum(boolean verifyChecksum) { + fileSystem.setVerifyChecksum(verifyChecksum); + } + + @Override + public void setWriteChecksum(boolean writeChecksum) { + fileSystem.setWriteChecksum(writeChecksum); + } + + @Override + public FsStatus getStatus() throws IOException { + return fileSystem.getStatus(); + } + + @Override + public FsStatus getStatus(Path p) throws IOException { + return fileSystem.getStatus(convertToDefaultPath(p)); + } + + @Override + public void setPermission(Path p, FsPermission permission) throws IOException { + fileSystem.setPermission(convertToDefaultPath(p), permission); + } + + @Override + public void setOwner(Path p, String username, String groupname) throws IOException { + fileSystem.setOwner(convertToDefaultPath(p), username, groupname); + } + + @Override + public void setTimes(Path p, long mtime, long atime) throws IOException { + fileSystem.setTimes(convertToDefaultPath(p), mtime, atime); + } + + @Override + public Path createSnapshot(Path path, String snapshotName) throws IOException { + return convertToHoodiePath(fileSystem.createSnapshot(convertToDefaultPath(path), snapshotName)); + } + + @Override + public void renameSnapshot(Path path, String snapshotOldName, String snapshotNewName) throws IOException { + fileSystem.renameSnapshot(convertToDefaultPath(path), snapshotOldName, snapshotNewName); + } + + @Override + public void deleteSnapshot(Path path, String snapshotName) throws IOException { + fileSystem.deleteSnapshot(convertToDefaultPath(path), snapshotName); + } + + @Override + public void modifyAclEntries(Path path, List aclSpec) throws IOException { + fileSystem.modifyAclEntries(convertToDefaultPath(path), aclSpec); + } + + @Override + public void removeAclEntries(Path path, List aclSpec) throws IOException { + fileSystem.removeAclEntries(convertToDefaultPath(path), aclSpec); + } + + @Override + public void removeDefaultAcl(Path path) throws IOException { + fileSystem.removeDefaultAcl(convertToDefaultPath(path)); + } + + @Override + public void removeAcl(Path path) throws IOException { + fileSystem.removeAcl(convertToDefaultPath(path)); + } + + @Override + public void setAcl(Path path, List aclSpec) throws IOException { + fileSystem.setAcl(convertToDefaultPath(path), aclSpec); + } + + @Override + public AclStatus getAclStatus(Path path) throws IOException { + return fileSystem.getAclStatus(convertToDefaultPath(path)); + } + + @Override + public void setXAttr(Path path, String name, byte[] value) throws IOException { + fileSystem.setXAttr(convertToDefaultPath(path), name, value); + } + + @Override + public void setXAttr(Path path, String name, byte[] value, EnumSet flag) throws IOException { + fileSystem.setXAttr(convertToDefaultPath(path), name, value, flag); + } + + @Override + public byte[] getXAttr(Path path, String name) throws IOException { + return fileSystem.getXAttr(convertToDefaultPath(path), name); + } + + @Override + public Map getXAttrs(Path path) throws IOException { + return fileSystem.getXAttrs(convertToDefaultPath(path)); + } + + @Override + public Map getXAttrs(Path path, List names) throws IOException { + return fileSystem.getXAttrs(convertToDefaultPath(path), names); + } + + @Override + public List listXAttrs(Path path) throws IOException { + return fileSystem.listXAttrs(convertToDefaultPath(path)); + } + + @Override + public void removeXAttr(Path path, String name) throws IOException { + fileSystem.removeXAttr(convertToDefaultPath(path), name); + } + + @Override + public Configuration getConf() { + return fileSystem.getConf(); + } + + @Override + public void setConf(Configuration conf) { + // ignore this. we will set conf on init + } + + @Override + public int hashCode() { + return fileSystem.hashCode(); + } + + @Override + public boolean equals(Object obj) { + return fileSystem.equals(obj); + } + + @Override + public String toString() { + return fileSystem.toString(); + } + + public Path convertToHoodiePath(Path oldPath) { + return convertPathWithScheme(oldPath, getHoodieScheme(getScheme())); + } + + private Path convertToDefaultPath(Path oldPath) { + return convertPathWithScheme(oldPath, getScheme()); + } + + private Path convertToLocalPath(Path oldPath) { + try { + return convertPathWithScheme(oldPath, FileSystem.getLocal(getConf()).getScheme()); + } catch (IOException e) { + throw new HoodieIOException(e.getMessage(), e); + } + } + + private Path[] convertLocalPaths(Path[] psrcs) { + Path[] psrcsNew = new Path[psrcs.length]; + for (int i = 0; i < psrcs.length; i++) { + psrcsNew[i] = convertToLocalPath(psrcs[i]); + } + return psrcsNew; + } + + private Path[] convertDefaults(Path[] psrcs) { + Path[] psrcsNew = new Path[psrcs.length]; + for (int i = 0; i < psrcs.length; i++) { + psrcsNew[i] = convertToDefaultPath(psrcs[i]); + } + return psrcsNew; + } + + public long getBytesWritten(Path file) { + if (openStreams.containsKey(file.getName())) { + return openStreams.get(file.getName()).getBytesWritten(); + } + // When the file is first written, we do not have a track of it + throw new IllegalArgumentException( + file.toString() + " does not have a open stream. Cannot get the bytes written on the stream"); + } + + public FileSystem getFileSystem() { + return fileSystem; + } +} + From 4a0d6febee88f6367d8be13fb9f5107972181c59 Mon Sep 17 00:00:00 2001 From: visu83 <60400453+visu83@users.noreply.github.com> Date: Fri, 24 Jul 2020 07:56:36 -0700 Subject: [PATCH 2/7] Delete DemoFile.java --- DemoFile.java | 886 -------------------------------------------------- 1 file changed, 886 deletions(-) delete mode 100644 DemoFile.java diff --git a/DemoFile.java b/DemoFile.java deleted file mode 100644 index dab6664..0000000 --- a/DemoFile.java +++ /dev/null @@ -1,886 +0,0 @@ -/* - * 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.hudi.common.fs; - -import org.apache.hudi.exception.HoodieException; -import org.apache.hudi.exception.HoodieIOException; - -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.fs.BlockLocation; -import org.apache.hadoop.fs.ContentSummary; -import org.apache.hadoop.fs.CreateFlag; -import org.apache.hadoop.fs.FSDataInputStream; -import org.apache.hadoop.fs.FSDataOutputStream; -import org.apache.hadoop.fs.FileChecksum; -import org.apache.hadoop.fs.FileStatus; -import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.fs.FsServerDefaults; -import org.apache.hadoop.fs.FsStatus; -import org.apache.hadoop.fs.LocatedFileStatus; -import org.apache.hadoop.fs.Options; -import org.apache.hadoop.fs.Path; -import org.apache.hadoop.fs.PathFilter; -import org.apache.hadoop.fs.RemoteIterator; -import org.apache.hadoop.fs.XAttrSetFlag; -import org.apache.hadoop.fs.permission.AclEntry; -import org.apache.hadoop.fs.permission.AclStatus; -import org.apache.hadoop.fs.permission.FsAction; -import org.apache.hadoop.fs.permission.FsPermission; -import org.apache.hadoop.security.Credentials; -import org.apache.hadoop.security.token.Token; -import org.apache.hadoop.util.Progressable; - -import java.io.IOException; -import java.net.URI; -import java.net.URISyntaxException; -import java.util.EnumSet; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.TimeoutException; - -/** - * HoodieWrapperFileSystem wraps the default file system. It holds state about the open streams in the file system to - * support getting the written size to each of the open streams. - */ -public class HoodieWrapperFileSystem extends FileSystem { - - public static final String HOODIE_SCHEME_PREFIX = "hoodie-"; - - private ConcurrentMap openStreams = new ConcurrentHashMap<>(); - private FileSystem fileSystem; - private URI uri; - private ConsistencyGuard consistencyGuard = new NoOpConsistencyGuard(); - - public HoodieWrapperFileSystem() {} - - public HoodieWrapperFileSystem(FileSystem fileSystem, ConsistencyGuard consistencyGuard) { - this.fileSystem = fileSystem; - this.uri = fileSystem.getUri(); - this.consistencyGuard = consistencyGuard; - } - - public static Path convertToHoodiePath(Path file, Configuration conf) { - try { - String scheme = FSUtils.getFs(file.toString(), conf).getScheme(); - return convertPathWithScheme(file, getHoodieScheme(scheme)); - } catch (HoodieIOException e) { - throw e; - } - } - - private static Path convertPathWithScheme(Path oldPath, String newScheme) { - URI oldURI = oldPath.toUri(); - URI newURI; - try { - newURI = new URI(newScheme, oldURI.getUserInfo(), oldURI.getHost(), oldURI.getPort(), oldURI.getPath(), - oldURI.getQuery(), oldURI.getFragment()); - return new Path(newURI); - } catch (URISyntaxException e) { - // TODO - Better Exception handling - throw new RuntimeException(e); - } - } - - public static String getHoodieScheme(String scheme) { - String newScheme; - if (StorageSchemes.isSchemeSupported(scheme)) { - newScheme = HOODIE_SCHEME_PREFIX + scheme; - } else { - throw new IllegalArgumentException("BlockAlignedAvroParquetWriter does not support scheme " + scheme); - } - return newScheme; - } - - @Override - public void initialize(URI uri, Configuration conf) { - // Get the default filesystem to decorate - Path path = new Path(uri); - // Remove 'hoodie-' prefix from path - if (path.toString().startsWith(HOODIE_SCHEME_PREFIX)) { - path = new Path(path.toString().replace(HOODIE_SCHEME_PREFIX, "")); - this.uri = path.toUri(); - } else { - this.uri = uri; - } - this.fileSystem = FSUtils.getFs(path.toString(), conf); - // Do not need to explicitly initialize the default filesystem, its done already in the above - // FileSystem.get - // fileSystem.initialize(FileSystem.getDefaultUri(conf), conf); - // fileSystem.setConf(conf); - } - - @Override - public URI getUri() { - return uri; - } - - @Override - public FSDataInputStream open(Path f, int bufferSize) throws IOException { - return fileSystem.open(convertToDefaultPath(f), bufferSize); - } - - @Override - public FSDataOutputStream create(Path f, FsPermission permission, boolean overwrite, int bufferSize, - short replication, long blockSize, Progressable progress) throws IOException { - final Path translatedPath = convertToDefaultPath(f); - return wrapOutputStream(f, - fileSystem.create(translatedPath, permission, overwrite, bufferSize, replication, blockSize, progress)); - } - - private FSDataOutputStream wrapOutputStream(final Path path, FSDataOutputStream fsDataOutputStream) - throws IOException { - if (fsDataOutputStream instanceof SizeAwareFSDataOutputStream) { - return fsDataOutputStream; - } - - SizeAwareFSDataOutputStream os = new SizeAwareFSDataOutputStream(path, fsDataOutputStream, consistencyGuard, - () -> openStreams.remove(path.getName())); - openStreams.put(path.getName(), os); - return os; - } - - @Override - public FSDataOutputStream create(Path f, boolean overwrite) throws IOException { - return wrapOutputStream(f, fileSystem.create(convertToDefaultPath(f), overwrite)); - } - - @Override - public FSDataOutputStream create(Path f) throws IOException { - return wrapOutputStream(f, fileSystem.create(convertToDefaultPath(f))); - } - - @Override - public FSDataOutputStream create(Path f, Progressable progress) throws IOException { - return wrapOutputStream(f, fileSystem.create(convertToDefaultPath(f), progress)); - } - - @Override - public FSDataOutputStream create(Path f, short replication) throws IOException { - return wrapOutputStream(f, fileSystem.create(convertToDefaultPath(f), replication)); - } - - @Override - public FSDataOutputStream create(Path f, short replication, Progressable progress) throws IOException { - return wrapOutputStream(f, fileSystem.create(convertToDefaultPath(f), replication, progress)); - } - - @Override - public FSDataOutputStream create(Path f, boolean overwrite, int bufferSize) throws IOException { - return wrapOutputStream(f, fileSystem.create(convertToDefaultPath(f), overwrite, bufferSize)); - } - - @Override - public FSDataOutputStream create(Path f, boolean overwrite, int bufferSize, Progressable progress) - throws IOException { - return wrapOutputStream(f, fileSystem.create(convertToDefaultPath(f), overwrite, bufferSize, progress)); - } - - @Override - public FSDataOutputStream create(Path f, boolean overwrite, int bufferSize, short replication, long blockSize, - Progressable progress) throws IOException { - return wrapOutputStream(f, - fileSystem.create(convertToDefaultPath(f), overwrite, bufferSize, replication, blockSize, progress)); - } - - @Override - public FSDataOutputStream create(Path f, FsPermission permission, EnumSet flags, int bufferSize, - short replication, long blockSize, Progressable progress) throws IOException { - return wrapOutputStream(f, - fileSystem.create(convertToDefaultPath(f), permission, flags, bufferSize, replication, blockSize, progress)); - } - - @Override - public FSDataOutputStream create(Path f, FsPermission permission, EnumSet flags, int bufferSize, - short replication, long blockSize, Progressable progress, Options.ChecksumOpt checksumOpt) throws IOException { - return wrapOutputStream(f, fileSystem.create(convertToDefaultPath(f), permission, flags, bufferSize, replication, - blockSize, progress, checksumOpt)); - } - - @Override - public FSDataOutputStream create(Path f, boolean overwrite, int bufferSize, short replication, long blockSize) - throws IOException { - return wrapOutputStream(f, - fileSystem.create(convertToDefaultPath(f), overwrite, bufferSize, replication, blockSize)); - } - - @Override - public FSDataOutputStream append(Path f, int bufferSize, Progressable progress) throws IOException { - return wrapOutputStream(f, fileSystem.append(convertToDefaultPath(f), bufferSize, progress)); - } - - @Override - public boolean rename(Path src, Path dst) throws IOException { - try { - consistencyGuard.waitTillFileAppears(convertToDefaultPath(src)); - } catch (TimeoutException e) { - throw new HoodieException("Timed out waiting for " + src + " to appear", e); - } - - boolean success = fileSystem.rename(convertToDefaultPath(src), convertToDefaultPath(dst)); - - if (success) { - try { - consistencyGuard.waitTillFileAppears(convertToDefaultPath(dst)); - } catch (TimeoutException e) { - throw new HoodieException("Timed out waiting for " + dst + " to appear", e); - } - - try { - consistencyGuard.waitTillFileDisappears(convertToDefaultPath(src)); - } catch (TimeoutException e) { - throw new HoodieException("Timed out waiting for " + src + " to disappear", e); - } - } - return success; - } - - @Override - public boolean delete(Path f, boolean recursive) throws IOException { - boolean success = fileSystem.delete(convertToDefaultPath(f), recursive); - - if (success) { - try { - consistencyGuard.waitTillFileDisappears(f); - } catch (TimeoutException e) { - throw new HoodieException("Timed out waiting for " + f + " to disappear", e); - } - } - return success; - } - - @Override - public FileStatus[] listStatus(Path f) throws IOException { - return fileSystem.listStatus(convertToDefaultPath(f)); - } - - @Override - public Path getWorkingDirectory() { - return convertToHoodiePath(fileSystem.getWorkingDirectory()); - } - - @Override - public void setWorkingDirectory(Path newDir) { - fileSystem.setWorkingDirectory(convertToDefaultPath(newDir)); - } - - @Override - public boolean mkdirs(Path f, FsPermission permission) throws IOException { - boolean success = fileSystem.mkdirs(convertToDefaultPath(f), permission); - if (success) { - try { - consistencyGuard.waitTillFileAppears(convertToDefaultPath(f)); - } catch (TimeoutException e) { - throw new HoodieException("Timed out waiting for directory " + f + " to appear", e); - } - } - return success; - } - - @Override - public FileStatus getFileStatus(Path f) throws IOException { - try { - consistencyGuard.waitTillFileAppears(convertToDefaultPath(f)); - } catch (TimeoutException e) { - // pass - } - return fileSystem.getFileStatus(convertToDefaultPath(f)); - } - - @Override - public String getScheme() { - return uri.getScheme(); - } - - @Override - public String getCanonicalServiceName() { - return fileSystem.getCanonicalServiceName(); - } - - @Override - public String getName() { - return fileSystem.getName(); - } - - @Override - public Path makeQualified(Path path) { - return convertToHoodiePath(fileSystem.makeQualified(convertToDefaultPath(path))); - } - - @Override - public Token getDelegationToken(String renewer) throws IOException { - return fileSystem.getDelegationToken(renewer); - } - - @Override - public Token[] addDelegationTokens(String renewer, Credentials credentials) throws IOException { - return fileSystem.addDelegationTokens(renewer, credentials); - } - - @Override - public FileSystem[] getChildFileSystems() { - return fileSystem.getChildFileSystems(); - } - - @Override - public BlockLocation[] getFileBlockLocations(FileStatus file, long start, long len) throws IOException { - return fileSystem.getFileBlockLocations(file, start, len); - } - - @Override - public BlockLocation[] getFileBlockLocations(Path p, long start, long len) throws IOException { - return fileSystem.getFileBlockLocations(convertToDefaultPath(p), start, len); - } - - @Override - public FsServerDefaults getServerDefaults() throws IOException { - return fileSystem.getServerDefaults(); - } - - @Override - public FsServerDefaults getServerDefaults(Path p) throws IOException { - return fileSystem.getServerDefaults(convertToDefaultPath(p)); - } - - @Override - public Path resolvePath(Path p) throws IOException { - return convertToHoodiePath(fileSystem.resolvePath(convertToDefaultPath(p))); - } - - @Override - public FSDataInputStream open(Path f) throws IOException { - return fileSystem.open(convertToDefaultPath(f)); - } - - @Override - public FSDataOutputStream createNonRecursive(Path f, boolean overwrite, int bufferSize, short replication, - long blockSize, Progressable progress) throws IOException { - Path p = convertToDefaultPath(f); - return wrapOutputStream(p, - fileSystem.createNonRecursive(p, overwrite, bufferSize, replication, blockSize, progress)); - } - - @Override - public FSDataOutputStream createNonRecursive(Path f, FsPermission permission, boolean overwrite, int bufferSize, - short replication, long blockSize, Progressable progress) throws IOException { - Path p = convertToDefaultPath(f); - return wrapOutputStream(p, - fileSystem.createNonRecursive(p, permission, overwrite, bufferSize, replication, blockSize, progress)); - } - - @Override - public FSDataOutputStream createNonRecursive(Path f, FsPermission permission, EnumSet flags, - int bufferSize, short replication, long blockSize, Progressable progress) throws IOException { - Path p = convertToDefaultPath(f); - return wrapOutputStream(p, - fileSystem.createNonRecursive(p, permission, flags, bufferSize, replication, blockSize, progress)); - } - - @Override - public boolean createNewFile(Path f) throws IOException { - boolean newFile = fileSystem.createNewFile(convertToDefaultPath(f)); - if (newFile) { - try { - consistencyGuard.waitTillFileAppears(convertToDefaultPath(f)); - } catch (TimeoutException e) { - throw new HoodieException("Timed out waiting for " + f + " to appear", e); - } - } - return newFile; - } - - @Override - public FSDataOutputStream append(Path f) throws IOException { - return wrapOutputStream(f, fileSystem.append(convertToDefaultPath(f))); - } - - @Override - public FSDataOutputStream append(Path f, int bufferSize) throws IOException { - return wrapOutputStream(f, fileSystem.append(convertToDefaultPath(f), bufferSize)); - } - - @Override - public void concat(Path trg, Path[] psrcs) throws IOException { - Path[] psrcsNew = convertDefaults(psrcs); - fileSystem.concat(convertToDefaultPath(trg), psrcsNew); - try { - consistencyGuard.waitTillFileAppears(convertToDefaultPath(trg)); - } catch (TimeoutException e) { - throw new HoodieException("Timed out waiting for " + trg + " to appear", e); - } - } - - @Override - public short getReplication(Path src) throws IOException { - return fileSystem.getReplication(convertToDefaultPath(src)); - } - - @Override - public boolean setReplication(Path src, short replication) throws IOException { - return fileSystem.setReplication(convertToDefaultPath(src), replication); - } - - @Override - public boolean delete(Path f) throws IOException { - return delete(f, true); - } - - @Override - public boolean deleteOnExit(Path f) throws IOException { - return fileSystem.deleteOnExit(convertToDefaultPath(f)); - } - - @Override - public boolean cancelDeleteOnExit(Path f) { - return fileSystem.cancelDeleteOnExit(convertToDefaultPath(f)); - } - - @Override - public boolean exists(Path f) throws IOException { - return fileSystem.exists(convertToDefaultPath(f)); - } - - @Override - public boolean isDirectory(Path f) throws IOException { - return fileSystem.isDirectory(convertToDefaultPath(f)); - } - - @Override - public boolean isFile(Path f) throws IOException { - return fileSystem.isFile(convertToDefaultPath(f)); - } - - @Override - public long getLength(Path f) throws IOException { - return fileSystem.getLength(convertToDefaultPath(f)); - } - - @Override - public ContentSummary getContentSummary(Path f) throws IOException { - return fileSystem.getContentSummary(convertToDefaultPath(f)); - } - - @Override - public RemoteIterator listCorruptFileBlocks(Path path) throws IOException { - return fileSystem.listCorruptFileBlocks(convertToDefaultPath(path)); - } - - @Override - public FileStatus[] listStatus(Path f, PathFilter filter) throws IOException { - return fileSystem.listStatus(convertToDefaultPath(f), filter); - } - - @Override - public FileStatus[] listStatus(Path[] files) throws IOException { - return fileSystem.listStatus(convertDefaults(files)); - } - - @Override - public FileStatus[] listStatus(Path[] files, PathFilter filter) throws IOException { - return fileSystem.listStatus(convertDefaults(files), filter); - } - - @Override - public FileStatus[] globStatus(Path pathPattern) throws IOException { - return fileSystem.globStatus(convertToDefaultPath(pathPattern)); - } - - @Override - public FileStatus[] globStatus(Path pathPattern, PathFilter filter) throws IOException { - return fileSystem.globStatus(convertToDefaultPath(pathPattern), filter); - } - - @Override - public RemoteIterator listLocatedStatus(Path f) throws IOException { - return fileSystem.listLocatedStatus(convertToDefaultPath(f)); - } - - @Override - public RemoteIterator listFiles(Path f, boolean recursive) throws IOException { - return fileSystem.listFiles(convertToDefaultPath(f), recursive); - } - - @Override - public Path getHomeDirectory() { - return convertToHoodiePath(fileSystem.getHomeDirectory()); - } - - @Override - public boolean mkdirs(Path f) throws IOException { - boolean success = fileSystem.mkdirs(convertToDefaultPath(f)); - if (success) { - try { - consistencyGuard.waitTillFileAppears(convertToDefaultPath(f)); - } catch (TimeoutException e) { - throw new HoodieException("Timed out waiting for directory " + f + " to appear", e); - } - } - return success; - } - - @Override - public void copyFromLocalFile(Path src, Path dst) throws IOException { - fileSystem.copyFromLocalFile(convertToLocalPath(src), convertToDefaultPath(dst)); - try { - consistencyGuard.waitTillFileAppears(convertToDefaultPath(dst)); - } catch (TimeoutException e) { - throw new HoodieException("Timed out waiting for destination " + dst + " to appear", e); - } - } - - @Override - public void moveFromLocalFile(Path[] srcs, Path dst) throws IOException { - fileSystem.moveFromLocalFile(convertLocalPaths(srcs), convertToDefaultPath(dst)); - try { - consistencyGuard.waitTillFileAppears(convertToDefaultPath(dst)); - } catch (TimeoutException e) { - throw new HoodieException("Timed out waiting for destination " + dst + " to appear", e); - } - } - - @Override - public void moveFromLocalFile(Path src, Path dst) throws IOException { - fileSystem.moveFromLocalFile(convertToLocalPath(src), convertToDefaultPath(dst)); - try { - consistencyGuard.waitTillFileAppears(convertToDefaultPath(dst)); - } catch (TimeoutException e) { - throw new HoodieException("Timed out waiting for destination " + dst + " to appear", e); - } - } - - @Override - public void copyFromLocalFile(boolean delSrc, Path src, Path dst) throws IOException { - fileSystem.copyFromLocalFile(delSrc, convertToLocalPath(src), convertToDefaultPath(dst)); - try { - consistencyGuard.waitTillFileAppears(convertToDefaultPath(dst)); - } catch (TimeoutException e) { - throw new HoodieException("Timed out waiting for destination " + dst + " to appear", e); - } - } - - @Override - public void copyFromLocalFile(boolean delSrc, boolean overwrite, Path[] srcs, Path dst) throws IOException { - fileSystem.copyFromLocalFile(delSrc, overwrite, convertLocalPaths(srcs), convertToDefaultPath(dst)); - try { - consistencyGuard.waitTillFileAppears(convertToDefaultPath(dst)); - } catch (TimeoutException e) { - throw new HoodieException("Timed out waiting for destination " + dst + " to appear", e); - } - } - - @Override - public void copyFromLocalFile(boolean delSrc, boolean overwrite, Path src, Path dst) throws IOException { - fileSystem.copyFromLocalFile(delSrc, overwrite, convertToLocalPath(src), convertToDefaultPath(dst)); - try { - consistencyGuard.waitTillFileAppears(convertToDefaultPath(dst)); - } catch (TimeoutException e) { - throw new HoodieException("Timed out waiting for destination " + dst + " to appear", e); - } - } - - @Override - public void copyToLocalFile(Path src, Path dst) throws IOException { - fileSystem.copyToLocalFile(convertToDefaultPath(src), convertToLocalPath(dst)); - } - - @Override - public void moveToLocalFile(Path src, Path dst) throws IOException { - fileSystem.moveToLocalFile(convertToDefaultPath(src), convertToLocalPath(dst)); - } - - @Override - public void copyToLocalFile(boolean delSrc, Path src, Path dst) throws IOException { - fileSystem.copyToLocalFile(delSrc, convertToDefaultPath(src), convertToLocalPath(dst)); - } - - @Override - public void copyToLocalFile(boolean delSrc, Path src, Path dst, boolean useRawLocalFileSystem) throws IOException { - fileSystem.copyToLocalFile(delSrc, convertToDefaultPath(src), convertToLocalPath(dst), useRawLocalFileSystem); - } - - @Override - public Path startLocalOutput(Path fsOutputFile, Path tmpLocalFile) throws IOException { - return convertToHoodiePath( - fileSystem.startLocalOutput(convertToDefaultPath(fsOutputFile), convertToDefaultPath(tmpLocalFile))); - } - - @Override - public void completeLocalOutput(Path fsOutputFile, Path tmpLocalFile) throws IOException { - fileSystem.completeLocalOutput(convertToDefaultPath(fsOutputFile), convertToDefaultPath(tmpLocalFile)); - } - - @Override - public void close() throws IOException { - // Don't close the wrapped `fileSystem` object. This will end up closing it for every thread since it - // could be cached across jvm. We don't own that object anyway. - super.close(); - } - - @Override - public long getUsed() throws IOException { - return fileSystem.getUsed(); - } - - @Override - public long getBlockSize(Path f) throws IOException { - return fileSystem.getBlockSize(convertToDefaultPath(f)); - } - - @Override - public long getDefaultBlockSize() { - return fileSystem.getDefaultBlockSize(); - } - - @Override - public long getDefaultBlockSize(Path f) { - return fileSystem.getDefaultBlockSize(convertToDefaultPath(f)); - } - - @Override - public short getDefaultReplication() { - return fileSystem.getDefaultReplication(); - } - - @Override - public short getDefaultReplication(Path path) { - return fileSystem.getDefaultReplication(convertToDefaultPath(path)); - } - - @Override - public void access(Path path, FsAction mode) throws IOException { - fileSystem.access(convertToDefaultPath(path), mode); - } - - @Override - public void createSymlink(Path target, Path link, boolean createParent) throws IOException { - fileSystem.createSymlink(convertToDefaultPath(target), convertToDefaultPath(link), createParent); - } - - @Override - public FileStatus getFileLinkStatus(Path f) throws IOException { - return fileSystem.getFileLinkStatus(convertToDefaultPath(f)); - } - - @Override - public boolean supportsSymlinks() { - return fileSystem.supportsSymlinks(); - } - - @Override - public Path getLinkTarget(Path f) throws IOException { - return convertToHoodiePath(fileSystem.getLinkTarget(convertToDefaultPath(f))); - } - - @Override - public FileChecksum getFileChecksum(Path f) throws IOException { - return fileSystem.getFileChecksum(convertToDefaultPath(f)); - } - - @Override - public FileChecksum getFileChecksum(Path f, long length) throws IOException { - return fileSystem.getFileChecksum(convertToDefaultPath(f), length); - } - - @Override - public void setVerifyChecksum(boolean verifyChecksum) { - fileSystem.setVerifyChecksum(verifyChecksum); - } - - @Override - public void setWriteChecksum(boolean writeChecksum) { - fileSystem.setWriteChecksum(writeChecksum); - } - - @Override - public FsStatus getStatus() throws IOException { - return fileSystem.getStatus(); - } - - @Override - public FsStatus getStatus(Path p) throws IOException { - return fileSystem.getStatus(convertToDefaultPath(p)); - } - - @Override - public void setPermission(Path p, FsPermission permission) throws IOException { - fileSystem.setPermission(convertToDefaultPath(p), permission); - } - - @Override - public void setOwner(Path p, String username, String groupname) throws IOException { - fileSystem.setOwner(convertToDefaultPath(p), username, groupname); - } - - @Override - public void setTimes(Path p, long mtime, long atime) throws IOException { - fileSystem.setTimes(convertToDefaultPath(p), mtime, atime); - } - - @Override - public Path createSnapshot(Path path, String snapshotName) throws IOException { - return convertToHoodiePath(fileSystem.createSnapshot(convertToDefaultPath(path), snapshotName)); - } - - @Override - public void renameSnapshot(Path path, String snapshotOldName, String snapshotNewName) throws IOException { - fileSystem.renameSnapshot(convertToDefaultPath(path), snapshotOldName, snapshotNewName); - } - - @Override - public void deleteSnapshot(Path path, String snapshotName) throws IOException { - fileSystem.deleteSnapshot(convertToDefaultPath(path), snapshotName); - } - - @Override - public void modifyAclEntries(Path path, List aclSpec) throws IOException { - fileSystem.modifyAclEntries(convertToDefaultPath(path), aclSpec); - } - - @Override - public void removeAclEntries(Path path, List aclSpec) throws IOException { - fileSystem.removeAclEntries(convertToDefaultPath(path), aclSpec); - } - - @Override - public void removeDefaultAcl(Path path) throws IOException { - fileSystem.removeDefaultAcl(convertToDefaultPath(path)); - } - - @Override - public void removeAcl(Path path) throws IOException { - fileSystem.removeAcl(convertToDefaultPath(path)); - } - - @Override - public void setAcl(Path path, List aclSpec) throws IOException { - fileSystem.setAcl(convertToDefaultPath(path), aclSpec); - } - - @Override - public AclStatus getAclStatus(Path path) throws IOException { - return fileSystem.getAclStatus(convertToDefaultPath(path)); - } - - @Override - public void setXAttr(Path path, String name, byte[] value) throws IOException { - fileSystem.setXAttr(convertToDefaultPath(path), name, value); - } - - @Override - public void setXAttr(Path path, String name, byte[] value, EnumSet flag) throws IOException { - fileSystem.setXAttr(convertToDefaultPath(path), name, value, flag); - } - - @Override - public byte[] getXAttr(Path path, String name) throws IOException { - return fileSystem.getXAttr(convertToDefaultPath(path), name); - } - - @Override - public Map getXAttrs(Path path) throws IOException { - return fileSystem.getXAttrs(convertToDefaultPath(path)); - } - - @Override - public Map getXAttrs(Path path, List names) throws IOException { - return fileSystem.getXAttrs(convertToDefaultPath(path), names); - } - - @Override - public List listXAttrs(Path path) throws IOException { - return fileSystem.listXAttrs(convertToDefaultPath(path)); - } - - @Override - public void removeXAttr(Path path, String name) throws IOException { - fileSystem.removeXAttr(convertToDefaultPath(path), name); - } - - @Override - public Configuration getConf() { - return fileSystem.getConf(); - } - - @Override - public void setConf(Configuration conf) { - // ignore this. we will set conf on init - } - - @Override - public int hashCode() { - return fileSystem.hashCode(); - } - - @Override - public boolean equals(Object obj) { - return fileSystem.equals(obj); - } - - @Override - public String toString() { - return fileSystem.toString(); - } - - public Path convertToHoodiePath(Path oldPath) { - return convertPathWithScheme(oldPath, getHoodieScheme(getScheme())); - } - - private Path convertToDefaultPath(Path oldPath) { - return convertPathWithScheme(oldPath, getScheme()); - } - - private Path convertToLocalPath(Path oldPath) { - try { - return convertPathWithScheme(oldPath, FileSystem.getLocal(getConf()).getScheme()); - } catch (IOException e) { - throw new HoodieIOException(e.getMessage(), e); - } - } - - private Path[] convertLocalPaths(Path[] psrcs) { - Path[] psrcsNew = new Path[psrcs.length]; - for (int i = 0; i < psrcs.length; i++) { - psrcsNew[i] = convertToLocalPath(psrcs[i]); - } - return psrcsNew; - } - - private Path[] convertDefaults(Path[] psrcs) { - Path[] psrcsNew = new Path[psrcs.length]; - for (int i = 0; i < psrcs.length; i++) { - psrcsNew[i] = convertToDefaultPath(psrcs[i]); - } - return psrcsNew; - } - - public long getBytesWritten(Path file) { - if (openStreams.containsKey(file.getName())) { - return openStreams.get(file.getName()).getBytesWritten(); - } - // When the file is first written, we do not have a track of it - throw new IllegalArgumentException( - file.toString() + " does not have a open stream. Cannot get the bytes written on the stream"); - } - - public FileSystem getFileSystem() { - return fileSystem; - } -} - From 6d1fa3561bc0d15f0cac691c58c1a770fb25c507 Mon Sep 17 00:00:00 2001 From: visu83 <60400453+visu83@users.noreply.github.com> Date: Fri, 24 Jul 2020 07:59:39 -0700 Subject: [PATCH 3/7] Create DemoFile.java --- DemoFile.java | 166 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 166 insertions(+) create mode 100644 DemoFile.java diff --git a/DemoFile.java b/DemoFile.java new file mode 100644 index 0000000..692f190 --- /dev/null +++ b/DemoFile.java @@ -0,0 +1,166 @@ + package com.shipmentEvents.handlers; + + import java.time.Duration; + import java.util.ArrayList; + import java.util.Iterator; + import java.util.List; + import java.util.Map; + import java.util.Map.Entry; + + import com.amazonaws.regions.Regions; + import com.amazonaws.services.lambda.runtime.Context; + import com.amazonaws.services.lambda.runtime.RequestHandler; + import com.amazonaws.services.lambda.runtime.LambdaLogger; + import com.amazonaws.services.lambda.runtime.events.ScheduledEvent; + import com.amazonaws.services.s3.AmazonS3; + import com.amazonaws.services.s3.AmazonS3ClientBuilder; + import com.amazonaws.services.s3.model.DeleteObjectsRequest; + import com.amazonaws.services.s3.model.DeleteObjectsRequest.KeyVersion; + import com.amazonaws.services.s3.model.ObjectListing; + import com.amazonaws.services.s3.model.S3ObjectSummary; + import com.shipmentEvents.util.Constants; + import java.util.concurrent.ConcurrentHashMap; + + + import org.apache.commons.lang3.tuple.MutablePair; + import org.apache.commons.lang3.tuple.Pair; + + + public class EventHandler implements RequestHandler { + + /** + * Shipment events for a carrier are uploaded to separate S3 buckets based on the source of events. E.g., events originating form + * the hand-held scanner are stored in a separate bucket than the ones from mobile App. The Lambda processes events from multiple + * sources and updates the latest status of the package in a summary S3 bucket every 15 minutes. + * + * The events are stored in following format: + * - Each status update is a file, where the name of the file is tracking number + random id + * - Each file has status and time-stamp as the first 2 lines respectively + * - The time at which is file is stored in S3 is not an indication of the time-stamp of the event + * - Once the status is marked as DELIVERED, we can stop tracking the package + * + * A Sample files looks as below: + * FILE-NAME-> '8787323232232332--55322798-dd29-4a04-97f4-93e18feed554' + * >status:IN TRANSIT + * >timestamp: 1573410202 + * >Other fields like...tracking history and address + */ + public String handleRequest(ScheduledEvent scheduledEvent, Context context) { + + final LambdaLogger logger = context.getLogger(); + try { + processShipmentUpdates(logger); + return "SUCCESS"; + } catch (final Exception ex) { + logger.log(String.format("Failed to process shipment Updates in %s due to %s", scheduledEvent.getAccount(), ex.getMessage())); + throw new RuntimeException(ex); + } + } + + + private void processShipmentUpdates(final LambdaLogger logger) throws InterruptedException { + + final List bucketsToProcess = Constants.BUCKETS_TO_PROCESS; + final ConcurrentHashMap> latestStatusForTrackingNumber = new ConcurrentHashMap>(); + final ConcurrentHashMap> filesToDelete = new ConcurrentHashMap>(); + bucketsToProcess.parallelStream().forEach(bucketName -> { + final List filesProcessed = processEventsInBucket(bucketName, logger, latestStatusForTrackingNumber); + filesToDelete.put(bucketName, filesProcessed); + }); + + final AmazonS3 s3Client = EventHandler.getS3Client(); + //Create a new file in the Constants.SUMMARY_BUCKET + logger.log("Map of statuses -> " + latestStatusForTrackingNumber); + String summaryUpdateName = Long.toString(System.currentTimeMillis()); + + EventHandler.getS3Client().putObject(Constants.SUMMARY_BUCKET, summaryUpdateName, latestStatusForTrackingNumber.toString()); + + Waiter waiter = EventHandler.getS3Client().waiters().objectExists(); + try{ + waiter.run(new WaiterParameters<>(new GetObjectMetadataRequest(Constants.SUMMARY_BUCKET, summaryUpdateName))); + deleteProcessedFiles(filesToDelete); + logger.log("All updates successfully processed"); + } catch (WaiterTimedOutException e){ + throw new RuntimeException("Failed to write sumary status, will be retried in 15 minutes"); + } + } + + private List processEventsInBucket(String bucketName, LambdaLogger logger, ConcurrentHashMap> latestStatusForTrackingNumber) { + + final AmazonS3 s3Client = EventHandler.getS3Client(); + logger.log("Processing Bucket: " + bucketName); + + ObjectListing files = s3Client.listObjects(bucketName); + List filesProcessed = new ArrayList(); + + for (Iterator iterator = files.getObjectSummaries().iterator(); iterator.hasNext(); ) { + S3ObjectSummary summary = (S3ObjectSummary) iterator.next(); + logger.log("Reading Object: " + summary.getKey()); + + String trackingNumber = summary.getKey().split("--")[0]; + Pair lastKnownStatus = latestStatusForTrackingNumber.get(trackingNumber); + + // Check if this shipment has already been delivered, skip this file + if (lastKnownStatus != null && "DELIVERED".equals(lastKnownStatus.getRight())) { + continue; + } + + String fileContents = s3Client.getObjectAsString(bucketName, summary.getKey()); + + if (!isValidFile(fileContents)) { + logger.log(String.format("Skipping invalid file %s", summary.getKey())); + continue; + } + + if (!fileContents.contains("\n")) { + + } + String[] lines = fileContents.split("\n"); + String line1 = lines[0]; + String line2 = lines[1]; + + String status = line1.split(":")[1]; + Long timeStamp = Long.parseLong(line2.split(":")[1]); + + + if (null == lastKnownStatus || lastKnownStatus.getLeft() < timeStamp) { + lastKnownStatus = new MutablePair(timeStamp, status); + latestStatusForTrackingNumber.put(trackingNumber, lastKnownStatus); + } + + //Add to list of processed files + filesProcessed.add(new KeyVersion(summary.getKey())); + logger.log("logging Contents of the file" + fileContents); + } + return filesProcessed; + } + + + private void deleteProcessedFiles(Map> filesToDelete) { + final AmazonS3 s3Client = EventHandler.getS3Client(); + for (Entry> entry : filesToDelete.entrySet()) { + final DeleteObjectsRequest deleteRequest = new DeleteObjectsRequest(entry.getKey()).withKeys(entry.getValue()).withQuiet(false); + s3Client.deleteObjects(deleteRequest); + } + } + + private boolean isValidFile(String fileContents) { + if (!fileContents.contains("\n")) { + return false; + } + String[] lines = fileContents.split("\n"); + for (String l: lines) { + if (!l.contains(":")) { + return false; + } + } + return true; + } + + public static AmazonS3 getS3Client() { + return AmazonS3ClientBuilder.standard().withRegion(Regions.DEFAULT_REGION).build(); + } + + + } + From 5e4a876ab6454ab53bf9e4d56bda82d0c3cbb0ea Mon Sep 17 00:00:00 2001 From: visu83 <60400453+visu83@users.noreply.github.com> Date: Tue, 28 Jul 2020 07:02:57 -0700 Subject: [PATCH 4/7] Delete DemoFile.java --- DemoFile.java | 166 -------------------------------------------------- 1 file changed, 166 deletions(-) delete mode 100644 DemoFile.java diff --git a/DemoFile.java b/DemoFile.java deleted file mode 100644 index 692f190..0000000 --- a/DemoFile.java +++ /dev/null @@ -1,166 +0,0 @@ - package com.shipmentEvents.handlers; - - import java.time.Duration; - import java.util.ArrayList; - import java.util.Iterator; - import java.util.List; - import java.util.Map; - import java.util.Map.Entry; - - import com.amazonaws.regions.Regions; - import com.amazonaws.services.lambda.runtime.Context; - import com.amazonaws.services.lambda.runtime.RequestHandler; - import com.amazonaws.services.lambda.runtime.LambdaLogger; - import com.amazonaws.services.lambda.runtime.events.ScheduledEvent; - import com.amazonaws.services.s3.AmazonS3; - import com.amazonaws.services.s3.AmazonS3ClientBuilder; - import com.amazonaws.services.s3.model.DeleteObjectsRequest; - import com.amazonaws.services.s3.model.DeleteObjectsRequest.KeyVersion; - import com.amazonaws.services.s3.model.ObjectListing; - import com.amazonaws.services.s3.model.S3ObjectSummary; - import com.shipmentEvents.util.Constants; - import java.util.concurrent.ConcurrentHashMap; - - - import org.apache.commons.lang3.tuple.MutablePair; - import org.apache.commons.lang3.tuple.Pair; - - - public class EventHandler implements RequestHandler { - - /** - * Shipment events for a carrier are uploaded to separate S3 buckets based on the source of events. E.g., events originating form - * the hand-held scanner are stored in a separate bucket than the ones from mobile App. The Lambda processes events from multiple - * sources and updates the latest status of the package in a summary S3 bucket every 15 minutes. - * - * The events are stored in following format: - * - Each status update is a file, where the name of the file is tracking number + random id - * - Each file has status and time-stamp as the first 2 lines respectively - * - The time at which is file is stored in S3 is not an indication of the time-stamp of the event - * - Once the status is marked as DELIVERED, we can stop tracking the package - * - * A Sample files looks as below: - * FILE-NAME-> '8787323232232332--55322798-dd29-4a04-97f4-93e18feed554' - * >status:IN TRANSIT - * >timestamp: 1573410202 - * >Other fields like...tracking history and address - */ - public String handleRequest(ScheduledEvent scheduledEvent, Context context) { - - final LambdaLogger logger = context.getLogger(); - try { - processShipmentUpdates(logger); - return "SUCCESS"; - } catch (final Exception ex) { - logger.log(String.format("Failed to process shipment Updates in %s due to %s", scheduledEvent.getAccount(), ex.getMessage())); - throw new RuntimeException(ex); - } - } - - - private void processShipmentUpdates(final LambdaLogger logger) throws InterruptedException { - - final List bucketsToProcess = Constants.BUCKETS_TO_PROCESS; - final ConcurrentHashMap> latestStatusForTrackingNumber = new ConcurrentHashMap>(); - final ConcurrentHashMap> filesToDelete = new ConcurrentHashMap>(); - bucketsToProcess.parallelStream().forEach(bucketName -> { - final List filesProcessed = processEventsInBucket(bucketName, logger, latestStatusForTrackingNumber); - filesToDelete.put(bucketName, filesProcessed); - }); - - final AmazonS3 s3Client = EventHandler.getS3Client(); - //Create a new file in the Constants.SUMMARY_BUCKET - logger.log("Map of statuses -> " + latestStatusForTrackingNumber); - String summaryUpdateName = Long.toString(System.currentTimeMillis()); - - EventHandler.getS3Client().putObject(Constants.SUMMARY_BUCKET, summaryUpdateName, latestStatusForTrackingNumber.toString()); - - Waiter waiter = EventHandler.getS3Client().waiters().objectExists(); - try{ - waiter.run(new WaiterParameters<>(new GetObjectMetadataRequest(Constants.SUMMARY_BUCKET, summaryUpdateName))); - deleteProcessedFiles(filesToDelete); - logger.log("All updates successfully processed"); - } catch (WaiterTimedOutException e){ - throw new RuntimeException("Failed to write sumary status, will be retried in 15 minutes"); - } - } - - private List processEventsInBucket(String bucketName, LambdaLogger logger, ConcurrentHashMap> latestStatusForTrackingNumber) { - - final AmazonS3 s3Client = EventHandler.getS3Client(); - logger.log("Processing Bucket: " + bucketName); - - ObjectListing files = s3Client.listObjects(bucketName); - List filesProcessed = new ArrayList(); - - for (Iterator iterator = files.getObjectSummaries().iterator(); iterator.hasNext(); ) { - S3ObjectSummary summary = (S3ObjectSummary) iterator.next(); - logger.log("Reading Object: " + summary.getKey()); - - String trackingNumber = summary.getKey().split("--")[0]; - Pair lastKnownStatus = latestStatusForTrackingNumber.get(trackingNumber); - - // Check if this shipment has already been delivered, skip this file - if (lastKnownStatus != null && "DELIVERED".equals(lastKnownStatus.getRight())) { - continue; - } - - String fileContents = s3Client.getObjectAsString(bucketName, summary.getKey()); - - if (!isValidFile(fileContents)) { - logger.log(String.format("Skipping invalid file %s", summary.getKey())); - continue; - } - - if (!fileContents.contains("\n")) { - - } - String[] lines = fileContents.split("\n"); - String line1 = lines[0]; - String line2 = lines[1]; - - String status = line1.split(":")[1]; - Long timeStamp = Long.parseLong(line2.split(":")[1]); - - - if (null == lastKnownStatus || lastKnownStatus.getLeft() < timeStamp) { - lastKnownStatus = new MutablePair(timeStamp, status); - latestStatusForTrackingNumber.put(trackingNumber, lastKnownStatus); - } - - //Add to list of processed files - filesProcessed.add(new KeyVersion(summary.getKey())); - logger.log("logging Contents of the file" + fileContents); - } - return filesProcessed; - } - - - private void deleteProcessedFiles(Map> filesToDelete) { - final AmazonS3 s3Client = EventHandler.getS3Client(); - for (Entry> entry : filesToDelete.entrySet()) { - final DeleteObjectsRequest deleteRequest = new DeleteObjectsRequest(entry.getKey()).withKeys(entry.getValue()).withQuiet(false); - s3Client.deleteObjects(deleteRequest); - } - } - - private boolean isValidFile(String fileContents) { - if (!fileContents.contains("\n")) { - return false; - } - String[] lines = fileContents.split("\n"); - for (String l: lines) { - if (!l.contains(":")) { - return false; - } - } - return true; - } - - public static AmazonS3 getS3Client() { - return AmazonS3ClientBuilder.standard().withRegion(Regions.DEFAULT_REGION).build(); - } - - - } - From 306520c66951e2fa4b944ef4cae3f1e49859d400 Mon Sep 17 00:00:00 2001 From: visu83 <60400453+visu83@users.noreply.github.com> Date: Tue, 28 Jul 2020 07:06:05 -0700 Subject: [PATCH 5/7] Create DemoFile.java --- DemoFile.java | 166 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 166 insertions(+) create mode 100644 DemoFile.java diff --git a/DemoFile.java b/DemoFile.java new file mode 100644 index 0000000..692f190 --- /dev/null +++ b/DemoFile.java @@ -0,0 +1,166 @@ + package com.shipmentEvents.handlers; + + import java.time.Duration; + import java.util.ArrayList; + import java.util.Iterator; + import java.util.List; + import java.util.Map; + import java.util.Map.Entry; + + import com.amazonaws.regions.Regions; + import com.amazonaws.services.lambda.runtime.Context; + import com.amazonaws.services.lambda.runtime.RequestHandler; + import com.amazonaws.services.lambda.runtime.LambdaLogger; + import com.amazonaws.services.lambda.runtime.events.ScheduledEvent; + import com.amazonaws.services.s3.AmazonS3; + import com.amazonaws.services.s3.AmazonS3ClientBuilder; + import com.amazonaws.services.s3.model.DeleteObjectsRequest; + import com.amazonaws.services.s3.model.DeleteObjectsRequest.KeyVersion; + import com.amazonaws.services.s3.model.ObjectListing; + import com.amazonaws.services.s3.model.S3ObjectSummary; + import com.shipmentEvents.util.Constants; + import java.util.concurrent.ConcurrentHashMap; + + + import org.apache.commons.lang3.tuple.MutablePair; + import org.apache.commons.lang3.tuple.Pair; + + + public class EventHandler implements RequestHandler { + + /** + * Shipment events for a carrier are uploaded to separate S3 buckets based on the source of events. E.g., events originating form + * the hand-held scanner are stored in a separate bucket than the ones from mobile App. The Lambda processes events from multiple + * sources and updates the latest status of the package in a summary S3 bucket every 15 minutes. + * + * The events are stored in following format: + * - Each status update is a file, where the name of the file is tracking number + random id + * - Each file has status and time-stamp as the first 2 lines respectively + * - The time at which is file is stored in S3 is not an indication of the time-stamp of the event + * - Once the status is marked as DELIVERED, we can stop tracking the package + * + * A Sample files looks as below: + * FILE-NAME-> '8787323232232332--55322798-dd29-4a04-97f4-93e18feed554' + * >status:IN TRANSIT + * >timestamp: 1573410202 + * >Other fields like...tracking history and address + */ + public String handleRequest(ScheduledEvent scheduledEvent, Context context) { + + final LambdaLogger logger = context.getLogger(); + try { + processShipmentUpdates(logger); + return "SUCCESS"; + } catch (final Exception ex) { + logger.log(String.format("Failed to process shipment Updates in %s due to %s", scheduledEvent.getAccount(), ex.getMessage())); + throw new RuntimeException(ex); + } + } + + + private void processShipmentUpdates(final LambdaLogger logger) throws InterruptedException { + + final List bucketsToProcess = Constants.BUCKETS_TO_PROCESS; + final ConcurrentHashMap> latestStatusForTrackingNumber = new ConcurrentHashMap>(); + final ConcurrentHashMap> filesToDelete = new ConcurrentHashMap>(); + bucketsToProcess.parallelStream().forEach(bucketName -> { + final List filesProcessed = processEventsInBucket(bucketName, logger, latestStatusForTrackingNumber); + filesToDelete.put(bucketName, filesProcessed); + }); + + final AmazonS3 s3Client = EventHandler.getS3Client(); + //Create a new file in the Constants.SUMMARY_BUCKET + logger.log("Map of statuses -> " + latestStatusForTrackingNumber); + String summaryUpdateName = Long.toString(System.currentTimeMillis()); + + EventHandler.getS3Client().putObject(Constants.SUMMARY_BUCKET, summaryUpdateName, latestStatusForTrackingNumber.toString()); + + Waiter waiter = EventHandler.getS3Client().waiters().objectExists(); + try{ + waiter.run(new WaiterParameters<>(new GetObjectMetadataRequest(Constants.SUMMARY_BUCKET, summaryUpdateName))); + deleteProcessedFiles(filesToDelete); + logger.log("All updates successfully processed"); + } catch (WaiterTimedOutException e){ + throw new RuntimeException("Failed to write sumary status, will be retried in 15 minutes"); + } + } + + private List processEventsInBucket(String bucketName, LambdaLogger logger, ConcurrentHashMap> latestStatusForTrackingNumber) { + + final AmazonS3 s3Client = EventHandler.getS3Client(); + logger.log("Processing Bucket: " + bucketName); + + ObjectListing files = s3Client.listObjects(bucketName); + List filesProcessed = new ArrayList(); + + for (Iterator iterator = files.getObjectSummaries().iterator(); iterator.hasNext(); ) { + S3ObjectSummary summary = (S3ObjectSummary) iterator.next(); + logger.log("Reading Object: " + summary.getKey()); + + String trackingNumber = summary.getKey().split("--")[0]; + Pair lastKnownStatus = latestStatusForTrackingNumber.get(trackingNumber); + + // Check if this shipment has already been delivered, skip this file + if (lastKnownStatus != null && "DELIVERED".equals(lastKnownStatus.getRight())) { + continue; + } + + String fileContents = s3Client.getObjectAsString(bucketName, summary.getKey()); + + if (!isValidFile(fileContents)) { + logger.log(String.format("Skipping invalid file %s", summary.getKey())); + continue; + } + + if (!fileContents.contains("\n")) { + + } + String[] lines = fileContents.split("\n"); + String line1 = lines[0]; + String line2 = lines[1]; + + String status = line1.split(":")[1]; + Long timeStamp = Long.parseLong(line2.split(":")[1]); + + + if (null == lastKnownStatus || lastKnownStatus.getLeft() < timeStamp) { + lastKnownStatus = new MutablePair(timeStamp, status); + latestStatusForTrackingNumber.put(trackingNumber, lastKnownStatus); + } + + //Add to list of processed files + filesProcessed.add(new KeyVersion(summary.getKey())); + logger.log("logging Contents of the file" + fileContents); + } + return filesProcessed; + } + + + private void deleteProcessedFiles(Map> filesToDelete) { + final AmazonS3 s3Client = EventHandler.getS3Client(); + for (Entry> entry : filesToDelete.entrySet()) { + final DeleteObjectsRequest deleteRequest = new DeleteObjectsRequest(entry.getKey()).withKeys(entry.getValue()).withQuiet(false); + s3Client.deleteObjects(deleteRequest); + } + } + + private boolean isValidFile(String fileContents) { + if (!fileContents.contains("\n")) { + return false; + } + String[] lines = fileContents.split("\n"); + for (String l: lines) { + if (!l.contains(":")) { + return false; + } + } + return true; + } + + public static AmazonS3 getS3Client() { + return AmazonS3ClientBuilder.standard().withRegion(Regions.DEFAULT_REGION).build(); + } + + + } + From e3c77c6322bcdcdbdbc82dfe389ff93fd31e8d80 Mon Sep 17 00:00:00 2001 From: visu83 <60400453+visu83@users.noreply.github.com> Date: Tue, 28 Jul 2020 07:20:45 -0700 Subject: [PATCH 6/7] Delete DemoFile.java --- DemoFile.java | 166 -------------------------------------------------- 1 file changed, 166 deletions(-) delete mode 100644 DemoFile.java diff --git a/DemoFile.java b/DemoFile.java deleted file mode 100644 index 692f190..0000000 --- a/DemoFile.java +++ /dev/null @@ -1,166 +0,0 @@ - package com.shipmentEvents.handlers; - - import java.time.Duration; - import java.util.ArrayList; - import java.util.Iterator; - import java.util.List; - import java.util.Map; - import java.util.Map.Entry; - - import com.amazonaws.regions.Regions; - import com.amazonaws.services.lambda.runtime.Context; - import com.amazonaws.services.lambda.runtime.RequestHandler; - import com.amazonaws.services.lambda.runtime.LambdaLogger; - import com.amazonaws.services.lambda.runtime.events.ScheduledEvent; - import com.amazonaws.services.s3.AmazonS3; - import com.amazonaws.services.s3.AmazonS3ClientBuilder; - import com.amazonaws.services.s3.model.DeleteObjectsRequest; - import com.amazonaws.services.s3.model.DeleteObjectsRequest.KeyVersion; - import com.amazonaws.services.s3.model.ObjectListing; - import com.amazonaws.services.s3.model.S3ObjectSummary; - import com.shipmentEvents.util.Constants; - import java.util.concurrent.ConcurrentHashMap; - - - import org.apache.commons.lang3.tuple.MutablePair; - import org.apache.commons.lang3.tuple.Pair; - - - public class EventHandler implements RequestHandler { - - /** - * Shipment events for a carrier are uploaded to separate S3 buckets based on the source of events. E.g., events originating form - * the hand-held scanner are stored in a separate bucket than the ones from mobile App. The Lambda processes events from multiple - * sources and updates the latest status of the package in a summary S3 bucket every 15 minutes. - * - * The events are stored in following format: - * - Each status update is a file, where the name of the file is tracking number + random id - * - Each file has status and time-stamp as the first 2 lines respectively - * - The time at which is file is stored in S3 is not an indication of the time-stamp of the event - * - Once the status is marked as DELIVERED, we can stop tracking the package - * - * A Sample files looks as below: - * FILE-NAME-> '8787323232232332--55322798-dd29-4a04-97f4-93e18feed554' - * >status:IN TRANSIT - * >timestamp: 1573410202 - * >Other fields like...tracking history and address - */ - public String handleRequest(ScheduledEvent scheduledEvent, Context context) { - - final LambdaLogger logger = context.getLogger(); - try { - processShipmentUpdates(logger); - return "SUCCESS"; - } catch (final Exception ex) { - logger.log(String.format("Failed to process shipment Updates in %s due to %s", scheduledEvent.getAccount(), ex.getMessage())); - throw new RuntimeException(ex); - } - } - - - private void processShipmentUpdates(final LambdaLogger logger) throws InterruptedException { - - final List bucketsToProcess = Constants.BUCKETS_TO_PROCESS; - final ConcurrentHashMap> latestStatusForTrackingNumber = new ConcurrentHashMap>(); - final ConcurrentHashMap> filesToDelete = new ConcurrentHashMap>(); - bucketsToProcess.parallelStream().forEach(bucketName -> { - final List filesProcessed = processEventsInBucket(bucketName, logger, latestStatusForTrackingNumber); - filesToDelete.put(bucketName, filesProcessed); - }); - - final AmazonS3 s3Client = EventHandler.getS3Client(); - //Create a new file in the Constants.SUMMARY_BUCKET - logger.log("Map of statuses -> " + latestStatusForTrackingNumber); - String summaryUpdateName = Long.toString(System.currentTimeMillis()); - - EventHandler.getS3Client().putObject(Constants.SUMMARY_BUCKET, summaryUpdateName, latestStatusForTrackingNumber.toString()); - - Waiter waiter = EventHandler.getS3Client().waiters().objectExists(); - try{ - waiter.run(new WaiterParameters<>(new GetObjectMetadataRequest(Constants.SUMMARY_BUCKET, summaryUpdateName))); - deleteProcessedFiles(filesToDelete); - logger.log("All updates successfully processed"); - } catch (WaiterTimedOutException e){ - throw new RuntimeException("Failed to write sumary status, will be retried in 15 minutes"); - } - } - - private List processEventsInBucket(String bucketName, LambdaLogger logger, ConcurrentHashMap> latestStatusForTrackingNumber) { - - final AmazonS3 s3Client = EventHandler.getS3Client(); - logger.log("Processing Bucket: " + bucketName); - - ObjectListing files = s3Client.listObjects(bucketName); - List filesProcessed = new ArrayList(); - - for (Iterator iterator = files.getObjectSummaries().iterator(); iterator.hasNext(); ) { - S3ObjectSummary summary = (S3ObjectSummary) iterator.next(); - logger.log("Reading Object: " + summary.getKey()); - - String trackingNumber = summary.getKey().split("--")[0]; - Pair lastKnownStatus = latestStatusForTrackingNumber.get(trackingNumber); - - // Check if this shipment has already been delivered, skip this file - if (lastKnownStatus != null && "DELIVERED".equals(lastKnownStatus.getRight())) { - continue; - } - - String fileContents = s3Client.getObjectAsString(bucketName, summary.getKey()); - - if (!isValidFile(fileContents)) { - logger.log(String.format("Skipping invalid file %s", summary.getKey())); - continue; - } - - if (!fileContents.contains("\n")) { - - } - String[] lines = fileContents.split("\n"); - String line1 = lines[0]; - String line2 = lines[1]; - - String status = line1.split(":")[1]; - Long timeStamp = Long.parseLong(line2.split(":")[1]); - - - if (null == lastKnownStatus || lastKnownStatus.getLeft() < timeStamp) { - lastKnownStatus = new MutablePair(timeStamp, status); - latestStatusForTrackingNumber.put(trackingNumber, lastKnownStatus); - } - - //Add to list of processed files - filesProcessed.add(new KeyVersion(summary.getKey())); - logger.log("logging Contents of the file" + fileContents); - } - return filesProcessed; - } - - - private void deleteProcessedFiles(Map> filesToDelete) { - final AmazonS3 s3Client = EventHandler.getS3Client(); - for (Entry> entry : filesToDelete.entrySet()) { - final DeleteObjectsRequest deleteRequest = new DeleteObjectsRequest(entry.getKey()).withKeys(entry.getValue()).withQuiet(false); - s3Client.deleteObjects(deleteRequest); - } - } - - private boolean isValidFile(String fileContents) { - if (!fileContents.contains("\n")) { - return false; - } - String[] lines = fileContents.split("\n"); - for (String l: lines) { - if (!l.contains(":")) { - return false; - } - } - return true; - } - - public static AmazonS3 getS3Client() { - return AmazonS3ClientBuilder.standard().withRegion(Regions.DEFAULT_REGION).build(); - } - - - } - From eb9f6883f0be3cf4cf77d7908c89a6479f5e09f0 Mon Sep 17 00:00:00 2001 From: visu83 <60400453+visu83@users.noreply.github.com> Date: Tue, 28 Jul 2020 08:13:09 -0700 Subject: [PATCH 7/7] Create DemoFile.java --- DemoFile.java | 166 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 166 insertions(+) create mode 100644 DemoFile.java diff --git a/DemoFile.java b/DemoFile.java new file mode 100644 index 0000000..692f190 --- /dev/null +++ b/DemoFile.java @@ -0,0 +1,166 @@ + package com.shipmentEvents.handlers; + + import java.time.Duration; + import java.util.ArrayList; + import java.util.Iterator; + import java.util.List; + import java.util.Map; + import java.util.Map.Entry; + + import com.amazonaws.regions.Regions; + import com.amazonaws.services.lambda.runtime.Context; + import com.amazonaws.services.lambda.runtime.RequestHandler; + import com.amazonaws.services.lambda.runtime.LambdaLogger; + import com.amazonaws.services.lambda.runtime.events.ScheduledEvent; + import com.amazonaws.services.s3.AmazonS3; + import com.amazonaws.services.s3.AmazonS3ClientBuilder; + import com.amazonaws.services.s3.model.DeleteObjectsRequest; + import com.amazonaws.services.s3.model.DeleteObjectsRequest.KeyVersion; + import com.amazonaws.services.s3.model.ObjectListing; + import com.amazonaws.services.s3.model.S3ObjectSummary; + import com.shipmentEvents.util.Constants; + import java.util.concurrent.ConcurrentHashMap; + + + import org.apache.commons.lang3.tuple.MutablePair; + import org.apache.commons.lang3.tuple.Pair; + + + public class EventHandler implements RequestHandler { + + /** + * Shipment events for a carrier are uploaded to separate S3 buckets based on the source of events. E.g., events originating form + * the hand-held scanner are stored in a separate bucket than the ones from mobile App. The Lambda processes events from multiple + * sources and updates the latest status of the package in a summary S3 bucket every 15 minutes. + * + * The events are stored in following format: + * - Each status update is a file, where the name of the file is tracking number + random id + * - Each file has status and time-stamp as the first 2 lines respectively + * - The time at which is file is stored in S3 is not an indication of the time-stamp of the event + * - Once the status is marked as DELIVERED, we can stop tracking the package + * + * A Sample files looks as below: + * FILE-NAME-> '8787323232232332--55322798-dd29-4a04-97f4-93e18feed554' + * >status:IN TRANSIT + * >timestamp: 1573410202 + * >Other fields like...tracking history and address + */ + public String handleRequest(ScheduledEvent scheduledEvent, Context context) { + + final LambdaLogger logger = context.getLogger(); + try { + processShipmentUpdates(logger); + return "SUCCESS"; + } catch (final Exception ex) { + logger.log(String.format("Failed to process shipment Updates in %s due to %s", scheduledEvent.getAccount(), ex.getMessage())); + throw new RuntimeException(ex); + } + } + + + private void processShipmentUpdates(final LambdaLogger logger) throws InterruptedException { + + final List bucketsToProcess = Constants.BUCKETS_TO_PROCESS; + final ConcurrentHashMap> latestStatusForTrackingNumber = new ConcurrentHashMap>(); + final ConcurrentHashMap> filesToDelete = new ConcurrentHashMap>(); + bucketsToProcess.parallelStream().forEach(bucketName -> { + final List filesProcessed = processEventsInBucket(bucketName, logger, latestStatusForTrackingNumber); + filesToDelete.put(bucketName, filesProcessed); + }); + + final AmazonS3 s3Client = EventHandler.getS3Client(); + //Create a new file in the Constants.SUMMARY_BUCKET + logger.log("Map of statuses -> " + latestStatusForTrackingNumber); + String summaryUpdateName = Long.toString(System.currentTimeMillis()); + + EventHandler.getS3Client().putObject(Constants.SUMMARY_BUCKET, summaryUpdateName, latestStatusForTrackingNumber.toString()); + + Waiter waiter = EventHandler.getS3Client().waiters().objectExists(); + try{ + waiter.run(new WaiterParameters<>(new GetObjectMetadataRequest(Constants.SUMMARY_BUCKET, summaryUpdateName))); + deleteProcessedFiles(filesToDelete); + logger.log("All updates successfully processed"); + } catch (WaiterTimedOutException e){ + throw new RuntimeException("Failed to write sumary status, will be retried in 15 minutes"); + } + } + + private List processEventsInBucket(String bucketName, LambdaLogger logger, ConcurrentHashMap> latestStatusForTrackingNumber) { + + final AmazonS3 s3Client = EventHandler.getS3Client(); + logger.log("Processing Bucket: " + bucketName); + + ObjectListing files = s3Client.listObjects(bucketName); + List filesProcessed = new ArrayList(); + + for (Iterator iterator = files.getObjectSummaries().iterator(); iterator.hasNext(); ) { + S3ObjectSummary summary = (S3ObjectSummary) iterator.next(); + logger.log("Reading Object: " + summary.getKey()); + + String trackingNumber = summary.getKey().split("--")[0]; + Pair lastKnownStatus = latestStatusForTrackingNumber.get(trackingNumber); + + // Check if this shipment has already been delivered, skip this file + if (lastKnownStatus != null && "DELIVERED".equals(lastKnownStatus.getRight())) { + continue; + } + + String fileContents = s3Client.getObjectAsString(bucketName, summary.getKey()); + + if (!isValidFile(fileContents)) { + logger.log(String.format("Skipping invalid file %s", summary.getKey())); + continue; + } + + if (!fileContents.contains("\n")) { + + } + String[] lines = fileContents.split("\n"); + String line1 = lines[0]; + String line2 = lines[1]; + + String status = line1.split(":")[1]; + Long timeStamp = Long.parseLong(line2.split(":")[1]); + + + if (null == lastKnownStatus || lastKnownStatus.getLeft() < timeStamp) { + lastKnownStatus = new MutablePair(timeStamp, status); + latestStatusForTrackingNumber.put(trackingNumber, lastKnownStatus); + } + + //Add to list of processed files + filesProcessed.add(new KeyVersion(summary.getKey())); + logger.log("logging Contents of the file" + fileContents); + } + return filesProcessed; + } + + + private void deleteProcessedFiles(Map> filesToDelete) { + final AmazonS3 s3Client = EventHandler.getS3Client(); + for (Entry> entry : filesToDelete.entrySet()) { + final DeleteObjectsRequest deleteRequest = new DeleteObjectsRequest(entry.getKey()).withKeys(entry.getValue()).withQuiet(false); + s3Client.deleteObjects(deleteRequest); + } + } + + private boolean isValidFile(String fileContents) { + if (!fileContents.contains("\n")) { + return false; + } + String[] lines = fileContents.split("\n"); + for (String l: lines) { + if (!l.contains(":")) { + return false; + } + } + return true; + } + + public static AmazonS3 getS3Client() { + return AmazonS3ClientBuilder.standard().withRegion(Regions.DEFAULT_REGION).build(); + } + + + } +