You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@apex.apache.org by th...@apache.org on 2016/04/22 05:30:33 UTC
[3/3] incubator-apex-malhar git commit: APEXMALHAR-1965 Removed
bucket concept from WAL and made it fault-tolerant
APEXMALHAR-1965 Removed bucket concept from WAL and made it fault-tolerant
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/commit/dad9f56f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/tree/dad9f56f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/diff/dad9f56f
Branch: refs/heads/master
Commit: dad9f56fd7d2460e76b7c0d1adb3331079798941
Parents: 988fc4b
Author: Chandni Singh <cs...@apache.org>
Authored: Thu Apr 14 22:56:00 2016 -0700
Committer: Chandni Singh <cs...@apache.org>
Committed: Thu Apr 21 19:35:25 2016 -0700
----------------------------------------------------------------------
.../apex/malhar/lib/utils/FileContextUtils.java | 65 ++
.../apache/apex/malhar/lib/utils/IOUtils.java | 84 +++
.../apex/malhar/lib/wal/FileSystemWAL.java | 686 +++++++++++++++----
.../org/apache/apex/malhar/lib/wal/WAL.java | 99 +--
.../apex/malhar/lib/wal/package-info.java | 21 +
.../apex/malhar/lib/utils/IOUtilsTest.java | 92 +++
.../apex/malhar/lib/wal/FileSystemWALTest.java | 483 +++++++++----
7 files changed, 1208 insertions(+), 322 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/dad9f56f/library/src/main/java/org/apache/apex/malhar/lib/utils/FileContextUtils.java
----------------------------------------------------------------------
diff --git a/library/src/main/java/org/apache/apex/malhar/lib/utils/FileContextUtils.java b/library/src/main/java/org/apache/apex/malhar/lib/utils/FileContextUtils.java
new file mode 100644
index 0000000..00976c2
--- /dev/null
+++ b/library/src/main/java/org/apache/apex/malhar/lib/utils/FileContextUtils.java
@@ -0,0 +1,65 @@
+/**
+ * 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.apex.malhar.lib.utils;
+
+import java.net.URI;
+
+import javax.annotation.Nullable;
+import javax.validation.constraints.NotNull;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileContext;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.UnsupportedFileSystemException;
+
+import com.google.common.base.Preconditions;
+
+public class FileContextUtils
+{
+ public static FileContext getFileContext(@NotNull Path path) throws UnsupportedFileSystemException
+ {
+ return getFileContext(path, null);
+ }
+
+ public static FileContext getFileContext(@NotNull String path) throws UnsupportedFileSystemException
+ {
+ Preconditions.checkNotNull(path, "path");
+ return getFileContext(new Path(path), null);
+ }
+
+ public static FileContext getFileContext(@NotNull String path, @Nullable Configuration conf)
+ throws UnsupportedFileSystemException
+ {
+ Preconditions.checkNotNull(path, "path");
+ return getFileContext(new Path(path), conf);
+ }
+
+ public static FileContext getFileContext(@NotNull Path path, @Nullable Configuration conf)
+ throws UnsupportedFileSystemException
+ {
+ Preconditions.checkNotNull(path, "path");
+ URI pathUri = path.toUri();
+
+ if (pathUri.getScheme() != null) {
+ return FileContext.getFileContext(pathUri, conf == null ? new Configuration() : conf);
+ } else {
+ return FileContext.getFileContext(conf == null ? new Configuration() : conf);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/dad9f56f/library/src/main/java/org/apache/apex/malhar/lib/utils/IOUtils.java
----------------------------------------------------------------------
diff --git a/library/src/main/java/org/apache/apex/malhar/lib/utils/IOUtils.java b/library/src/main/java/org/apache/apex/malhar/lib/utils/IOUtils.java
new file mode 100644
index 0000000..123e41a
--- /dev/null
+++ b/library/src/main/java/org/apache/apex/malhar/lib/utils/IOUtils.java
@@ -0,0 +1,84 @@
+/**
+ * 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.apex.malhar.lib.utils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class IOUtils
+{
+ /**
+ * Utility method to copy partial data from input stream to output stream.<br/>
+ * This can be deprecated when we can move to a newer version of hadoop common.
+ *
+ * @param inputStream input stream
+ * @param length length of bytes from the start of input stream
+ * @param outputStream output stream
+ * @throws IOException
+ */
+ public static void copyPartial(InputStream inputStream, long length, OutputStream outputStream)
+ throws IOException
+ {
+ copyPartial(inputStream, 0, length, outputStream);
+ }
+
+ /**
+ * Utility method to copy partial data from input stream to output stream.<br/>
+ * This can be deprecated when we can move to a newer version of hadoop common.
+ *
+ * @param inputStream inputStream
+ * @param fromOffset offset in the input stream to copy data from
+ * @param length length of bytes in the input stream
+ * @param outputStream output stream
+ * @throws IOException
+ */
+ public static void copyPartial(InputStream inputStream, long fromOffset, long length, OutputStream outputStream)
+ throws IOException
+ {
+ if (fromOffset > 0) {
+ long skippedBytes = inputStream.skip(fromOffset);
+ if (skippedBytes < fromOffset) {
+ LOGGER.debug("unable to skip bytes {} {}", skippedBytes, fromOffset);
+ }
+ }
+ byte[] buffer;
+ if (length < COPY_BUFFER_SIZE) {
+ buffer = new byte[(int)length];
+ } else {
+ buffer = new byte[COPY_BUFFER_SIZE];
+ }
+
+ int bytesRead = 0;
+
+ while (bytesRead < length) {
+ long remainingBytes = length - bytesRead;
+ int bytesToWrite = remainingBytes < COPY_BUFFER_SIZE ? (int)remainingBytes : COPY_BUFFER_SIZE;
+ bytesRead += inputStream.read(buffer);
+ outputStream.write(buffer, 0, bytesToWrite);
+ }
+ }
+
+ private static final int COPY_BUFFER_SIZE = 1024;
+ private static final Logger LOGGER = LoggerFactory.getLogger(IOUtils.class);
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/dad9f56f/library/src/main/java/org/apache/apex/malhar/lib/wal/FileSystemWAL.java
----------------------------------------------------------------------
diff --git a/library/src/main/java/org/apache/apex/malhar/lib/wal/FileSystemWAL.java b/library/src/main/java/org/apache/apex/malhar/lib/wal/FileSystemWAL.java
index 6f0c719..6edf204 100644
--- a/library/src/main/java/org/apache/apex/malhar/lib/wal/FileSystemWAL.java
+++ b/library/src/main/java/org/apache/apex/malhar/lib/wal/FileSystemWAL.java
@@ -18,252 +18,676 @@
*/
package org.apache.apex.malhar.lib.wal;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
import java.io.IOException;
+import java.util.EnumSet;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+
+import javax.validation.constraints.Min;
+import javax.validation.constraints.NotNull;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.apex.malhar.lib.utils.FileContextUtils;
+import org.apache.apex.malhar.lib.utils.IOUtils;
+import org.apache.hadoop.fs.CreateFlag;
+import org.apache.hadoop.fs.FileContext;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Options;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.hadoop.fs.Syncable;
+import org.apache.hadoop.fs.local.LocalFs;
+import org.apache.hadoop.fs.local.RawLocalFs;
+
+import com.google.common.base.Preconditions;
+
+import com.datatorrent.api.annotation.Stateless;
+import com.datatorrent.netlet.util.Slice;
-import com.datatorrent.lib.fileaccess.FileAccess;
-
-public class FileSystemWAL<T> implements WAL<T, FileSystemWAL.WalPointer>
+/**
+ * A WAL implementation that is file based.
+ * <p/>
+ * Note:<br/>
+ * The FileSystem Writer and Reader operations should not alternate because intermingling these operations will cause
+ * problems. Typically the WAL Reader will only used in recovery.<br/>
+ *
+ * Also this implementation is thread unsafe- the filesystem wal writer and reader operations should be performed in
+ * operator's thread.
+ */
+public class FileSystemWAL implements WAL<FileSystemWAL.FileSystemWALReader, FileSystemWAL.FileSystemWALWriter>
{
- private final String WAL_FILE_PREFIX = "wal";
- private final FileAccess bfs;
- private final Serde<T> serializer;
- private final long bucketKey;
- private String baseDir;
- public FileSystemWAL(FileAccess bfs, WAL.Serde<T> serializer, long bucketKey)
+ @NotNull
+ private String filePath;
+
+ //max length of the file
+ @Min(0)
+ private long maxLength;
+
+ private FileSystemWALPointer walStartPointer = new FileSystemWALPointer(0, 0);
+
+ @NotNull
+ private FileSystemWAL.FileSystemWALReader fileSystemWALReader = new FileSystemWALReader(this);
+
+ @NotNull
+ private FileSystemWAL.FileSystemWALWriter fileSystemWALWriter = new FileSystemWALWriter(this);
+
+ //part => tmp file path;
+ private final Map<Integer, String> tempPartFiles = new TreeMap<>();
+
+ private long lastCheckpointedWindow = Stateless.WINDOW_ID;
+
+ @Override
+ public void setup()
{
- this.bfs = bfs;
- this.serializer = serializer;
- this.bucketKey = bucketKey;
+ try {
+ FileContext fileContext = FileContextUtils.getFileContext(filePath);
+ if (maxLength == 0) {
+ maxLength = fileContext.getDefaultFileSystem().getServerDefaults().getBlockSize();
+ }
+ fileSystemWALWriter.open(fileContext);
+ fileSystemWALReader.open(fileContext);
+ } catch (IOException e) {
+ throw new RuntimeException("during setup", e);
+ }
}
@Override
- public WALReader<T, WalPointer> getReader() throws IOException
+ public void beforeCheckpoint(long window)
{
- return new RollingFSWalReader();
+ try {
+ lastCheckpointedWindow = window;
+ fileSystemWALWriter.flush();
+ } catch (IOException e) {
+ throw new RuntimeException("during before cp", e);
+ }
}
+ /**
+ * A temporary WAL file is not renamed as soon as it is closed (completed) instead it is tagged to be renamed with
+ * respect to the window it gets closed. The actual renaming is deferred until the window gets committed.<br/>
+ * <br/>
+ * The reason to do so is because after a failure, when the WAL is restored to a checkpoint it might expect to write
+ * to a temporary file. For example, the writer starts writing to WAL_10_x.tmp in window 59 and then gets
+ * check-pointed.<br/>
+ * In window 60 it completes writing to WAL_10_x.tmp and then closes it. If the file is renamed right then and there
+ * is a failure then after restoration with the state of window 59, it will expect to continue writing to WAL_10_x.tmp
+ * which will not be found.<br/>
+ * <br/>
+ * Writing to the final target (WAL part file) is avoided unless it is certain that the file will not be open to write
+ * again and this happens when the window in which the file was completed gets committed.
+ *
+ * @param window committed window
+ */
@Override
- public WALWriter<T, WalPointer> getWriter() throws IOException
+ public void committed(long window)
{
- return new RollingFSWALWriter();
+ try {
+ fileSystemWALWriter.finalizeFiles(window);
+ } catch (IOException e) {
+ throw new RuntimeException("during committed", e);
+ }
}
- protected String getName(int id)
+ @Override
+ public void teardown()
{
- return baseDir + "/" + WAL_FILE_PREFIX + "_" + id;
+ try {
+ fileSystemWALReader.close();
+ fileSystemWALWriter.close();
+ } catch (IOException e) {
+ throw new RuntimeException("during teardown", e);
+ }
}
- public String getBaseDir()
+ protected long getLastCheckpointedWindow()
{
- return baseDir;
+ return lastCheckpointedWindow;
}
- public void setBaseDir(String baseDir)
+ protected String getPartFilePath(int partNumber)
{
- this.baseDir = baseDir;
+ return filePath + "_" + partNumber;
}
- public static class WalPointer implements Comparable<WalPointer>
+ @Override
+ public FileSystemWALReader getReader()
{
- public int fileId;
- public long offset;
+ return fileSystemWALReader;
+ }
+
+ /**
+ * Sets the File System WAL Reader. This can be used to override the default wal reader.
+ *
+ * @param fileSystemWALReader wal reader.
+ */
+ public void setFileSystemWALReader(@NotNull FileSystemWALReader fileSystemWALReader)
+ {
+ this.fileSystemWALReader = Preconditions.checkNotNull(fileSystemWALReader, "filesystem wal reader");
+ }
+
+ @Override
+ public FileSystemWALWriter getWriter()
+ {
+ return fileSystemWALWriter;
+ }
+
+ /**
+ * Sets the File System WAL Writer. This can be used to override the default wal writer.
+ *
+ * @param fileSystemWALWriter wal writer.
+ */
+ public void setFileSystemWALWriter(@NotNull FileSystemWALWriter fileSystemWALWriter)
+ {
+ this.fileSystemWALWriter = Preconditions.checkNotNull(fileSystemWALWriter, "filesystem wal writer");
+ }
+
+ /**
+ * @return WAL file path
+ */
+ public String getFilePath()
+ {
+ return filePath;
+ }
+
+ /**
+ * Sets the WAL file path.
+ *
+ * @param filePath WAL file path
+ */
+ public void setFilePath(@NotNull String filePath)
+ {
+ this.filePath = Preconditions.checkNotNull(filePath, "filePath");
+ }
+
+ /**
+ * @return max length of a WAL part file.
+ */
+ public long getMaxLength()
+ {
+ return maxLength;
+ }
- public WalPointer(int fileId, long offset)
+ /**
+ * Sets the maximum length of a WAL part file.
+ *
+ * @param maxLength max length of the WAL part file
+ */
+ public void setMaxLength(long maxLength)
+ {
+ this.maxLength = maxLength;
+ }
+
+ public static class FileSystemWALPointer implements Comparable<FileSystemWALPointer>
+ {
+ private final int partNum;
+ private long offset;
+
+ private FileSystemWALPointer()
{
- this.fileId = fileId;
+ //for kryo
+ partNum = -1;
+ }
+
+ public FileSystemWALPointer(int partNum, long offset)
+ {
+ this.partNum = partNum;
this.offset = offset;
}
@Override
- public int compareTo(WalPointer o)
+ public int compareTo(@NotNull FileSystemWALPointer o)
{
- if (this.fileId < o.fileId) {
- return -1;
- }
- if (this.fileId > o.fileId) {
- return 1;
- }
- if (this.offset < o.offset) {
- return -1;
- }
- if (this.offset > o.offset) {
- return 1;
- }
- return 0;
+ int partNumComparison = Integer.compare(this.partNum, o.partNum);
+ return partNumComparison == 0 ? Long.compare(this.offset, o.offset) : partNumComparison;
+ }
+
+ public int getPartNum()
+ {
+ return partNum;
+ }
+
+ public long getOffset()
+ {
+ return offset;
}
@Override
public String toString()
{
- return "WalPointer{" +
- "fileId=" + fileId +
- ", offset=" + offset +
- '}';
+ return "FileSystemWalPointer{" + "partNum=" + partNum + ", offset=" + offset + '}';
}
}
- public class RollingFSWalReader implements WAL.WALReader<T, WalPointer>
+ /**
+ * A FileSystem Wal Reader
+ */
+ public static class FileSystemWALReader implements WAL.WALReader<FileSystemWALPointer>
{
- private WALReader<T, Long> currentReader = null;
- private WalPointer currOffset;
- private T entry;
- private boolean eof = false;
+ private FileSystemWALPointer currentPointer;
+
+ private transient DataInputStream inputStream;
+ private transient Path currentOpenPath;
+ private transient boolean isOpenPathTmp;
- public RollingFSWalReader() throws IOException
+ private final FileSystemWAL fileSystemWAL;
+ private transient FileContext fileContext;
+
+ private FileSystemWALReader()
{
- this.currOffset = new WalPointer(0, 0);
+ //for kryo
+ fileSystemWAL = null;
}
- @Override
- public WalPointer getOffset()
+ public FileSystemWALReader(@NotNull FileSystemWAL fileSystemWal)
{
- return currOffset;
+ this.fileSystemWAL = Preconditions.checkNotNull(fileSystemWal, "wal");
+ currentPointer = new FileSystemWALPointer(fileSystemWal.walStartPointer.partNum,
+ fileSystemWal.walStartPointer.offset);
}
- @Override
- public void close() throws IOException
+ protected void open(@NotNull FileContext fileContext) throws IOException
+ {
+ this.fileContext = Preconditions.checkNotNull(fileContext, "fileContext");
+ }
+
+ protected void close() throws IOException
{
- if (currentReader != null) {
- currentReader.close();
- currentReader = null;
+ if (inputStream != null) {
+ inputStream.close();
+ inputStream = null;
}
}
@Override
- public void seek(WalPointer offset) throws IOException
+ public void seek(FileSystemWALPointer pointer) throws IOException
{
- currentReader = new FSWal<T>(bfs, serializer, bucketKey, getName(offset.fileId)).getReader();
- currentReader.seek(offset.offset);
- currOffset = offset;
- eof = false;
+ Preconditions.checkArgument(pointer.compareTo(fileSystemWAL.walStartPointer) >= 0, "invalid pointer");
+ if (inputStream != null) {
+ close();
+ }
+ inputStream = getInputStream(pointer);
+ currentPointer = pointer;
}
/**
* Move to the next WAL segment.
*
- * @return
+ * @return true if the next part file exists and is opened; false otherwise.
* @throws IOException
*/
private boolean nextSegment() throws IOException
{
- if (currentReader == null) {
- int fileId = currOffset.fileId;
- fileId++;
- if (bfs.exists(bucketKey, getName(fileId))) {
- currentReader = new FSWal<T>(bfs, serializer, bucketKey, getName(fileId)).getReader();
- currOffset = new WalPointer(fileId, 0);
- return true;
- } else {
- eof = true;
- entry = null;
- return false;
- }
+ if (inputStream != null) {
+ close();
}
- return true;
+
+ currentPointer = new FileSystemWALPointer(currentPointer.partNum + 1, 0);
+ inputStream = getInputStream(currentPointer);
+
+ return inputStream != null;
}
- @Override
- public boolean advance() throws IOException
+ private DataInputStream getInputStream(FileSystemWALPointer walPointer) throws IOException
{
- if (eof) {
- return false;
+ Preconditions.checkArgument(inputStream == null, "input stream not null");
+ Path pathToReadFrom;
+ String tmpPath = fileSystemWAL.tempPartFiles.get(walPointer.getPartNum());
+ if (tmpPath != null) {
+ pathToReadFrom = new Path(tmpPath);
+ isOpenPathTmp = true;
+ } else {
+ pathToReadFrom = new Path(fileSystemWAL.getPartFilePath(walPointer.partNum));
+ isOpenPathTmp = false;
}
- boolean ret = false;
+ LOG.debug("path to read {} and pointer {}", pathToReadFrom, walPointer);
+ if (fileContext.util().exists(pathToReadFrom)) {
+ DataInputStream stream = fileContext.open(pathToReadFrom);
+ if (walPointer.offset > 0) {
+ stream.skip(walPointer.offset);
+ }
+ currentOpenPath = pathToReadFrom;
+ return stream;
+ }
+ return null;
+ }
+
+ @Override
+ public Slice next() throws IOException
+ {
do {
- if (!nextSegment()) {
- break;
+ if (inputStream == null) {
+ inputStream = getInputStream(currentPointer);
}
- ret = currentReader.advance();
- if (ret) {
- entry = currentReader.get();
- break;
- } else {
+ if (inputStream != null && isOpenPathTmp && !fileSystemWAL.tempPartFiles.containsKey(currentPointer.partNum)) {
+ //if the tmp path was finalized the path may not exist any more
close();
- currentReader = null;
+ inputStream = getInputStream(currentPointer);
}
- } while (true);
+ if (inputStream != null && currentPointer.offset < fileContext.getFileStatus(currentOpenPath).getLen()) {
+ int len = inputStream.readInt();
+ Preconditions.checkState(len >= 0, "negative length");
+
+ byte[] data = new byte[len];
+ inputStream.readFully(data);
- return ret;
+ currentPointer.offset += data.length + 4;
+ return new Slice(data);
+ }
+ } while (nextSegment());
+ close();
+ return null;
}
@Override
- public T get()
+ public FileSystemWALPointer getStartPointer()
{
- return entry;
+ return fileSystemWAL.walStartPointer;
}
}
- public class RollingFSWALWriter implements WAL.WALWriter<T, WalPointer>
+ /**
+ * A FileSystem WAL Writer.
+ */
+ public static class FileSystemWALWriter implements WAL.WALWriter<FileSystemWALPointer>
{
- private transient WALWriter<T, Long> currentWriter = null;
- private int walFileId = -1;
- private long bucketKey = 0;
- private long maxSegmentSize = 128 * 1024 * 1024; // 128MB default.
+ private FileSystemWALPointer currentPointer = new FileSystemWALPointer(0, 0);
+ private transient DataOutputStream outputStream;
- @Override
- public void close() throws IOException
+ //windowId => Latest part which can be finalized.
+ private final Map<Long, Integer> pendingFinalization = new TreeMap<>();
+
+ private final FileSystemWAL fileSystemWAL;
+ private transient FileContext fileContext;
+
+ private int latestFinalizedPart = -1;
+ private int lowestDeletedPart = -1;
+
+ private FileSystemWALWriter()
+ {
+ //for kryo
+ fileSystemWAL = null;
+ }
+
+ public FileSystemWALWriter(@NotNull FileSystemWAL fileSystemWal)
+ {
+ this.fileSystemWAL = Preconditions.checkNotNull(fileSystemWal, "wal");
+ }
+
+ protected void open(@NotNull FileContext fileContext) throws IOException
+ {
+ this.fileContext = Preconditions.checkNotNull(fileContext, "file context");
+ recover();
+ }
+
+ private void recover() throws IOException
+ {
+ LOG.debug("current point", currentPointer);
+ String tmpFilePath = fileSystemWAL.tempPartFiles.get(currentPointer.getPartNum());
+ if (tmpFilePath != null) {
+
+ Path tmpPath = new Path(tmpFilePath);
+ if (fileContext.util().exists(tmpPath)) {
+ LOG.debug("tmp path exists {}", tmpPath);
+
+ outputStream = getOutputStream(new FileSystemWALPointer(currentPointer.partNum, 0));
+ DataInputStream inputStreamOldTmp = fileContext.open(tmpPath);
+
+ IOUtils.copyPartial(inputStreamOldTmp, currentPointer.offset, outputStream);
+
+ outputStream.flush();
+ //remove old tmp
+ inputStreamOldTmp.close();
+ LOG.debug("delete tmp {}", tmpPath);
+ fileContext.delete(tmpPath, true);
+ }
+ }
+
+ //find all valid path names
+ Set<String> validPathNames = new HashSet<>();
+ for (Map.Entry<Integer, String> entry : fileSystemWAL.tempPartFiles.entrySet()) {
+ if (entry.getKey() <= currentPointer.partNum) {
+ validPathNames.add(new Path(entry.getValue()).getName());
+ }
+ }
+ LOG.debug("valid names {}", validPathNames);
+
+ //there can be a failure just between the flush and the actual checkpoint which can leave some stray tmp files
+ //which aren't accounted by tmp files map
+ Path walPath = new Path(fileSystemWAL.filePath);
+ Path parentWAL = walPath.getParent();
+ if (parentWAL != null && fileContext.util().exists(parentWAL)) {
+ RemoteIterator<FileStatus> remoteIterator = fileContext.listStatus(parentWAL);
+ while (remoteIterator.hasNext()) {
+ FileStatus status = remoteIterator.next();
+ String fileName = status.getPath().getName();
+ if (fileName.startsWith(walPath.getName()) && fileName.endsWith(TMP_EXTENSION) &&
+ !validPathNames.contains(fileName)) {
+ LOG.debug("delete stray tmp {}", status.getPath());
+ fileContext.delete(status.getPath(), true);
+ }
+ }
+ }
+
+ }
+
+ protected void close() throws IOException
{
- if (currentWriter != null) {
- currentWriter.close();
+ if (outputStream != null) {
+ outputStream.close();
+ outputStream = null;
+ LOG.debug("closed {}", currentPointer.partNum);
}
}
@Override
- public int append(T entry) throws IOException
+ public int append(Slice entry) throws IOException
{
- logMayRoll();
- if (currentWriter == null) {
- currentWriter = new FSWal(bfs, serializer, bucketKey, getName(walFileId)).getWriter();
+ if (outputStream == null) {
+ outputStream = getOutputStream(currentPointer);
+ }
+
+ int entryLength = entry.length + 4;
+
+ // rotate if needed
+ if (shouldRotate(entryLength)) {
+ rotate(true);
}
- int len = currentWriter.append(entry);
- return len;
+ outputStream.writeInt(entry.length);
+ outputStream.write(entry.buffer, entry.offset, entry.length);
+ currentPointer.offset += entryLength;
+
+ if (currentPointer.offset == fileSystemWAL.maxLength) {
+ //if the file is completed then we can rotate it. do not have to wait for next entry
+ rotate(false);
+ }
+
+ return entryLength;
}
@Override
- public void flush() throws IOException
+ public FileSystemWALPointer getPointer()
{
- if (currentWriter != null) {
- currentWriter.flush();
- }
+ return currentPointer;
}
@Override
- public WalPointer getOffset()
+ public void delete(FileSystemWALPointer pointer) throws IOException
{
- return new WalPointer(walFileId, currentWriter.getOffset());
+ if (pointer.compareTo(currentPointer) <= 0) {
+ fileSystemWAL.walStartPointer = pointer;
+ deleteFinalizedParts(pointer);
+ }
}
- void logMayRoll() throws IOException
+ private void deleteFinalizedParts(FileSystemWALPointer pointer) throws IOException
{
- if (currentWriter == null) {
- newWalSegment();
- } else {
- if (currentWriter.getOffset() >= maxSegmentSize) {
- currentWriter.close();
- newWalSegment();
+
+ int lastPartDeleted = -1;
+
+ //delete all part files completely which are smaller than pointer.partNum
+ for (int i = lowestDeletedPart + 1; i < pointer.partNum; i++) {
+ if (i <= latestFinalizedPart) {
+ //delete a part only if it is finalized.
+ Path partPath = new Path(fileSystemWAL.getPartFilePath(i));
+ if (fileContext.util().exists(partPath)) {
+ LOG.debug("delete {}", partPath);
+ fileContext.delete(partPath, true);
+ lastPartDeleted = i;
+ } else {
+ break;
+ }
+ }
+ }
+ if (lastPartDeleted != -1) {
+ lowestDeletedPart = lastPartDeleted;
+ }
+
+ //truncate the pointer.partNum till pointer.offset
+ if (pointer.partNum <= latestFinalizedPart && pointer.offset > 0) {
+ String part = fileSystemWAL.getPartFilePath(pointer.partNum);
+ Path inputPartPath = new Path(part);
+ long length = fileContext.getFileStatus(inputPartPath).getLen();
+
+ LOG.debug("truncate {} from {} length {}", part, pointer.offset, length);
+
+ if (length > pointer.offset) {
+
+ String temp = getTmpFilePath(part);
+ Path tmpPart = new Path(temp);
+
+ DataInputStream inputStream = fileContext.open(inputPartPath);
+ DataOutputStream outputStream = fileContext.create(tmpPart, EnumSet.of(CreateFlag.CREATE, CreateFlag.APPEND),
+ Options.CreateOpts.CreateParent.createParent());
+
+ IOUtils.copyPartial(inputStream, pointer.offset, length - pointer.offset, outputStream);
+ inputStream.close();
+ outputStream.close();
+
+ if (fileSystemWAL.walStartPointer.partNum == pointer.partNum) {
+ //Since the file is abridged
+ fileSystemWAL.walStartPointer.offset = 0;
+ }
+
+ fileContext.rename(tmpPart, inputPartPath, Options.Rename.OVERWRITE);
+ }
+ }
+ }
+
+ protected void flush() throws IOException
+ {
+ if (outputStream != null) {
+ if (fileContext.getDefaultFileSystem() instanceof LocalFs ||
+ fileContext.getDefaultFileSystem() instanceof RawLocalFs) {
+ //until the stream is closed on the local FS, readers don't see any data.
+ close();
+ } else {
+ Syncable syncableOutputStream = (Syncable)outputStream;
+ syncableOutputStream.hflush();
+ syncableOutputStream.hsync();
}
}
}
- private void newWalSegment() throws IOException
+ protected boolean shouldRotate(int entryLength)
{
- walFileId++;
- currentWriter = new FSWal(bfs, serializer, bucketKey, getName(walFileId)).getWriter();
+ return currentPointer.offset + entryLength > fileSystemWAL.maxLength;
}
- public long getMaxSegmentSize()
+ protected void rotate(boolean openNextFile) throws IOException
{
- return maxSegmentSize;
+ flush();
+ close();
+ //all parts up to current part can be finalized.
+ pendingFinalization.put(fileSystemWAL.getLastCheckpointedWindow(), currentPointer.partNum);
+
+ LOG.debug("rotate {} to {}", currentPointer.partNum, currentPointer.partNum + 1);
+ currentPointer = new FileSystemWALPointer(currentPointer.partNum + 1, 0);
+ if (openNextFile) {
+ //if adding the new entry to the file can cause the current file to exceed the max length then it is rotated.
+ outputStream = getOutputStream(currentPointer);
+ }
}
- public void setMaxSegmentSize(long maxSegmentSize)
+ protected void finalizeFiles(long window) throws IOException
{
- this.maxSegmentSize = maxSegmentSize;
+ if (!fileSystemWAL.tempPartFiles.isEmpty()) {
+ //finalize temporary files
+ int largestPartAvailable = fileSystemWAL.tempPartFiles.keySet().iterator().next();
+
+ Iterator<Map.Entry<Long, Integer>> pendingFinalizeIter = pendingFinalization.entrySet().iterator();
+ while (pendingFinalizeIter.hasNext()) {
+ Map.Entry<Long, Integer> entry = pendingFinalizeIter.next();
+
+ if (entry.getKey() >= window) {
+ //finalize files which were requested for finalization in the window < committed window
+ break;
+ }
+ pendingFinalizeIter.remove();
+
+ int partToFinalizeTill = entry.getValue();
+ for (int i = largestPartAvailable; i <= partToFinalizeTill; i++) {
+ String tmpToFinalize = fileSystemWAL.tempPartFiles.remove(i);
+ Path tmpPath = new Path(tmpToFinalize);
+
+ if (fileContext.util().exists(tmpPath)) {
+ LOG.debug("finalize {} of part {}", tmpToFinalize, i);
+ fileContext.rename(tmpPath, new Path(fileSystemWAL.getPartFilePath(i)), Options.Rename.OVERWRITE);
+ latestFinalizedPart = i;
+ }
+ }
+ largestPartAvailable = partToFinalizeTill + 1;
+ }
+ }
+
+ if (lowestDeletedPart != -1 && lowestDeletedPart < fileSystemWAL.walStartPointer.partNum) {
+ //delete any pending finalized files which were not deleted when the delete request was made.
+ deleteFinalizedParts(fileSystemWAL.walStartPointer);
+ }
}
+
+ private DataOutputStream getOutputStream(FileSystemWALPointer pointer) throws IOException
+ {
+ Preconditions.checkArgument(outputStream == null, "output stream is not null");
+
+ if (pointer.offset > 0 && (fileContext.getDefaultFileSystem() instanceof LocalFs ||
+ fileContext.getDefaultFileSystem() instanceof RawLocalFs)) {
+ //on local file system the stream is closed instead of flush so we open it again in append mode if the
+ //offset > 0.
+ return fileContext.create(new Path(fileSystemWAL.tempPartFiles.get(pointer.partNum)),
+ EnumSet.of(CreateFlag.CREATE, CreateFlag.APPEND), Options.CreateOpts.CreateParent.createParent());
+ }
+
+ String partFile = fileSystemWAL.getPartFilePath(pointer.partNum);
+ String tmpFilePath = getTmpFilePath(partFile);
+ fileSystemWAL.tempPartFiles.put(pointer.partNum, tmpFilePath);
+
+ Preconditions.checkArgument(pointer.offset == 0, "offset > 0");
+ LOG.debug("open {} => {}", pointer.partNum, tmpFilePath);
+ outputStream = fileContext.create(new Path(tmpFilePath),
+ EnumSet.of(CreateFlag.CREATE, CreateFlag.OVERWRITE), Options.CreateOpts.CreateParent.createParent());
+ return outputStream;
+ }
+
}
+
+ private static String getTmpFilePath(String filePath)
+ {
+ return filePath + '.' + System.currentTimeMillis() + TMP_EXTENSION;
+ }
+
+ private static final String TMP_EXTENSION = ".tmp";
+
+ private static final Logger LOG = LoggerFactory.getLogger(FileSystemWAL.class);
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/dad9f56f/library/src/main/java/org/apache/apex/malhar/lib/wal/WAL.java
----------------------------------------------------------------------
diff --git a/library/src/main/java/org/apache/apex/malhar/lib/wal/WAL.java b/library/src/main/java/org/apache/apex/malhar/lib/wal/WAL.java
index 5a11b76..10d8359 100644
--- a/library/src/main/java/org/apache/apex/malhar/lib/wal/WAL.java
+++ b/library/src/main/java/org/apache/apex/malhar/lib/wal/WAL.java
@@ -18,113 +18,84 @@
*/
package org.apache.apex.malhar.lib.wal;
-import java.io.Closeable;
import java.io.IOException;
+import com.datatorrent.netlet.util.Slice;
+
/**
* This interface represents a write ahead log that can be used by operator.
* the WAL is split into two interfaces, a WALWriter which allows writing
* data, and WALReader which provides iterator like interface to read entries
- * writen to the WAL.
+ * written to the WAL.
*
- * @param <T> Tuple type
- * @param <P> WAL Pointer Type.
+ * @param <READER> Type of WAL Reader
+ * @param <WRITER> WAL Pointer Type.
*/
-public interface WAL<T, P>
+public interface WAL<READER extends WAL.WALReader, WRITER extends WAL.WALWriter>
{
- WALReader<T, P> getReader() throws IOException;
+ void setup();
+
+ void teardown();
+
+ void beforeCheckpoint(long window);
+
+ void committed(long window);
- WALWriter<T, P> getWriter() throws IOException;
+ READER getReader();
+
+ WRITER getWriter();
/**
* Provides iterator like interface to read entries from the WAL.
- * @param <T> type of WAL entries
* @param <P> type of Pointer in the WAL
*/
- interface WALReader<T, P> extends Closeable
+ interface WALReader<P>
{
/**
- * Close WAL after read.
- *
- * @param offset seek offset.
- * @throws IOException
- */
- @Override
- void close() throws IOException;
-
- /**
* Seek to middle of the WAL. This is used primarily during recovery,
* when we need to start recovering data from middle of WAL file.
*/
- void seek(P offset) throws IOException;
+ void seek(P pointer) throws IOException;
/**
- * Advance WAL by one entry, returns true if it can advance, else false
- * in case of any other error throws an Exception.
- *
- * @return true if next data item is read successfully, false if data can not be read.
- * @throws IOException
+ * Returns the next entry from WAL. It returns null if data isn't available.
+ * @return next entry when available; null otherwise.
*/
- boolean advance() throws IOException;
+ Slice next() throws IOException;
/**
- * Return current entry from WAL, returns null if end of file has reached.
+ * Returns the start pointer from which data is available to read.<br/>
+ * WAL Writer supports purging of aged data so the start pointer will change over time.
*
- * @return MutableKeyValue
- */
- T get();
-
- /**
- * Return the offset corresponding to the last read entry.
- * @return
+ * @return the start pointer from which the data is available to read.
*/
- P getOffset();
+ P getStartPointer();
}
/**
* Provide method to write entries to the WAL.
- * @param <T>
- * @param <P>
+ * @param <P> type of Pointer in the WAL
*/
- interface WALWriter<T, P>
+ interface WALWriter<P>
{
/**
- * flush pending data to disk and close file.
- *
- * @throws IOException
- */
- void close() throws IOException;
-
- /**
- * Write an entry to the WAL, this operation need not flush the data.
+ * Write an entry to the WAL
*/
- int append(T entry) throws IOException;
+ int append(Slice entry) throws IOException;
/**
- * Flush data to persistent storage.
*
- * @throws IOException
+ * @return current pointer in the WAL.
*/
- void flush() throws IOException;
+ P getPointer();
/**
- * Returns size of the WAL, last part of the log may not be persisted on disk.
- * In case of file backed WAL this will be the size of file, in case of kafka
- * like log, this will be similar to the message offset.
+ * Deletes the WAL data till the given pointer.<br/>
*
- * @return The log size
+ * @param pointer pointer in the WAL
+ * @throws IOException
*/
- P getOffset();
- }
-
- /**
- * Serializer interface used while reading and writing entries to the WAL.
- * @param <T>
- */
- interface Serde<T>
- {
- byte[] toBytes(T tuple);
+ void delete(P pointer) throws IOException;
- T fromBytes(byte[] data);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/dad9f56f/library/src/main/java/org/apache/apex/malhar/lib/wal/package-info.java
----------------------------------------------------------------------
diff --git a/library/src/main/java/org/apache/apex/malhar/lib/wal/package-info.java b/library/src/main/java/org/apache/apex/malhar/lib/wal/package-info.java
new file mode 100644
index 0000000..511962d
--- /dev/null
+++ b/library/src/main/java/org/apache/apex/malhar/lib/wal/package-info.java
@@ -0,0 +1,21 @@
+/**
+ * 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.
+ */
+
+@org.apache.hadoop.classification.InterfaceStability.Unstable
+package org.apache.apex.malhar.lib.wal;
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/dad9f56f/library/src/test/java/org/apache/apex/malhar/lib/utils/IOUtilsTest.java
----------------------------------------------------------------------
diff --git a/library/src/test/java/org/apache/apex/malhar/lib/utils/IOUtilsTest.java b/library/src/test/java/org/apache/apex/malhar/lib/utils/IOUtilsTest.java
new file mode 100644
index 0000000..9ddad77
--- /dev/null
+++ b/library/src/test/java/org/apache/apex/malhar/lib/utils/IOUtilsTest.java
@@ -0,0 +1,92 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.apex.malhar.lib.utils;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.util.EnumSet;
+import java.util.Random;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.fs.CreateFlag;
+import org.apache.hadoop.fs.FileContext;
+import org.apache.hadoop.fs.Options;
+import org.apache.hadoop.fs.Path;
+
+public class IOUtilsTest
+{
+
+ @Test
+ public void testCopyPartial() throws IOException
+ {
+ testCopyPartialHelper(1024, 0, 512);
+ }
+
+ @Test
+ public void testCopyPartialLarge() throws IOException
+ {
+ testCopyPartialHelper(2048, 0, 1068);
+ }
+
+ @Test
+ public void testCopyFromOffset() throws IOException
+ {
+ testCopyPartialHelper(2048, 1024, 1024);
+ }
+
+ private void testCopyPartialHelper(int dataSize, int offset, long size) throws IOException
+ {
+ FileUtils.deleteQuietly(new File("target/IOUtilsTest"));
+ File file = new File("target/IOUtilsTest/testCopyPartial/input");
+ createDataFile(file, dataSize);
+
+ FileContext fileContext = FileContext.getFileContext();
+ DataInputStream inputStream = fileContext.open(new Path(file.getAbsolutePath()));
+
+ Path output = new Path("target/IOUtilsTest/testCopyPartial/output");
+ DataOutputStream outputStream = fileContext.create(output, EnumSet
+ .of(CreateFlag.CREATE, CreateFlag.OVERWRITE), Options.CreateOpts.CreateParent.createParent());
+
+ if (offset == 0) {
+ IOUtils.copyPartial(inputStream, size, outputStream);
+ } else {
+ IOUtils.copyPartial(inputStream, offset, size, outputStream);
+ }
+
+ outputStream.close();
+
+ Assert.assertTrue("output exists", fileContext.util().exists(output));
+ Assert.assertEquals("output size", size, fileContext.getFileStatus(output).getLen());
+// FileUtils.deleteQuietly(new File("target/IOUtilsTest"));
+ }
+
+ private void createDataFile(File file, int dataSize) throws IOException
+ {
+ byte[] val = new byte[dataSize];
+ Random random = new Random();
+ random.nextBytes(val);
+
+ FileUtils.write(file, new String(val));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/dad9f56f/library/src/test/java/org/apache/apex/malhar/lib/wal/FileSystemWALTest.java
----------------------------------------------------------------------
diff --git a/library/src/test/java/org/apache/apex/malhar/lib/wal/FileSystemWALTest.java b/library/src/test/java/org/apache/apex/malhar/lib/wal/FileSystemWALTest.java
index f22d0da..cf8bb34 100644
--- a/library/src/test/java/org/apache/apex/malhar/lib/wal/FileSystemWALTest.java
+++ b/library/src/test/java/org/apache/apex/malhar/lib/wal/FileSystemWALTest.java
@@ -21,30 +21,63 @@ package org.apache.apex.malhar.lib.wal;
import java.io.File;
import java.io.IOException;
import java.util.Random;
-import java.util.Set;
import org.junit.Assert;
+import org.junit.Rule;
import org.junit.Test;
+import org.junit.rules.TestWatcher;
+import org.junit.runner.Description;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.commons.io.FileUtils;
+import org.apache.apex.malhar.lib.utils.FileContextUtils;
+import org.apache.hadoop.fs.FileContext;
+import org.apache.hadoop.fs.Path;
-import com.google.common.collect.Sets;
-
-import com.datatorrent.lib.fileaccess.FileAccessFSImpl;
+import com.datatorrent.lib.util.KryoCloneUtils;
+import com.datatorrent.lib.util.TestUtils;
+import com.datatorrent.netlet.util.Slice;
public class FileSystemWALTest
{
- static final Random rand = new Random();
- private static final Logger logger = LoggerFactory.getLogger(FileSystemWALTest.class);
- File file = new File("target/hds");
+ private static final Random RAND = new Random();
- static byte[] genRandomByteArray(int len)
+ private static Slice getRandomSlice(int len)
{
byte[] val = new byte[len];
- rand.nextBytes(val);
- return val;
+ RAND.nextBytes(val);
+ return new Slice(val);
+ }
+
+ private class TestMeta extends TestWatcher
+ {
+ private String targetDir;
+ FileSystemWAL fsWAL = new FileSystemWAL();
+
+ @Override
+ protected void starting(Description description)
+ {
+ TestUtils.deleteTargetTestClassFolder(description);
+ targetDir = "target/" + description.getClassName() + "/" + description.getMethodName();
+ fsWAL = new FileSystemWAL();
+ fsWAL.setFilePath(targetDir + "/WAL");
+ }
+
+ @Override
+ protected void finished(Description description)
+ {
+ TestUtils.deleteTargetTestClassFolder(description);
+ }
+ }
+
+ @Rule
+ public TestMeta testMeta = new TestMeta();
+
+ @Test
+ public void testSerde() throws IOException
+ {
+ FileSystemWAL deserialized = KryoCloneUtils.cloneObject(testMeta.fsWAL);
+ Assert.assertNotNull("File System WAL", deserialized);
}
/**
@@ -56,33 +89,24 @@ public class FileSystemWALTest
@Test
public void testWalWriteAndRead() throws IOException
{
- FileUtils.deleteDirectory(file);
- FileAccessFSImpl bfs = new MockFileAccess();
- bfs.setBasePath(file.getAbsolutePath());
- bfs.init();
-
+ testMeta.fsWAL.setup();
int numTuples = 100;
- FSWal<byte[]> fsWal = new FSWal<>(bfs, new ByteArraySerde(), 1, file.getAbsoluteFile().toString() + "/1/WAL-0");
-
- WAL.WALWriter wWriter = fsWal.getWriter();
+ FileSystemWAL.FileSystemWALWriter fsWALWriter = testMeta.fsWAL.getWriter();
for (int i = 0; i < numTuples; i++) {
- int len = rand.nextInt(100);
- wWriter.append(genRandomByteArray(len));
+ int len = RAND.nextInt(100);
+ fsWALWriter.append(getRandomSlice(len));
}
- wWriter.close();
+ fsWALWriter.rotate(true);
+ testMeta.fsWAL.beforeCheckpoint(0);
+ testMeta.fsWAL.committed(0);
+ File walFile = new File(testMeta.fsWAL.getPartFilePath(0));
+ Assert.assertEquals("WAL file created ", true, walFile.exists());
- File wal0 = new File(file.getAbsoluteFile().toString() + "/1/WAL-0");
- Assert.assertEquals("WAL file created ", true, wal0.exists());
+ FileSystemWAL.FileSystemWALReader fsWALReader = testMeta.fsWAL.getReader();
+ assertNumTuplesRead(fsWALReader, numTuples);
- WAL.WALReader wReader = fsWal.getReader();
- int tuples = 0;
- while (wReader.advance()) {
- byte[] data = (byte[])wReader.get();
- tuples++;
- }
- wReader.close();
- Assert.assertEquals("Write and read same number of tuples ", numTuples, tuples);
+ testMeta.fsWAL.teardown();
}
/**
@@ -92,35 +116,33 @@ public class FileSystemWALTest
@Test
public void testWalSkip() throws IOException
{
- FileUtils.deleteDirectory(file);
- FileAccessFSImpl bfs = new MockFileAccess();
- bfs.setBasePath(file.getAbsolutePath());
- bfs.init();
+ testMeta.fsWAL.setup();
+
+ FileSystemWAL.FileSystemWALWriter fsWALWriter = testMeta.fsWAL.getWriter();
- long offset = 0;
- FSWal<byte[]> wal = new FSWal<>(bfs, new ByteArraySerde(), 1, file.getAbsoluteFile().toString() + "WAL-0");
- WAL.WALWriter wWriter = wal.getWriter();
int totalTuples = 100;
- int recoveryTuples = 30;
+ int totalBytes = 0;
+ long offset = 0;
for (int i = 0; i < totalTuples; i++) {
- wWriter.append(genRandomByteArray(100));
- if (i == recoveryTuples) {
- offset = (long)wWriter.getOffset();
+ if (i == 30) {
+ offset = totalBytes;
}
+ totalBytes += fsWALWriter.append(getRandomSlice(100));
}
- logger.info("total file size is " + wWriter.getOffset() + " recovery offset is " + offset);
- wWriter.close();
-
- WAL.WALReader wReader = wal.getReader();
- wReader.seek(offset);
- int read = 0;
- while (wReader.advance()) {
- read++;
- wReader.get();
- }
- wReader.close();
- Assert.assertEquals("Number of tuples read after skipping", read, (totalTuples - recoveryTuples - 1));
+ fsWALWriter.rotate(true);
+ testMeta.fsWAL.beforeCheckpoint(0);
+ testMeta.fsWAL.committed(0);
+
+ File walFile = new File(testMeta.fsWAL.getPartFilePath(0));
+ Assert.assertEquals("WAL file size ", totalBytes, walFile.length());
+
+ FileSystemWAL.FileSystemWALReader fsWALReader = testMeta.fsWAL.getReader();
+
+ fsWALReader.seek(new FileSystemWAL.FileSystemWALPointer(0, offset));
+ assertNumTuplesRead(fsWALReader, totalTuples - 30);
+
+ testMeta.fsWAL.teardown();
}
/**
@@ -132,105 +154,312 @@ public class FileSystemWALTest
@Test
public void testWalRolling() throws IOException
{
- File file = new File("target/hds");
- FileUtils.deleteDirectory(file);
- FileAccessFSImpl bfs = new MockFileAccess();
- bfs.setBasePath(file.getAbsolutePath());
- bfs.init();
+ testMeta.fsWAL.setMaxLength(32 * 1024);
+ testMeta.fsWAL.setup();
- FileSystemWAL<byte[]> wal = new FileSystemWAL<>(bfs, new ByteArraySerde(), 1);
- wal.setBaseDir(file.getAbsoluteFile().toString());
- FileSystemWAL<byte[]>.RollingFSWALWriter writer = (FileSystemWAL<byte[]>.RollingFSWALWriter)wal.getWriter();
- writer.setMaxSegmentSize(32 * 1024);
+ FileSystemWAL.FileSystemWALWriter fsWALWriter = testMeta.fsWAL.getWriter();
/* write each record of size 1K, each file will have 32 records, last file will have
* (count % 32) records. */
- int wrote = 100;
- for (int i = 0; i < wrote; i++) {
- writer.append(genRandomByteArray(1020));
- }
- writer.close();
+ int numRecords = 100;
+ write1KRecords(fsWALWriter, numRecords);
+
+ testMeta.fsWAL.beforeCheckpoint(0);
+ testMeta.fsWAL.committed(0);
/** Read from start till the end */
- FileSystemWAL<byte[]>.RollingFSWalReader reader = (FileSystemWAL<byte[]>.RollingFSWalReader)wal.getReader();
- reader.seek(new FileSystemWAL.WalPointer(0, 0));
+ FileSystemWAL.FileSystemWALReader reader = testMeta.fsWAL.getReader();
+ assertNumTuplesRead(reader, numRecords);
- int read = 0;
- while (reader.advance()) {
- Object o = reader.get();
- read++;
- }
+ /* skip first file for reading, and then read other entries till the end */
+ reader.seek(new FileSystemWAL.FileSystemWALPointer(1, 0));
+ assertNumTuplesRead(reader, numRecords - 32);
- Assert.assertEquals("Number of records read from start ", wrote, read);
+ /* skip first file and few records from the next file, and read till the end */
+ reader.seek(new FileSystemWAL.FileSystemWALPointer(1, 16 * 1024));
+ assertNumTuplesRead(reader, numRecords - (32 + 16));
- /* skip first file for reading, and then read other entries till the end */
- reader.seek(new FileSystemWAL.WalPointer(1, 0));
+ testMeta.fsWAL.teardown();
+ }
- read = 0;
- while (reader.advance()) {
- Object o = reader.get();
- read++;
- }
- Assert.assertEquals("Number of record read after skipping one file ", wrote - 32, read);
+ @Test
+ public void testWalRollingWithPartialFiles() throws IOException
+ {
+ testMeta.fsWAL.setMaxLength(1024);
+ testMeta.fsWAL.setup();
- /* skip first file and few records from the next file, and read till the end */
- reader.seek(new FileSystemWAL.WalPointer(1, 16 * 1024));
- read = 0;
- while (reader.advance()) {
- Object o = reader.get();
- read++;
- }
- Assert.assertEquals("Number of record read after skipping one file ", wrote - (32 + 16), read);
+ FileSystemWAL.FileSystemWALWriter fsWALWriter = testMeta.fsWAL.getWriter();
+
+ Slice first = getRandomSlice(1000);
+ fsWALWriter.append(first);
+
+ Slice second = getRandomSlice(1000);
+ fsWALWriter.append(second);
+ testMeta.fsWAL.beforeCheckpoint(0);
+
+ /** Read from start till the end */
+ FileSystemWAL.FileSystemWALReader reader = testMeta.fsWAL.getReader();
+ assertNumTuplesRead(reader, 2);
+
+ testMeta.fsWAL.teardown();
}
- static class ByteArraySerde implements WAL.Serde<byte[]>
+
+ @Test
+ public void testFinalizeAfterDelay() throws IOException
{
- @Override
- public byte[] toBytes(byte[] tuple)
- {
- return tuple;
- }
+ testMeta.fsWAL.setMaxLength(32 * 1024);
+ testMeta.fsWAL.setup();
- @Override
- public byte[] fromBytes(byte[] data)
- {
- return data;
- }
+ FileSystemWAL.FileSystemWALWriter fsWALWriter = testMeta.fsWAL.getWriter();
+
+ // write 32 records of size 1K
+ write1KRecords(fsWALWriter, 32);
+ testMeta.fsWAL.beforeCheckpoint(0);
+
+ write1KRecords(fsWALWriter, 32);
+ testMeta.fsWAL.beforeCheckpoint(1);
+
+ write1KRecords(fsWALWriter, 32);
+ testMeta.fsWAL.beforeCheckpoint(2);
+
+ testMeta.fsWAL.committed(2);
+
+ /** Read from start till the end */
+ FileSystemWAL.FileSystemWALReader reader = testMeta.fsWAL.getReader();
+ assertNumTuplesRead(reader, 32 * 3);
}
- /**
- * File storage for testing.
- */
- public static class MockFileAccess extends FileAccessFSImpl
+ @Test
+ public void testRecovery() throws IOException
{
- private final Set<String> deletedFiles = Sets.newHashSet();
+ testMeta.fsWAL.setMaxLength(2 * 1024);
+ testMeta.fsWAL.setup();
- public void disableChecksum()
- {
- fs.setVerifyChecksum(false);
- }
+ FileSystemWAL.FileSystemWALWriter fsWALWriter = testMeta.fsWAL.getWriter();
- @Override
- public void delete(long bucketKey, String fileName) throws IOException
- {
- super.delete(bucketKey, fileName);
- deletedFiles.add("" + bucketKey + fileName);
- }
+ /* Write each record of size 1K, each complete file will have 2 records.
+ * If we write 3 records the first tmp file will get rotated with 2 records but the next one will just have 1
+ * entry.
+ */
+ write1KRecords(fsWALWriter, 3);
- @Override
- public FileReader getReader(final long bucketKey, final String fileName) throws IOException
- {
- return null;
+ testMeta.fsWAL.beforeCheckpoint(0);
+ testMeta.fsWAL.committed(0);
+
+ FileSystemWAL.FileSystemWALReader fsWALReader = testMeta.fsWAL.getReader();
+ assertNumTuplesRead(fsWALReader, 3);
+
+ // a failure occurred
+ fsWALWriter.close();
+ testMeta.fsWAL.setup();
+ write1KRecords(fsWALWriter, 1); //2nd part is completed and rotated
+ testMeta.fsWAL.beforeCheckpoint(1);
+ testMeta.fsWAL.committed(1);
+
+ fsWALReader.seek(new FileSystemWAL.FileSystemWALPointer(1, 0));
+ assertNumTuplesRead(fsWALReader, 2);
+
+ testMeta.fsWAL.teardown();
+ }
+
+ @Test
+ public void testDeleteOfTmpFiles() throws IOException
+ {
+ testMeta.fsWAL.setMaxLength(2 * 1024);
+ testMeta.fsWAL.setup();
+
+ FileSystemWAL.FileSystemWALWriter fsWALWriter = testMeta.fsWAL.getWriter();
+
+ write1KRecords(fsWALWriter, 2);
+ testMeta.fsWAL.beforeCheckpoint(0);
+ FileSystemWAL fsWalCheckpointed = KryoCloneUtils.cloneObject(testMeta.fsWAL);
+
+ write1KRecords(fsWALWriter, 2);
+ testMeta.fsWAL.beforeCheckpoint(1);
+
+ // a failure occurred
+ fsWALWriter.close();
+
+ fsWalCheckpointed.setup();
+ fsWALWriter = fsWalCheckpointed.getWriter();
+ write1KRecords(fsWALWriter, 2);
+ fsWalCheckpointed.beforeCheckpoint(1);
+ fsWalCheckpointed.committed(1);
+
+ FileSystemWAL.FileSystemWALReader fsWALReader = fsWalCheckpointed.getReader();
+ assertNumTuplesRead(fsWALReader, 4);
+
+ fsWalCheckpointed.teardown();
+ }
+
+ @Test
+ public void testReadWithInterceptingFinalize() throws IOException
+ {
+ testMeta.fsWAL.setMaxLength(2 * 1024);
+ testMeta.fsWAL.setup();
+
+ FileSystemWAL.FileSystemWALWriter fsWALWriter = testMeta.fsWAL.getWriter();
+
+ write1KRecords(fsWALWriter, 1);
+ testMeta.fsWAL.beforeCheckpoint(0);
+
+ FileSystemWAL.FileSystemWALReader fsWALReader = testMeta.fsWAL.getReader();
+ Assert.assertNotNull("one entry", fsWALReader.next());
+
+ write1KRecords(fsWALWriter, 1);
+ testMeta.fsWAL.beforeCheckpoint(1);
+ testMeta.fsWAL.committed(1);
+ assertNumTuplesRead(fsWALReader, 1);
+
+ testMeta.fsWAL.teardown();
+ }
+
+ @Test
+ public void testDeleteEverything() throws IOException
+ {
+ testMeta.fsWAL.setMaxLength(2 * 1024);
+ testMeta.fsWAL.setup();
+
+ FileSystemWAL.FileSystemWALWriter fsWALWriter = testMeta.fsWAL.getWriter();
+
+ write1KRecords(fsWALWriter, 10);
+ testMeta.fsWAL.beforeCheckpoint(0);
+ testMeta.fsWAL.committed(0);
+
+ FileSystemWAL.FileSystemWALReader fsWALReader = testMeta.fsWAL.getReader();
+ assertNumTuplesRead(fsWALReader, 10);
+
+ FileSystemWAL.FileSystemWALPointer writerPointer = fsWALWriter.getPointer();
+ fsWALWriter.delete(writerPointer);
+
+ FileContext fileContext = FileContextUtils.getFileContext(testMeta.fsWAL.getFilePath());
+ for (int i = 0; i < 5; i++) {
+ Assert.assertTrue("part exists " + i,
+ !fileContext.util().exists(new Path(testMeta.fsWAL.getPartFilePath(i))));
}
- @Override
- public FileWriter getWriter(final long bucketKey, final String fileName) throws IOException
- {
- return null;
+ testMeta.fsWAL.teardown();
+ }
+
+ @Test
+ public void testDeleteFullParts() throws IOException
+ {
+ testMeta.fsWAL.setMaxLength(2 * 1024);
+ testMeta.fsWAL.setup();
+
+ FileSystemWAL.FileSystemWALWriter fsWALWriter = testMeta.fsWAL.getWriter();
+
+ write1KRecords(fsWALWriter, 10);
+ testMeta.fsWAL.beforeCheckpoint(0);
+ testMeta.fsWAL.committed(0);
+
+ FileSystemWAL.FileSystemWALReader fsWALReader = testMeta.fsWAL.getReader();
+ assertNumTuplesRead(fsWALReader, 10);
+
+ fsWALWriter.delete(new FileSystemWAL.FileSystemWALPointer(3, 0));
+ fsWALReader.seek(fsWALReader.getStartPointer());
+
+ assertNumTuplesRead(fsWALReader, 4);
+ }
+
+ @Test
+ public void testDeletePartialParts() throws IOException
+ {
+ testMeta.fsWAL.setMaxLength(2 * 1024);
+ testMeta.fsWAL.setup();
+
+ FileSystemWAL.FileSystemWALWriter fsWALWriter = testMeta.fsWAL.getWriter();
+
+ write1KRecords(fsWALWriter, 4);
+ testMeta.fsWAL.beforeCheckpoint(0);
+ testMeta.fsWAL.committed(0);
+
+ FileSystemWAL.FileSystemWALReader fsWALReader = testMeta.fsWAL.getReader();
+ assertNumTuplesRead(fsWALReader, 4);
+
+ fsWALWriter.delete(new FileSystemWAL.FileSystemWALPointer(1, 1024));
+
+ fsWALReader.seek(fsWALReader.getStartPointer());
+ assertNumTuplesRead(fsWALReader, 1);
+ }
+
+ @Test
+ public void testFinalizeWithDelete() throws IOException
+ {
+ testMeta.fsWAL.setMaxLength(2 * 1024);
+ testMeta.fsWAL.setup();
+
+ FileSystemWAL.FileSystemWALWriter fsWALWriter = testMeta.fsWAL.getWriter();
+
+ write1KRecords(fsWALWriter, 2);
+ testMeta.fsWAL.beforeCheckpoint(0);
+
+ write1KRecords(fsWALWriter, 2);
+ testMeta.fsWAL.beforeCheckpoint(1);
+
+ write1KRecords(fsWALWriter, 2);
+ testMeta.fsWAL.beforeCheckpoint(2);
+
+ FileSystemWAL.FileSystemWALReader fsWALReader = testMeta.fsWAL.getReader();
+ assertNumTuplesRead(fsWALReader, 6);
+
+ testMeta.fsWAL.committed(0);
+
+ fsWALWriter.delete(new FileSystemWAL.FileSystemWALPointer(2, 0));
+
+ FileContext fileContext = FileContextUtils.getFileContext(testMeta.fsWAL.getFilePath());
+ Assert.assertTrue("part 0 exists ", !fileContext.util().exists(new Path(testMeta.fsWAL.getPartFilePath(0))));
+
+ testMeta.fsWAL.committed(1);
+ Assert.assertTrue("part 1 exists ", !fileContext.util().exists(new Path(testMeta.fsWAL.getPartFilePath(1))));
+
+ fsWALReader.seek(fsWALReader.getStartPointer());
+ assertNumTuplesRead(fsWALReader, 2);
+ }
+
+ @Test
+ public void testPointerComparisons()
+ {
+ FileSystemWAL.FileSystemWALPointer pointer1 = new FileSystemWAL.FileSystemWALPointer(0, 10);
+ FileSystemWAL.FileSystemWALPointer pointer2 = new FileSystemWAL.FileSystemWALPointer(0, 10);
+
+ Assert.assertTrue("equal", pointer1.compareTo(pointer2) == 0);
+
+ pointer1 = new FileSystemWAL.FileSystemWALPointer(0, 11);
+ Assert.assertTrue("offset greater", pointer1.compareTo(pointer2) == 1);
+
+ pointer1 = new FileSystemWAL.FileSystemWALPointer(0, 3);
+ Assert.assertTrue("offset smaller", pointer1.compareTo(pointer2) == -1);
+
+ pointer1 = new FileSystemWAL.FileSystemWALPointer(1, 10);
+ Assert.assertTrue("part greater", pointer1.compareTo(pointer2) == 1);
+
+ pointer1 = new FileSystemWAL.FileSystemWALPointer(0, 10);
+ pointer2 = new FileSystemWAL.FileSystemWALPointer(3, 10);
+ Assert.assertTrue("part smaller", pointer1.compareTo(pointer2) == -1);
+ }
+
+ private static void assertNumTuplesRead(FileSystemWAL.FileSystemWALReader reader, int expectedTuples)
+ throws IOException
+ {
+ int tuples = 0;
+ while (reader.next() != null) {
+ tuples++;
}
+ Assert.assertEquals("num tuples", expectedTuples, tuples);
+ }
+ private static void write1KRecords(FileSystemWAL.FileSystemWALWriter writer, int numRecords)
+ throws IOException
+ {
+ for (int i = 0; i < numRecords; i++) {
+ writer.append(getRandomSlice(1020));
+ }
}
+ private static final Logger LOG = LoggerFactory.getLogger(FileSystemWALTest.class);
+
}