2.23

filer: leveldb and hbase may miss files when listing large directories more than 1024
fix https://github.com/chrislusf/seaweedfs/issues/1768
2026-10-01 20:26:27 +00:00 · 2021-01-31 20:18:06 -08:00 · 2021-01-31 20:11:44 -08:00 · 2021-01-31 18:26:26 -08:00 · 2021-01-30 13:38:44 -08:00 · 2021-01-30 06:18:41 -08:00
76 changed files with 1259 additions and 985 deletions
@@ -9,6 +9,7 @@ assignees: ''

 Sponsors SeaweedFS via Patreon https://www.patreon.com/seaweedfs
 Report issues here. Ask questions here https://stackoverflow.com/questions/tagged/seaweedfs
+Please ask questions in https://github.com/chrislusf/seaweedfs/discussions

 example of a good issue report:
 https://github.com/chrislusf/seaweedfs/issues/1005
@@ -1,11 +1,14 @@
 # SeaweedFS


+[![Slack](https://img.shields.io/badge/slack-purple)](https://join.slack.com/t/seaweedfs/shared_invite/enQtMzI4MTMwMjU2MzA3LTEyYzZmZWYzOGQ3MDJlZWMzYmI0OTE4OTJiZjJjODBmMzUxNmYwODg0YjY3MTNlMjBmZDQ1NzQ5NDJhZWI2ZmY)
+[![Twitter](https://img.shields.io/twitter/follow/seaweedfs.svg?style=social&label=Follow)](https://twitter.com/intent/follow?screen_name=seaweedfs)
 [![Build Status](https://travis-ci.org/chrislusf/seaweedfs.svg?branch=master)](https://travis-ci.org/chrislusf/seaweedfs)
 [![GoDoc](https://godoc.org/github.com/chrislusf/seaweedfs/weed?status.svg)](https://godoc.org/github.com/chrislusf/seaweedfs/weed)
 [![Wiki](https://img.shields.io/badge/docs-wiki-blue.svg)](https://github.com/chrislusf/seaweedfs/wiki)
 [![Docker Pulls](https://img.shields.io/docker/pulls/chrislusf/seaweedfs.svg?maxAge=4800)](https://hub.docker.com/r/chrislusf/seaweedfs/)

+
 ![SeaweedFS Logo](https://raw.githubusercontent.com/chrislusf/seaweedfs/master/note/seaweedfs.png)

 <h2 align="center"><a href="https://www.patreon.com/seaweedfs">Sponsor SeaweedFS via Patreon</a></h2>
@@ -12,14 +12,26 @@ build:
 dev: build
 	docker-compose -f local-dev-compose.yml -p seaweedfs up

+dev_mount: build
+	docker-compose -f local-mount-compose.yml -p seaweedfs up
+
 k8s: build
 	docker-compose -f local-k8s-compose.yml -p seaweedfs up

 dev_registry: build
 	docker-compose -f local-registry-compose.yml -p seaweedfs up

+dev_replicate: build
+	docker-compose -f local-replicate-compose.yml -p seaweedfs up
+
 cluster: build
 	docker-compose -f local-cluster-compose.yml -p seaweedfs up

+2clusters: build
+	docker-compose -f local-clusters-compose.yml -p seaweedfs up
+
+filer_etcd: build
+	docker stack deploy -c swarm-etcd.yml fs
+
 clean:
 	rm ./weed
@@ -0,0 +1,23 @@
+version: '2'
+
+services:
+  server1:
+    image: chrislusf/seaweedfs:local
+    ports:
+      - 9333:9333
+      - 19333:19333
+      - 8084:8080
+      - 18084:18080
+      - 8888:8888
+      - 18888:18888
+    command: "server -ip=server1 -filer -volume.max=0 -master.volumeSizeLimitMB=1024 -volume.preStopSeconds=1"
+  server2:
+    image: chrislusf/seaweedfs:local
+    ports:
+      - 9334:9333
+      - 19334:19333
+      - 8085:8080
+      - 18085:18080
+      - 8889:8888
+      - 18889:18888
+    command: "server -ip=server2 -filer -volume.max=0 -master.volumeSizeLimitMB=1024 -volume.preStopSeconds=1"
@@ -0,0 +1,46 @@
+version: '2'
+
+services:
+  master:
+    image: chrislusf/seaweedfs:local
+    ports:
+      - 9333:9333
+      - 19333:19333
+    command: "master -ip=master"
+  volume:
+    image: chrislusf/seaweedfs:local
+    ports:
+      - 7455:8080
+      - 9325:9325
+    command: 'volume -mserver="master:9333" -port=8080 -metricsPort=9325 -preStopSeconds=1 -publicUrl=localhost:7455'
+    depends_on:
+      - master
+  filer:
+    image: chrislusf/seaweedfs:local
+    ports:
+      - 8888:8888
+      - 18888:18888
+      - 9326:9326
+    command: 'filer -master="master:9333"  -metricsPort=9326'
+    tty: true
+    stdin_open: true
+    depends_on:
+      - master
+      - volume
+  mount_1:
+    image: chrislusf/seaweedfs:local
+    privileged: true
+    entrypoint: '/bin/sh -c "mkdir -p t1 && mkdir -p cache/t1 && weed -v=4 mount -filer=filer:8888 -cacheDir=./cache/t1 -dir=./t1 -filer.path=/c1" -volumeServerAccess=filerProxy'
+    depends_on:
+      - master
+      - volume
+      - filer
+  mount_2:
+    image: chrislusf/seaweedfs:local
+    privileged: true
+    entrypoint: '/bin/sh -c "mkdir -p t2 && mkdir -p cache/t2 && weed -v=4  mount -filer=filer:8888 -cacheDir=./cache/t2 -dir=./t2 -filer.path=/c1" -volumeServerAcess=publicUrl'
+    depends_on:
+      - master
+      - volume
+      - filer
+      - mount_1
@@ -0,0 +1,59 @@
+version: '2'
+
+services:
+  master:
+    image: chrislusf/seaweedfs:local
+    ports:
+      - 9333:9333
+      - 19333:19333
+    command: "master -ip=master"
+  volume:
+    image: chrislusf/seaweedfs:local
+    ports:
+      - 8080:8080
+      - 18080:18080
+    command: "volume -mserver=master:9333 -port=8080 -ip=volume -preStopSeconds=1"
+    depends_on:
+      - master
+  filer:
+    image: chrislusf/seaweedfs:local
+    ports:
+      - 8888:8888
+      - 18888:18888
+    command: '-v=9 filer -master="master:9333"'
+    restart: on-failure
+    volumes:
+      - ./notification.toml:/etc/seaweedfs/notification.toml
+    depends_on:
+      - master
+      - volume
+      - rabbitmq
+      - replicate
+    environment:
+      RABBIT_SERVER_URL: "amqp://guest:guest@rabbitmq:5672/"
+  replicate:
+    image: chrislusf/seaweedfs:local
+    command: '-v=9 filer.replicate'
+    restart: on-failure
+    volumes:
+      - ./notification.toml:/etc/seaweedfs/notification.toml
+      - ./replication.toml:/etc/seaweedfs/replication.toml
+    depends_on:
+      - rabbitmq
+    environment:
+      RABBIT_SERVER_URL: "amqp://guest:guest@rabbitmq:5672/"
+  s3:
+    image: chrislusf/seaweedfs:local
+    ports:
+      - 8333:8333
+    command: 's3 -filer="filer:8888"'
+    depends_on:
+      - master
+      - volume
+      - filer
+  rabbitmq:
+    image: rabbitmq:3.8.10-management-alpine
+    ports:
+      - 5672:5672
+      - 15671:15671
+      - 15672:15672
@@ -0,0 +1,17 @@
+[notification.log]
+# this is only for debugging perpose and does not work with "weed filer.replicate"
+enabled = false
+
+
+[notification.gocdk_pub_sub]
+# The Go Cloud Development Kit (https://gocloud.dev).
+# PubSub API (https://godoc.org/gocloud.dev/pubsub).
+# Supports AWS SNS/SQS, Azure Service Bus, Google PubSub, NATS and RabbitMQ.
+enabled = true
+# This URL will Dial the RabbitMQ server at the URL in the environment
+# variable RABBIT_SERVER_URL and open the exchange "myexchange".
+# The exchange must have already been created by some other means, like
+# the RabbitMQ management plugin. Сreate myexchange of type fanout and myqueue then
+# create binding myexchange => myqueue
+topic_url = "rabbit://swexchange"
+sub_url = "rabbit://swqueue"
@@ -0,0 +1,11 @@
+[source.filer]
+enabled = true
+grpcAddress = "filer:18888"
+# all files under this directory tree are replicated.
+# this is not a directory on your hard drive, but on your filer.
+# i.e., all files with this "prefix" are sent to notification message queue.
+directory = "/buckets"
+
+[sink.local_incremental]
+enabled = true
+directory = "/data"
@@ -0,0 +1,84 @@
+# 2021-01-30 16:25:30
+version: '3.8'
+
+services:
+
+  etcd:
+    image: gasparekatapy/etcd
+    networks:
+      - net
+    deploy:
+      mode: replicated
+      replicas: 3
+
+  master:
+    image: chrislusf/seaweedfs:local
+    environment:
+      WEED_MASTER_FILER_DEFAULT: "filer:8888"
+      WEED_MASTER_SEQUENCER_TYPE: "raft"
+    ports:
+      - "9333:9333"
+      - "19333:19333"
+    networks:
+      - net
+    command:
+      - 'master'
+      - '-resumeState=true'
+      - '-ip=master'
+      - '-port=9333'
+    deploy:
+      mode: replicated
+      replicas: 1
+
+  filer:
+    image: chrislusf/seaweedfs:local
+    environment:
+      WEED_LEVELDB2_ENABLED: "false"
+      WEED_ETCD_ENABLED: "true"
+      WEED_ETCD_SERVERS: "etcd:2379"
+    ports:
+      - target: 8888
+        published: 8888
+        protocol: tcp
+        mode: host
+      - target: 18888
+        published: 18888
+        protocol: tcp
+        mode: host
+    networks:
+      - net
+    command:
+      - 'filer'
+      - '-ip=filer'
+      - '-port=8888'
+      - '-port.readonly=28888'
+      - '-master=master:9333'
+      - '-disableDirListing=true'
+    deploy:
+      mode: replicated
+      replicas: 1
+
+  volume:
+    image: chrislusf/seaweedfs:local
+    ports:
+      - target: 8080
+        published: 8080
+        protocol: tcp
+        mode: host
+      - target: 18080
+        published: 18080
+        protocol: tcp
+        mode: host
+    networks:
+      - net
+    command:
+      - 'volume'
+      - '-mserver=master:9333'
+      - '-port=8080'
+    deploy:
+      mode: global
+
+  ###########################################################################
+
+networks:
+  net:
@@ -56,7 +56,7 @@ require (
 	github.com/prometheus/client_golang v1.3.0
 	github.com/rakyll/statik v0.1.7
 	github.com/rcrowley/go-metrics v0.0.0-20190826022208-cac0b30c2563 // indirect
-	github.com/seaweedfs/fuse v1.0.9
+	github.com/seaweedfs/fuse v1.1.1
 	github.com/seaweedfs/goexif v1.0.2
 	github.com/skip2/go-qrcode v0.0.0-20200617195104-da1b6568686e
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
@@ -577,6 +577,10 @@ github.com/seaweedfs/fuse v1.0.8 h1:HBPJTC77OlxwSd2JiTwvLPn8bWTElqQp3xs9vf3C15s=
 github.com/seaweedfs/fuse v1.0.8/go.mod h1:W7ubwr1l7KQsMeUpxFFOFOSxUL/ucTRMAlVYs4xdfQ8=
 github.com/seaweedfs/fuse v1.0.9 h1:3JZoGsW7cmmrd/U5KYcIGR2+EqyBvCYCoCpEdZAz/Dc=
 github.com/seaweedfs/fuse v1.0.9/go.mod h1:+PP6WlkrRUG6KPE+Th2EX5To/PjHaFsvqg/UgQ39aj8=
+github.com/seaweedfs/fuse v1.1.0 h1:cL1qPHFNtFv0UuJTLjKKgWDzfJ4iZzTa4Y7ipc2acGw=
+github.com/seaweedfs/fuse v1.1.0/go.mod h1:+PP6WlkrRUG6KPE+Th2EX5To/PjHaFsvqg/UgQ39aj8=
+github.com/seaweedfs/fuse v1.1.1 h1:WD51YFJcBViOx8I89jeqPD+vAKl4EowzBy9GUw0plb0=
+github.com/seaweedfs/fuse v1.1.1/go.mod h1:+PP6WlkrRUG6KPE+Th2EX5To/PjHaFsvqg/UgQ39aj8=
 github.com/seaweedfs/goexif v1.0.2 h1:p+rTXYdQ2mgxd+1JaTrQ9N8DvYuw9UH9xgYmJ+Bb29E=
 github.com/seaweedfs/goexif v1.0.2/go.mod h1:MrKs5LK0HXdffrdCZrW3OIMegL2xXpC6ThLyXMyjdrk=
 github.com/secsy/goftp v0.0.0-20190720192957-f31499d7c79a h1:C6IhVTxNkhlb0tlCB6JfHOUv1f0xHPK7V8X4HlJZEJw=
@@ -1,5 +1,5 @@
 apiVersion: v1
 description: SeaweedFS
 name: seaweedfs
-appVersion: "2.21"
-version: 2.21
+appVersion: "2.23"
+version: 2.23
@@ -4,7 +4,7 @@ global:
  registry: ""
  repository: ""
  imageName: chrislusf/seaweedfs
-  # imageTag: "2.21" - started using {.Chart.appVersion}
+  # imageTag: "2.23" - started using {.Chart.appVersion}
  imagePullPolicy: IfNotPresent
  imagePullSecrets: imagepullsecret
  restartPolicy: Always
@@ -5,7 +5,7 @@

    <groupId>com.github.chrislusf</groupId>
    <artifactId>seaweedfs-client</artifactId>
-    <version>1.5.6</version>
+    <version>1.5.7</version>

    <parent>
        <groupId>org.sonatype.oss</groupId>
@@ -5,7 +5,7 @@

    <groupId>com.github.chrislusf</groupId>
    <artifactId>seaweedfs-client</artifactId>
-    <version>1.5.6</version>
+    <version>1.5.7</version>

    <parent>
        <groupId>org.sonatype.oss</groupId>
@@ -5,7 +5,7 @@

    <groupId>com.github.chrislusf</groupId>
    <artifactId>seaweedfs-client</artifactId>
-    <version>1.5.6</version>
+    <version>1.5.7</version>

    <parent>
        <groupId>org.sonatype.oss</groupId>
@@ -1,28 +1,22 @@
-package seaweed.hdfs;
+package seaweedfs.client;

 // based on org.apache.hadoop.fs.azurebfs.services.AbfsInputStream

-import org.apache.hadoop.fs.ByteBufferReadable;
-import org.apache.hadoop.fs.FSExceptionMessages;
-import org.apache.hadoop.fs.FSInputStream;
-import org.apache.hadoop.fs.FileSystem.Statistics;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import seaweedfs.client.FilerGrpcClient;
-import seaweedfs.client.FilerProto;
-import seaweedfs.client.SeaweedRead;

 import java.io.EOFException;
 import java.io.IOException;
+import java.io.InputStream;
 import java.nio.ByteBuffer;
 import java.util.List;

-public class SeaweedInputStream extends FSInputStream implements ByteBufferReadable {
+public class SeaweedInputStream extends InputStream {

    private static final Logger LOG = LoggerFactory.getLogger(SeaweedInputStream.class);
+    private static final IOException EXCEPTION_STREAM_IS_CLOSED = new IOException("Stream is closed!");

    private final FilerGrpcClient filerGrpcClient;
-    private final Statistics statistics;
    private final String path;
    private final FilerProto.Entry entry;
    private final List<SeaweedRead.VisibleInterval> visibleIntervalList;
@@ -34,11 +28,9 @@ public class SeaweedInputStream extends FSInputStream implements ByteBufferReada

    public SeaweedInputStream(
            final FilerGrpcClient filerGrpcClient,
-            final Statistics statistics,
            final String path,
            final FilerProto.Entry entry) throws IOException {
        this.filerGrpcClient = filerGrpcClient;
-        this.statistics = statistics;
        this.path = path;
        this.entry = entry;
        this.contentLength = SeaweedRead.fileSize(entry);
@@ -86,7 +78,6 @@ public class SeaweedInputStream extends FSInputStream implements ByteBufferReada
    }

    // implement ByteBufferReadable
-    @Override
    public synchronized int read(ByteBuffer buf) throws IOException {

        if (position < 0) {
@@ -111,45 +102,32 @@ public class SeaweedInputStream extends FSInputStream implements ByteBufferReada

        if (bytesRead > 0) {
            this.position += bytesRead;
-            if (statistics != null) {
-                statistics.incrementBytesRead(bytesRead);
-            }
        }

        return (int) bytesRead;
    }

-    /**
-     * Seek to given position in stream.
-     *
-     * @param n position to seek to
-     * @throws IOException  if there is an error
-     * @throws EOFException if attempting to seek past end of file
-     */
-    @Override
    public synchronized void seek(long n) throws IOException {
        if (closed) {
-            throw new IOException(FSExceptionMessages.STREAM_IS_CLOSED);
+            throw EXCEPTION_STREAM_IS_CLOSED;
        }
        if (n < 0) {
-            throw new EOFException(FSExceptionMessages.NEGATIVE_SEEK);
+            throw new EOFException("Cannot seek to a negative offset");
        }
        if (n > contentLength) {
-            throw new EOFException(FSExceptionMessages.CANNOT_SEEK_PAST_EOF);
+            throw new EOFException("Attempted to seek or read past the end of the file");
        }
-
        this.position = n;
-
    }

    @Override
    public synchronized long skip(long n) throws IOException {
        if (closed) {
-            throw new IOException(FSExceptionMessages.STREAM_IS_CLOSED);
+            throw EXCEPTION_STREAM_IS_CLOSED;
        }
        if (this.position == contentLength) {
            if (n > 0) {
-                throw new EOFException(FSExceptionMessages.CANNOT_SEEK_PAST_EOF);
+                throw new EOFException("Attempted to seek or read past the end of the file");
            }
        }
        long newPos = this.position + n;
@@ -177,10 +155,9 @@ public class SeaweedInputStream extends FSInputStream implements ByteBufferReada
    @Override
    public synchronized int available() throws IOException {
        if (closed) {
-            throw new IOException(
-                    FSExceptionMessages.STREAM_IS_CLOSED);
+            throw EXCEPTION_STREAM_IS_CLOSED;
        }
-        final long remaining = this.contentLength - this.getPos();
+        final long remaining = this.contentLength - this.position;
        return remaining <= Integer.MAX_VALUE
                ? (int) remaining : Integer.MAX_VALUE;
    }
@@ -195,65 +172,21 @@ public class SeaweedInputStream extends FSInputStream implements ByteBufferReada
     */
    public long length() throws IOException {
        if (closed) {
-            throw new IOException(FSExceptionMessages.STREAM_IS_CLOSED);
+            throw EXCEPTION_STREAM_IS_CLOSED;
        }
        return contentLength;
    }

-    /**
-     * Return the current offset from the start of the file
-     *
-     * @throws IOException throws {@link IOException} if there is an error
-     */
-    @Override
    public synchronized long getPos() throws IOException {
        if (closed) {
-            throw new IOException(FSExceptionMessages.STREAM_IS_CLOSED);
+            throw EXCEPTION_STREAM_IS_CLOSED;
        }
        return position;
    }

-    /**
-     * Seeks a different copy of the data.  Returns true if
-     * found a new source, false otherwise.
-     *
-     * @throws IOException throws {@link IOException} if there is an error
-     */
-    @Override
-    public boolean seekToNewSource(long l) throws IOException {
-        return false;
-    }
-
    @Override
    public synchronized void close() throws IOException {
        closed = true;
    }

-    /**
-     * Not supported by this stream. Throws {@link UnsupportedOperationException}
-     *
-     * @param readlimit ignored
-     */
-    @Override
-    public synchronized void mark(int readlimit) {
-        throw new UnsupportedOperationException("mark()/reset() not supported on this stream");
-    }
-
-    /**
-     * Not supported by this stream. Throws {@link UnsupportedOperationException}
-     */
-    @Override
-    public synchronized void reset() throws IOException {
-        throw new UnsupportedOperationException("mark()/reset() not supported on this stream");
-    }
-
-    /**
-     * gets whether mark and reset are supported by {@code ADLFileInputStream}. Always returns false.
-     *
-     * @return always {@code false}
-     */
-    @Override
-    public boolean markSupported() {
-        return false;
-    }
 }
@@ -1,16 +1,9 @@
-package seaweed.hdfs;
+package seaweedfs.client;

 // adapted from org.apache.hadoop.fs.azurebfs.services.AbfsOutputStream

-import com.google.common.base.Preconditions;
-import org.apache.hadoop.fs.FSExceptionMessages;
-import org.apache.hadoop.fs.Path;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import seaweedfs.client.ByteBufferPool;
-import seaweedfs.client.FilerGrpcClient;
-import seaweedfs.client.FilerProto;
-import seaweedfs.client.SeaweedWrite;

 import java.io.IOException;
 import java.io.InterruptedIOException;
@@ -18,20 +11,18 @@ import java.io.OutputStream;
 import java.nio.ByteBuffer;
 import java.util.concurrent.*;

-import static seaweed.hdfs.SeaweedFileSystemStore.getParentDirectory;
-
 public class SeaweedOutputStream extends OutputStream {

    private static final Logger LOG = LoggerFactory.getLogger(SeaweedOutputStream.class);

    private final FilerGrpcClient filerGrpcClient;
-    private final Path path;
+    private final String path;
    private final int bufferSize;
    private final int maxConcurrentRequestCount;
    private final ThreadPoolExecutor threadExecutor;
    private final ExecutorCompletionService<Void> completionService;
    private final FilerProto.Entry.Builder entry;
-    private final boolean supportFlush = false; // true;
+    protected final boolean supportFlush = false; // true;
    private final ConcurrentLinkedDeque<WriteOperation> writeOperations;
    private long position;
    private boolean closed;
@@ -42,7 +33,7 @@ public class SeaweedOutputStream extends OutputStream {
    private long outputIndex;
    private String replication = "000";

-    public SeaweedOutputStream(FilerGrpcClient filerGrpcClient, final Path path, FilerProto.Entry.Builder entry,
+    public SeaweedOutputStream(FilerGrpcClient filerGrpcClient, final String path, FilerProto.Entry.Builder entry,
                               final long position, final int bufferSize, final String replication) {
        this.filerGrpcClient = filerGrpcClient;
        this.replication = replication;
@@ -70,6 +61,14 @@ public class SeaweedOutputStream extends OutputStream {

    }

+    public static String getParentDirectory(String path) {
+        if (path.equals("/")) {
+            return path;
+        }
+        int lastSlashIndex = path.lastIndexOf("/");
+        return path.substring(0, lastSlashIndex);
+    }
+
    private synchronized void flushWrittenBytesToServiceInternal(final long offset) throws IOException {
        try {
            SeaweedWrite.writeMeta(filerGrpcClient, getParentDirectory(path), entry);
@@ -89,7 +88,9 @@ public class SeaweedOutputStream extends OutputStream {
            throws IOException {
        maybeThrowLastError();

-        Preconditions.checkArgument(data != null, "null data");
+        if (data == null) {
+            return;
+        }

        if (off < 0 || length < 0 || length > data.length - off) {
            throw new IndexOutOfBoundsException();
@@ -152,7 +153,7 @@ public class SeaweedOutputStream extends OutputStream {
            flushInternal();
            threadExecutor.shutdown();
        } finally {
-            lastError = new IOException(FSExceptionMessages.STREAM_IS_CLOSED);
+            lastError = new IOException("Stream is closed!");
            ByteBufferPool.release(buffer);
            buffer = null;
            outputIndex = 0;
@@ -185,7 +186,7 @@ public class SeaweedOutputStream extends OutputStream {
        }
        final Future<Void> job = completionService.submit(() -> {
            // System.out.println(path + " is going to save [" + (writePosition) + "," + ((writePosition) + bytesLength) + ")");
-            SeaweedWrite.writeData(entry, replication, filerGrpcClient, writePosition, bufferToWrite.array(), bufferToWrite.position(), bufferToWrite.limit(), path.toUri().getPath());
+            SeaweedWrite.writeData(entry, replication, filerGrpcClient, writePosition, bufferToWrite.array(), bufferToWrite.position(), bufferToWrite.limit(), path);
            // System.out.println(path + " saved [" + (writePosition) + "," + ((writePosition) + bytesLength) + ")");
            ByteBufferPool.release(bufferToWrite);
            return null;
@@ -239,13 +240,13 @@ public class SeaweedOutputStream extends OutputStream {
        }
    }

-    private synchronized void flushInternal() throws IOException {
+    protected synchronized void flushInternal() throws IOException {
        maybeThrowLastError();
        writeCurrentBufferToService();
        flushWrittenBytesToService();
    }

-    private synchronized void flushInternalAsync() throws IOException {
+    protected synchronized void flushInternalAsync() throws IOException {
        maybeThrowLastError();
        writeCurrentBufferToService();
        flushWrittenBytesToServiceAsync();
@@ -278,10 +279,6 @@ public class SeaweedOutputStream extends OutputStream {
        private final long length;

        WriteOperation(final Future<Void> task, final long startOffset, final long length) {
-            Preconditions.checkNotNull(task, "task");
-            Preconditions.checkArgument(startOffset >= 0, "startOffset");
-            Preconditions.checkArgument(length >= 0, "length");
-
            this.task = task;
            this.startOffset = startOffset;
            this.length = length;
@@ -11,13 +11,13 @@
        <dependency>
            <groupId>com.github.chrislusf</groupId>
            <artifactId>seaweedfs-client</artifactId>
-            <version>1.5.6</version>
+            <version>1.5.7</version>
            <scope>compile</scope>
        </dependency>
        <dependency>
            <groupId>com.github.chrislusf</groupId>
            <artifactId>seaweedfs-hadoop2-client</artifactId>
-            <version>1.5.6</version>
+            <version>1.5.7</version>
            <scope>compile</scope>
        </dependency>
        <dependency>
@@ -1,6 +1,6 @@
 package com.seaweedfs.examples;

-import seaweed.hdfs.SeaweedInputStream;
+import seaweed.hdfs.SeaweedHadoopInputStream;
 import seaweedfs.client.FilerClient;
 import seaweedfs.client.FilerGrpcClient;

@@ -24,7 +24,7 @@ public class UnzipFile {

        long localProcessTime = startTime2 - startTime;

-        SeaweedInputStream seaweedInputStream = new SeaweedInputStream(
+        SeaweedHadoopInputStream seaweedInputStream = new SeaweedHadoopInputStream(
                filerGrpcClient,
                new org.apache.hadoop.fs.FileSystem.Statistics(""),
                "/",
@@ -301,7 +301,7 @@
    </snapshotRepository>
  </distributionManagement>
  <properties>
-    <seaweedfs.client.version>1.5.6</seaweedfs.client.version>
+    <seaweedfs.client.version>1.5.7</seaweedfs.client.version>
    <hadoop.version>2.9.2</hadoop.version>
  </properties>
 </project>
@@ -5,7 +5,7 @@
    <modelVersion>4.0.0</modelVersion>

    <properties>
-        <seaweedfs.client.version>1.5.6</seaweedfs.client.version>
+        <seaweedfs.client.version>1.5.7</seaweedfs.client.version>
        <hadoop.version>2.9.2</hadoop.version>
    </properties>

@@ -216,7 +216,7 @@ public class SeaweedFileSystemStore {
            SeaweedWrite.writeMeta(filerGrpcClient, getParentDirectory(path), entry);
        }

-        return new SeaweedOutputStream(filerGrpcClient, path, entry, writePosition, bufferSize, replication);
+        return new SeaweedHadoopOutputStream(filerGrpcClient, path.toString(), entry, writePosition, bufferSize, replication);

    }

@@ -230,7 +230,7 @@ public class SeaweedFileSystemStore {
            throw new FileNotFoundException("read non-exist file " + path);
        }

-        return new SeaweedInputStream(filerGrpcClient,
+        return new SeaweedHadoopInputStream(filerGrpcClient,
            statistics,
            path.toUri().getPath(),
            entry);
@@ -0,0 +1,150 @@
+package seaweed.hdfs;
+
+// based on org.apache.hadoop.fs.azurebfs.services.AbfsInputStream
+
+import org.apache.hadoop.fs.ByteBufferReadable;
+import org.apache.hadoop.fs.FSInputStream;
+import org.apache.hadoop.fs.FileSystem.Statistics;
+import seaweedfs.client.FilerGrpcClient;
+import seaweedfs.client.FilerProto;
+import seaweedfs.client.SeaweedInputStream;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+public class SeaweedHadoopInputStream extends FSInputStream implements ByteBufferReadable {
+
+    private final SeaweedInputStream seaweedInputStream;
+    private final Statistics statistics;
+
+    public SeaweedHadoopInputStream(
+            final FilerGrpcClient filerGrpcClient,
+            final Statistics statistics,
+            final String path,
+            final FilerProto.Entry entry) throws IOException {
+        this.seaweedInputStream = new SeaweedInputStream(filerGrpcClient, path, entry);
+        this.statistics = statistics;
+    }
+
+    @Override
+    public int read() throws IOException {
+        return seaweedInputStream.read();
+    }
+
+    @Override
+    public int read(final byte[] b, final int off, final int len) throws IOException {
+        return seaweedInputStream.read(b, off, len);
+    }
+
+    // implement ByteBufferReadable
+    @Override
+    public synchronized int read(ByteBuffer buf) throws IOException {
+        int bytesRead = seaweedInputStream.read(buf);
+
+        if (bytesRead > 0) {
+            if (statistics != null) {
+                statistics.incrementBytesRead(bytesRead);
+            }
+        }
+
+        return bytesRead;
+    }
+
+    /**
+     * Seek to given position in stream.
+     *
+     * @param n position to seek to
+     * @throws IOException  if there is an error
+     * @throws EOFException if attempting to seek past end of file
+     */
+    @Override
+    public synchronized void seek(long n) throws IOException {
+        seaweedInputStream.seek(n);
+    }
+
+    @Override
+    public synchronized long skip(long n) throws IOException {
+        return seaweedInputStream.skip(n);
+    }
+
+    /**
+     * Return the size of the remaining available bytes
+     * if the size is less than or equal to {@link Integer#MAX_VALUE},
+     * otherwise, return {@link Integer#MAX_VALUE}.
+     * <p>
+     * This is to match the behavior of DFSInputStream.available(),
+     * which some clients may rely on (HBase write-ahead log reading in
+     * particular).
+     */
+    @Override
+    public synchronized int available() throws IOException {
+        return seaweedInputStream.available();
+    }
+
+    /**
+     * Returns the length of the file that this stream refers to. Note that the length returned is the length
+     * as of the time the Stream was opened. Specifically, if there have been subsequent appends to the file,
+     * they wont be reflected in the returned length.
+     *
+     * @return length of the file.
+     * @throws IOException if the stream is closed
+     */
+    public long length() throws IOException {
+        return seaweedInputStream.length();
+    }
+
+    /**
+     * Return the current offset from the start of the file
+     *
+     * @throws IOException throws {@link IOException} if there is an error
+     */
+    @Override
+    public synchronized long getPos() throws IOException {
+        return seaweedInputStream.getPos();
+    }
+
+    /**
+     * Seeks a different copy of the data.  Returns true if
+     * found a new source, false otherwise.
+     *
+     * @throws IOException throws {@link IOException} if there is an error
+     */
+    @Override
+    public boolean seekToNewSource(long l) throws IOException {
+        return false;
+    }
+
+    @Override
+    public synchronized void close() throws IOException {
+        seaweedInputStream.close();
+    }
+
+    /**
+     * Not supported by this stream. Throws {@link UnsupportedOperationException}
+     *
+     * @param readlimit ignored
+     */
+    @Override
+    public synchronized void mark(int readlimit) {
+        throw new UnsupportedOperationException("mark()/reset() not supported on this stream");
+    }
+
+    /**
+     * Not supported by this stream. Throws {@link UnsupportedOperationException}
+     */
+    @Override
+    public synchronized void reset() throws IOException {
+        throw new UnsupportedOperationException("mark()/reset() not supported on this stream");
+    }
+
+    /**
+     * gets whether mark and reset are supported by {@code ADLFileInputStream}. Always returns false.
+     *
+     * @return always {@code false}
+     */
+    @Override
+    public boolean markSupported() {
+        return false;
+    }
+}
@@ -0,0 +1,16 @@
+package seaweed.hdfs;
+
+// adapted from org.apache.hadoop.fs.azurebfs.services.AbfsOutputStream
+
+import seaweedfs.client.FilerGrpcClient;
+import seaweedfs.client.FilerProto;
+import seaweedfs.client.SeaweedOutputStream;
+
+public class SeaweedHadoopOutputStream extends SeaweedOutputStream {
+
+    public SeaweedHadoopOutputStream(FilerGrpcClient filerGrpcClient, final String path, FilerProto.Entry.Builder entry,
+                                     final long position, final int bufferSize, final String replication) {
+        super(filerGrpcClient, path.toString(), entry, position, bufferSize, replication);
+    }
+
+}
@@ -309,7 +309,7 @@
    </snapshotRepository>
  </distributionManagement>
  <properties>
-    <seaweedfs.client.version>1.5.6</seaweedfs.client.version>
+    <seaweedfs.client.version>1.5.7</seaweedfs.client.version>
    <hadoop.version>3.1.1</hadoop.version>
  </properties>
 </project>
@@ -5,7 +5,7 @@
    <modelVersion>4.0.0</modelVersion>

    <properties>
-        <seaweedfs.client.version>1.5.6</seaweedfs.client.version>
+        <seaweedfs.client.version>1.5.7</seaweedfs.client.version>
        <hadoop.version>3.1.1</hadoop.version>
    </properties>

@@ -216,7 +216,7 @@ public class SeaweedFileSystemStore {
            SeaweedWrite.writeMeta(filerGrpcClient, getParentDirectory(path), entry);
        }

-        return new SeaweedOutputStream(filerGrpcClient, path, entry, writePosition, bufferSize, replication);
+        return new SeaweedHadoopOutputStream(filerGrpcClient, path.toString(), entry, writePosition, bufferSize, replication);

    }

@@ -230,7 +230,7 @@ public class SeaweedFileSystemStore {
            throw new FileNotFoundException("read non-exist file " + path);
        }

-        return new SeaweedInputStream(filerGrpcClient,
+        return new SeaweedHadoopInputStream(filerGrpcClient,
            statistics,
            path.toUri().getPath(),
            entry);
@@ -0,0 +1,150 @@
+package seaweed.hdfs;
+
+// based on org.apache.hadoop.fs.azurebfs.services.AbfsInputStream
+
+import org.apache.hadoop.fs.ByteBufferReadable;
+import org.apache.hadoop.fs.FSInputStream;
+import org.apache.hadoop.fs.FileSystem.Statistics;
+import seaweedfs.client.FilerGrpcClient;
+import seaweedfs.client.FilerProto;
+import seaweedfs.client.SeaweedInputStream;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+public class SeaweedHadoopInputStream extends FSInputStream implements ByteBufferReadable {
+
+    private final SeaweedInputStream seaweedInputStream;
+    private final Statistics statistics;
+
+    public SeaweedHadoopInputStream(
+            final FilerGrpcClient filerGrpcClient,
+            final Statistics statistics,
+            final String path,
+            final FilerProto.Entry entry) throws IOException {
+        this.seaweedInputStream = new SeaweedInputStream(filerGrpcClient, path, entry);
+        this.statistics = statistics;
+    }
+
+    @Override
+    public int read() throws IOException {
+        return seaweedInputStream.read();
+    }
+
+    @Override
+    public int read(final byte[] b, final int off, final int len) throws IOException {
+        return seaweedInputStream.read(b, off, len);
+    }
+
+    // implement ByteBufferReadable
+    @Override
+    public synchronized int read(ByteBuffer buf) throws IOException {
+        int bytesRead = seaweedInputStream.read(buf);
+
+        if (bytesRead > 0) {
+            if (statistics != null) {
+                statistics.incrementBytesRead(bytesRead);
+            }
+        }
+
+        return bytesRead;
+    }
+
+    /**
+     * Seek to given position in stream.
+     *
+     * @param n position to seek to
+     * @throws IOException  if there is an error
+     * @throws EOFException if attempting to seek past end of file
+     */
+    @Override
+    public synchronized void seek(long n) throws IOException {
+        seaweedInputStream.seek(n);
+    }
+
+    @Override
+    public synchronized long skip(long n) throws IOException {
+        return seaweedInputStream.skip(n);
+    }
+
+    /**
+     * Return the size of the remaining available bytes
+     * if the size is less than or equal to {@link Integer#MAX_VALUE},
+     * otherwise, return {@link Integer#MAX_VALUE}.
+     * <p>
+     * This is to match the behavior of DFSInputStream.available(),
+     * which some clients may rely on (HBase write-ahead log reading in
+     * particular).
+     */
+    @Override
+    public synchronized int available() throws IOException {
+        return seaweedInputStream.available();
+    }
+
+    /**
+     * Returns the length of the file that this stream refers to. Note that the length returned is the length
+     * as of the time the Stream was opened. Specifically, if there have been subsequent appends to the file,
+     * they wont be reflected in the returned length.
+     *
+     * @return length of the file.
+     * @throws IOException if the stream is closed
+     */
+    public long length() throws IOException {
+        return seaweedInputStream.length();
+    }
+
+    /**
+     * Return the current offset from the start of the file
+     *
+     * @throws IOException throws {@link IOException} if there is an error
+     */
+    @Override
+    public synchronized long getPos() throws IOException {
+        return seaweedInputStream.getPos();
+    }
+
+    /**
+     * Seeks a different copy of the data.  Returns true if
+     * found a new source, false otherwise.
+     *
+     * @throws IOException throws {@link IOException} if there is an error
+     */
+    @Override
+    public boolean seekToNewSource(long l) throws IOException {
+        return false;
+    }
+
+    @Override
+    public synchronized void close() throws IOException {
+        seaweedInputStream.close();
+    }
+
+    /**
+     * Not supported by this stream. Throws {@link UnsupportedOperationException}
+     *
+     * @param readlimit ignored
+     */
+    @Override
+    public synchronized void mark(int readlimit) {
+        throw new UnsupportedOperationException("mark()/reset() not supported on this stream");
+    }
+
+    /**
+     * Not supported by this stream. Throws {@link UnsupportedOperationException}
+     */
+    @Override
+    public synchronized void reset() throws IOException {
+        throw new UnsupportedOperationException("mark()/reset() not supported on this stream");
+    }
+
+    /**
+     * gets whether mark and reset are supported by {@code ADLFileInputStream}. Always returns false.
+     *
+     * @return always {@code false}
+     */
+    @Override
+    public boolean markSupported() {
+        return false;
+    }
+}
@@ -0,0 +1,64 @@
+package seaweed.hdfs;
+
+// adapted from org.apache.hadoop.fs.azurebfs.services.AbfsOutputStream
+
+import org.apache.hadoop.fs.StreamCapabilities;
+import org.apache.hadoop.fs.Syncable;
+import seaweedfs.client.FilerGrpcClient;
+import seaweedfs.client.FilerProto;
+import seaweedfs.client.SeaweedOutputStream;
+
+import java.io.IOException;
+import java.util.Locale;
+
+public class SeaweedHadoopOutputStream extends SeaweedOutputStream implements Syncable, StreamCapabilities {
+
+    public SeaweedHadoopOutputStream(FilerGrpcClient filerGrpcClient, final String path, FilerProto.Entry.Builder entry,
+                                     final long position, final int bufferSize, final String replication) {
+        super(filerGrpcClient, path, entry, position, bufferSize, replication);
+    }
+
+    /**
+     * Similar to posix fsync, flush out the data in client's user buffer
+     * all the way to the disk device (but the disk may have it in its cache).
+     *
+     * @throws IOException if error occurs
+     */
+    @Override
+    public void hsync() throws IOException {
+        if (supportFlush) {
+            flushInternal();
+        }
+    }
+
+    /**
+     * Flush out the data in client's user buffer. After the return of
+     * this call, new readers will see the data.
+     *
+     * @throws IOException if any error occurs
+     */
+    @Override
+    public void hflush() throws IOException {
+        if (supportFlush) {
+            flushInternal();
+        }
+    }
+
+    /**
+     * Query the stream for a specific capability.
+     *
+     * @param capability string to query the stream support for.
+     * @return true for hsync and hflush.
+     */
+    @Override
+    public boolean hasCapability(String capability) {
+        switch (capability.toLowerCase(Locale.ENGLISH)) {
+            case StreamCapabilities.HSYNC:
+            case StreamCapabilities.HFLUSH:
+                return supportFlush;
+            default:
+                return false;
+        }
+    }
+
+}
@@ -1,259 +0,0 @@
-package seaweed.hdfs;
-
-// based on org.apache.hadoop.fs.azurebfs.services.AbfsInputStream
-
-import org.apache.hadoop.fs.ByteBufferReadable;
-import org.apache.hadoop.fs.FSExceptionMessages;
-import org.apache.hadoop.fs.FSInputStream;
-import org.apache.hadoop.fs.FileSystem.Statistics;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import seaweedfs.client.FilerGrpcClient;
-import seaweedfs.client.FilerProto;
-import seaweedfs.client.SeaweedRead;
-
-import java.io.EOFException;
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.List;
-
-public class SeaweedInputStream extends FSInputStream implements ByteBufferReadable {
-
-    private static final Logger LOG = LoggerFactory.getLogger(SeaweedInputStream.class);
-
-    private final FilerGrpcClient filerGrpcClient;
-    private final Statistics statistics;
-    private final String path;
-    private final FilerProto.Entry entry;
-    private final List<SeaweedRead.VisibleInterval> visibleIntervalList;
-    private final long contentLength;
-
-    private long position = 0;  // cursor of the file
-
-    private boolean closed = false;
-
-    public SeaweedInputStream(
-            final FilerGrpcClient filerGrpcClient,
-            final Statistics statistics,
-            final String path,
-            final FilerProto.Entry entry) throws IOException {
-        this.filerGrpcClient = filerGrpcClient;
-        this.statistics = statistics;
-        this.path = path;
-        this.entry = entry;
-        this.contentLength = SeaweedRead.fileSize(entry);
-
-        this.visibleIntervalList = SeaweedRead.nonOverlappingVisibleIntervals(filerGrpcClient, entry.getChunksList());
-
-        LOG.debug("new path:{} entry:{} visibleIntervalList:{}", path, entry, visibleIntervalList);
-
-    }
-
-    public String getPath() {
-        return path;
-    }
-
-    @Override
-    public int read() throws IOException {
-        byte[] b = new byte[1];
-        int numberOfBytesRead = read(b, 0, 1);
-        if (numberOfBytesRead < 0) {
-            return -1;
-        } else {
-            return (b[0] & 0xFF);
-        }
-    }
-
-    @Override
-    public int read(final byte[] b, final int off, final int len) throws IOException {
-
-        if (b == null) {
-            throw new IllegalArgumentException("null byte array passed in to read() method");
-        }
-        if (off >= b.length) {
-            throw new IllegalArgumentException("offset greater than length of array");
-        }
-        if (len < 0) {
-            throw new IllegalArgumentException("requested read length is less than zero");
-        }
-        if (len > (b.length - off)) {
-            throw new IllegalArgumentException("requested read length is more than will fit after requested offset in buffer");
-        }
-
-        ByteBuffer buf = ByteBuffer.wrap(b, off, len);
-        return read(buf);
-
-    }
-
-    // implement ByteBufferReadable
-    @Override
-    public synchronized int read(ByteBuffer buf) throws IOException {
-
-        if (position < 0) {
-            throw new IllegalArgumentException("attempting to read from negative offset");
-        }
-        if (position >= contentLength) {
-            return -1;  // Hadoop prefers -1 to EOFException
-        }
-
-        long bytesRead = 0;
-        int len = buf.remaining();
-        int start = (int) this.position;
-        if (start+len <= entry.getContent().size()) {
-            entry.getContent().substring(start, start+len).copyTo(buf);
-        } else {
-            bytesRead = SeaweedRead.read(this.filerGrpcClient, this.visibleIntervalList, this.position, buf, SeaweedRead.fileSize(entry));
-        }
-
-        if (bytesRead > Integer.MAX_VALUE) {
-            throw new IOException("Unexpected Content-Length");
-        }
-
-        if (bytesRead > 0) {
-            this.position += bytesRead;
-            if (statistics != null) {
-                statistics.incrementBytesRead(bytesRead);
-            }
-        }
-
-        return (int) bytesRead;
-    }
-
-    /**
-     * Seek to given position in stream.
-     *
-     * @param n position to seek to
-     * @throws IOException  if there is an error
-     * @throws EOFException if attempting to seek past end of file
-     */
-    @Override
-    public synchronized void seek(long n) throws IOException {
-        if (closed) {
-            throw new IOException(FSExceptionMessages.STREAM_IS_CLOSED);
-        }
-        if (n < 0) {
-            throw new EOFException(FSExceptionMessages.NEGATIVE_SEEK);
-        }
-        if (n > contentLength) {
-            throw new EOFException(FSExceptionMessages.CANNOT_SEEK_PAST_EOF);
-        }
-
-        this.position = n;
-
-    }
-
-    @Override
-    public synchronized long skip(long n) throws IOException {
-        if (closed) {
-            throw new IOException(FSExceptionMessages.STREAM_IS_CLOSED);
-        }
-        if (this.position == contentLength) {
-            if (n > 0) {
-                throw new EOFException(FSExceptionMessages.CANNOT_SEEK_PAST_EOF);
-            }
-        }
-        long newPos = this.position + n;
-        if (newPos < 0) {
-            newPos = 0;
-            n = newPos - this.position;
-        }
-        if (newPos > contentLength) {
-            newPos = contentLength;
-            n = newPos - this.position;
-        }
-        seek(newPos);
-        return n;
-    }
-
-    /**
-     * Return the size of the remaining available bytes
-     * if the size is less than or equal to {@link Integer#MAX_VALUE},
-     * otherwise, return {@link Integer#MAX_VALUE}.
-     * <p>
-     * This is to match the behavior of DFSInputStream.available(),
-     * which some clients may rely on (HBase write-ahead log reading in
-     * particular).
-     */
-    @Override
-    public synchronized int available() throws IOException {
-        if (closed) {
-            throw new IOException(
-                    FSExceptionMessages.STREAM_IS_CLOSED);
-        }
-        final long remaining = this.contentLength - this.getPos();
-        return remaining <= Integer.MAX_VALUE
-                ? (int) remaining : Integer.MAX_VALUE;
-    }
-
-    /**
-     * Returns the length of the file that this stream refers to. Note that the length returned is the length
-     * as of the time the Stream was opened. Specifically, if there have been subsequent appends to the file,
-     * they wont be reflected in the returned length.
-     *
-     * @return length of the file.
-     * @throws IOException if the stream is closed
-     */
-    public long length() throws IOException {
-        if (closed) {
-            throw new IOException(FSExceptionMessages.STREAM_IS_CLOSED);
-        }
-        return contentLength;
-    }
-
-    /**
-     * Return the current offset from the start of the file
-     *
-     * @throws IOException throws {@link IOException} if there is an error
-     */
-    @Override
-    public synchronized long getPos() throws IOException {
-        if (closed) {
-            throw new IOException(FSExceptionMessages.STREAM_IS_CLOSED);
-        }
-        return position;
-    }
-
-    /**
-     * Seeks a different copy of the data.  Returns true if
-     * found a new source, false otherwise.
-     *
-     * @throws IOException throws {@link IOException} if there is an error
-     */
-    @Override
-    public boolean seekToNewSource(long l) throws IOException {
-        return false;
-    }
-
-    @Override
-    public synchronized void close() throws IOException {
-        closed = true;
-    }
-
-    /**
-     * Not supported by this stream. Throws {@link UnsupportedOperationException}
-     *
-     * @param readlimit ignored
-     */
-    @Override
-    public synchronized void mark(int readlimit) {
-        throw new UnsupportedOperationException("mark()/reset() not supported on this stream");
-    }
-
-    /**
-     * Not supported by this stream. Throws {@link UnsupportedOperationException}
-     */
-    @Override
-    public synchronized void reset() throws IOException {
-        throw new UnsupportedOperationException("mark()/reset() not supported on this stream");
-    }
-
-    /**
-     * gets whether mark and reset are supported by {@code ADLFileInputStream}. Always returns false.
-     *
-     * @return always {@code false}
-     */
-    @Override
-    public boolean markSupported() {
-        return false;
-    }
-}
@@ -1,337 +0,0 @@
-package seaweed.hdfs;
-
-// adapted from org.apache.hadoop.fs.azurebfs.services.AbfsOutputStream
-
-import com.google.common.base.Preconditions;
-import org.apache.hadoop.fs.FSExceptionMessages;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.StreamCapabilities;
-import org.apache.hadoop.fs.Syncable;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import seaweedfs.client.ByteBufferPool;
-import seaweedfs.client.FilerGrpcClient;
-import seaweedfs.client.FilerProto;
-import seaweedfs.client.SeaweedWrite;
-
-import java.io.IOException;
-import java.io.InterruptedIOException;
-import java.io.OutputStream;
-import java.nio.ByteBuffer;
-import java.util.Locale;
-import java.util.concurrent.*;
-
-import static seaweed.hdfs.SeaweedFileSystemStore.getParentDirectory;
-
-public class SeaweedOutputStream extends OutputStream implements Syncable, StreamCapabilities {
-
-    private static final Logger LOG = LoggerFactory.getLogger(SeaweedOutputStream.class);
-
-    private final FilerGrpcClient filerGrpcClient;
-    private final Path path;
-    private final int bufferSize;
-    private final int maxConcurrentRequestCount;
-    private final ThreadPoolExecutor threadExecutor;
-    private final ExecutorCompletionService<Void> completionService;
-    private final FilerProto.Entry.Builder entry;
-    private final boolean supportFlush = false; // true;
-    private final ConcurrentLinkedDeque<WriteOperation> writeOperations;
-    private long position;
-    private boolean closed;
-    private volatile IOException lastError;
-    private long lastFlushOffset;
-    private long lastTotalAppendOffset = 0;
-    private ByteBuffer buffer;
-    private long outputIndex;
-    private String replication = "000";
-
-    public SeaweedOutputStream(FilerGrpcClient filerGrpcClient, final Path path, FilerProto.Entry.Builder entry,
-                               final long position, final int bufferSize, final String replication) {
-        this.filerGrpcClient = filerGrpcClient;
-        this.replication = replication;
-        this.path = path;
-        this.position = position;
-        this.closed = false;
-        this.lastError = null;
-        this.lastFlushOffset = 0;
-        this.bufferSize = bufferSize;
-        this.buffer = ByteBufferPool.request(bufferSize);
-        this.outputIndex = 0;
-        this.writeOperations = new ConcurrentLinkedDeque<>();
-
-        this.maxConcurrentRequestCount = Runtime.getRuntime().availableProcessors();
-
-        this.threadExecutor
-                = new ThreadPoolExecutor(maxConcurrentRequestCount,
-                maxConcurrentRequestCount,
-                120L,
-                TimeUnit.SECONDS,
-                new LinkedBlockingQueue<Runnable>());
-        this.completionService = new ExecutorCompletionService<>(this.threadExecutor);
-
-        this.entry = entry;
-
-    }
-
-    private synchronized void flushWrittenBytesToServiceInternal(final long offset) throws IOException {
-        try {
-            SeaweedWrite.writeMeta(filerGrpcClient, getParentDirectory(path), entry);
-        } catch (Exception ex) {
-            throw new IOException(ex);
-        }
-        this.lastFlushOffset = offset;
-    }
-
-    @Override
-    public void write(final int byteVal) throws IOException {
-        write(new byte[]{(byte) (byteVal & 0xFF)});
-    }
-
-    @Override
-    public synchronized void write(final byte[] data, final int off, final int length)
-            throws IOException {
-        maybeThrowLastError();
-
-        Preconditions.checkArgument(data != null, "null data");
-
-        if (off < 0 || length < 0 || length > data.length - off) {
-            throw new IndexOutOfBoundsException();
-        }
-
-        // System.out.println(path + " write [" + (outputIndex + off) + "," + ((outputIndex + off) + length) + ")");
-
-        int currentOffset = off;
-        int writableBytes = bufferSize - buffer.position();
-        int numberOfBytesToWrite = length;
-
-        while (numberOfBytesToWrite > 0) {
-
-            if (numberOfBytesToWrite < writableBytes) {
-                buffer.put(data, currentOffset, numberOfBytesToWrite);
-                outputIndex += numberOfBytesToWrite;
-                break;
-            }
-
-            // System.out.println(path + "     [" + (outputIndex + currentOffset) + "," + ((outputIndex + currentOffset) + writableBytes) + ") " + buffer.capacity());
-            buffer.put(data, currentOffset, writableBytes);
-            outputIndex += writableBytes;
-            currentOffset += writableBytes;
-            writeCurrentBufferToService();
-            numberOfBytesToWrite = numberOfBytesToWrite - writableBytes;
-            writableBytes = bufferSize - buffer.position();
-        }
-
-    }
-
-    /**
-     * Flushes this output stream and forces any buffered output bytes to be
-     * written out. If any data remains in the payload it is committed to the
-     * service. Data is queued for writing and forced out to the service
-     * before the call returns.
-     */
-    @Override
-    public void flush() throws IOException {
-        if (supportFlush) {
-            flushInternalAsync();
-        }
-    }
-
-    /**
-     * Similar to posix fsync, flush out the data in client's user buffer
-     * all the way to the disk device (but the disk may have it in its cache).
-     *
-     * @throws IOException if error occurs
-     */
-    @Override
-    public void hsync() throws IOException {
-        if (supportFlush) {
-            flushInternal();
-        }
-    }
-
-    /**
-     * Flush out the data in client's user buffer. After the return of
-     * this call, new readers will see the data.
-     *
-     * @throws IOException if any error occurs
-     */
-    @Override
-    public void hflush() throws IOException {
-        if (supportFlush) {
-            flushInternal();
-        }
-    }
-
-    /**
-     * Query the stream for a specific capability.
-     *
-     * @param capability string to query the stream support for.
-     * @return true for hsync and hflush.
-     */
-    @Override
-    public boolean hasCapability(String capability) {
-        switch (capability.toLowerCase(Locale.ENGLISH)) {
-            case StreamCapabilities.HSYNC:
-            case StreamCapabilities.HFLUSH:
-                return supportFlush;
-            default:
-                return false;
-        }
-    }
-
-    /**
-     * Force all data in the output stream to be written to Azure storage.
-     * Wait to return until this is complete. Close the access to the stream and
-     * shutdown the upload thread pool.
-     * If the blob was created, its lease will be released.
-     * Any error encountered caught in threads and stored will be rethrown here
-     * after cleanup.
-     */
-    @Override
-    public synchronized void close() throws IOException {
-        if (closed) {
-            return;
-        }
-
-        LOG.debug("close path: {}", path);
-        try {
-            flushInternal();
-            threadExecutor.shutdown();
-        } finally {
-            lastError = new IOException(FSExceptionMessages.STREAM_IS_CLOSED);
-            ByteBufferPool.release(buffer);
-            buffer = null;
-            outputIndex = 0;
-            closed = true;
-            writeOperations.clear();
-            if (!threadExecutor.isShutdown()) {
-                threadExecutor.shutdownNow();
-            }
-        }
-    }
-
-    private synchronized void writeCurrentBufferToService() throws IOException {
-        if (buffer.position() == 0) {
-            return;
-        }
-
-        position += submitWriteBufferToService(buffer, position);
-
-        buffer = ByteBufferPool.request(bufferSize);
-
-    }
-
-    private synchronized int submitWriteBufferToService(final ByteBuffer bufferToWrite, final long writePosition) throws IOException {
-
-        bufferToWrite.flip();
-        int bytesLength = bufferToWrite.limit() - bufferToWrite.position();
-
-        if (threadExecutor.getQueue().size() >= maxConcurrentRequestCount) {
-            waitForTaskToComplete();
-        }
-        final Future<Void> job = completionService.submit(() -> {
-            // System.out.println(path + " is going to save [" + (writePosition) + "," + ((writePosition) + bytesLength) + ")");
-            SeaweedWrite.writeData(entry, replication, filerGrpcClient, writePosition, bufferToWrite.array(), bufferToWrite.position(), bufferToWrite.limit(), path.toUri().getPath());
-            // System.out.println(path + " saved [" + (writePosition) + "," + ((writePosition) + bytesLength) + ")");
-            ByteBufferPool.release(bufferToWrite);
-            return null;
-        });
-
-        writeOperations.add(new WriteOperation(job, writePosition, bytesLength));
-
-        // Try to shrink the queue
-        shrinkWriteOperationQueue();
-
-        return bytesLength;
-
-    }
-
-    private void waitForTaskToComplete() throws IOException {
-        boolean completed;
-        for (completed = false; completionService.poll() != null; completed = true) {
-            // keep polling until there is no data
-        }
-
-        if (!completed) {
-            try {
-                completionService.take();
-            } catch (InterruptedException e) {
-                lastError = (IOException) new InterruptedIOException(e.toString()).initCause(e);
-                throw lastError;
-            }
-        }
-    }
-
-    private void maybeThrowLastError() throws IOException {
-        if (lastError != null) {
-            throw lastError;
-        }
-    }
-
-    /**
-     * Try to remove the completed write operations from the beginning of write
-     * operation FIFO queue.
-     */
-    private synchronized void shrinkWriteOperationQueue() throws IOException {
-        try {
-            while (writeOperations.peek() != null && writeOperations.peek().task.isDone()) {
-                writeOperations.peek().task.get();
-                lastTotalAppendOffset += writeOperations.peek().length;
-                writeOperations.remove();
-            }
-        } catch (Exception e) {
-            lastError = new IOException(e);
-            throw lastError;
-        }
-    }
-
-    private synchronized void flushInternal() throws IOException {
-        maybeThrowLastError();
-        writeCurrentBufferToService();
-        flushWrittenBytesToService();
-    }
-
-    private synchronized void flushInternalAsync() throws IOException {
-        maybeThrowLastError();
-        writeCurrentBufferToService();
-        flushWrittenBytesToServiceAsync();
-    }
-
-    private synchronized void flushWrittenBytesToService() throws IOException {
-        for (WriteOperation writeOperation : writeOperations) {
-            try {
-                writeOperation.task.get();
-            } catch (Exception ex) {
-                lastError = new IOException(ex);
-                throw lastError;
-            }
-        }
-        LOG.debug("flushWrittenBytesToService: {} position:{}", path, position);
-        flushWrittenBytesToServiceInternal(position);
-    }
-
-    private synchronized void flushWrittenBytesToServiceAsync() throws IOException {
-        shrinkWriteOperationQueue();
-
-        if (this.lastTotalAppendOffset > this.lastFlushOffset) {
-            this.flushWrittenBytesToServiceInternal(this.lastTotalAppendOffset);
-        }
-    }
-
-    private static class WriteOperation {
-        private final Future<Void> task;
-        private final long startOffset;
-        private final long length;
-
-        WriteOperation(final Future<Void> task, final long startOffset, final long length) {
-            Preconditions.checkNotNull(task, "task");
-            Preconditions.checkArgument(startOffset >= 0, "startOffset");
-            Preconditions.checkArgument(length >= 0, "length");
-
-            this.task = task;
-            this.startOffset = startOffset;
-            this.length = length;
-        }
-    }
-
-}
@@ -3,9 +3,11 @@ package command
 import (
 	"context"
 	"fmt"
+	"github.com/golang/protobuf/jsonpb"
 	jsoniter "github.com/json-iterator/go"
 	"github.com/olivere/elastic/v7"
 	"io"
+	"os"
 	"path/filepath"
 	"strings"
 	"time"
@@ -25,6 +27,10 @@ var cmdFilerMetaTail = &Command{
 	Short:     "see recent changes on a filer",
 	Long: `See recent changes on a filer.

+	weed filer.meta.tail -timeAgo=30h | grep truncate
+	weed filer.meta.tail -timeAgo=30h | jq .
+	weed filer.meta.tail -timeAgo=30h | jq .eventNotification.newEntry.name
+
  `,
 }

@@ -80,8 +86,12 @@ func runFilerMetaTail(cmd *Command, args []string) bool {
 		return false
 	}

+	jsonpbMarshaler := jsonpb.Marshaler{
+		EmitDefaults: false,
+	}
 	eachEntryFunc := func(resp *filer_pb.SubscribeMetadataResponse) error {
-		fmt.Printf("dir:%s %+v\n", resp.Directory, resp.EventNotification)
+		jsonpbMarshaler.Marshal(os.Stdout, resp)
+		fmt.Fprintln(os.Stdout)
 		return nil
 	}
 	if *esServers != "" {
@@ -11,6 +11,7 @@ import (
 	_ "github.com/chrislusf/seaweedfs/weed/replication/sink/b2sink"
 	_ "github.com/chrislusf/seaweedfs/weed/replication/sink/filersink"
 	_ "github.com/chrislusf/seaweedfs/weed/replication/sink/gcssink"
+	_ "github.com/chrislusf/seaweedfs/weed/replication/sink/localsink"
 	_ "github.com/chrislusf/seaweedfs/weed/replication/sink/s3sink"
 	"github.com/chrislusf/seaweedfs/weed/replication/sub"
 	"github.com/chrislusf/seaweedfs/weed/util"
@@ -97,13 +98,19 @@ func runFilerReplicate(cmd *Command, args []string) bool {
 	replicator := replication.NewReplicator(config, "source.filer.", dataSink)

 	for {
-		key, m, err := notificationInput.ReceiveMessage()
+		key, m, onSuccessFn, onFailureFn, err := notificationInput.ReceiveMessage()
 		if err != nil {
 			glog.Errorf("receive %s: %+v", key, err)
+			if onFailureFn != nil {
+				onFailureFn()
+			}
 			continue
 		}
 		if key == "" {
 			// long poll received no messages
+			if onSuccessFn != nil {
+				onSuccessFn()
+			}
 			continue
 		}
 		if m.OldEntry != nil && m.NewEntry == nil {
@@ -115,8 +122,14 @@ func runFilerReplicate(cmd *Command, args []string) bool {
 		}
 		if err = replicator.Replicate(context.Background(), key, m); err != nil {
 			glog.Errorf("replicate %s: %+v", key, err)
+			if onFailureFn != nil {
+				onFailureFn()
+			}
 		} else {
 			glog.V(1).Infof("replicated %s", key)
+			if onSuccessFn != nil {
+				onSuccessFn()
+			}
 		}
 	}

@@ -33,6 +33,8 @@ type SyncOptions struct {
 	bTtlSec         *int
 	aDebug          *bool
 	bDebug          *bool
+	aProxyByFiler   *bool
+	bProxyByFiler   *bool
 }

 var (
@@ -43,7 +45,7 @@ var (

 func init() {
 	cmdFilerSynchronize.Run = runFilerSynchronize // break init cycle
-	syncOptions.isActivePassive = cmdFilerSynchronize.Flag.Bool("isActivePassive", false, "one directional follow if true")
+	syncOptions.isActivePassive = cmdFilerSynchronize.Flag.Bool("isActivePassive", false, "one directional follow from A to B if true")
 	syncOptions.filerA = cmdFilerSynchronize.Flag.String("a", "", "filer A in one SeaweedFS cluster")
 	syncOptions.filerB = cmdFilerSynchronize.Flag.String("b", "", "filer B in the other SeaweedFS cluster")
 	syncOptions.aPath = cmdFilerSynchronize.Flag.String("a.path", "/", "directory to sync on filer A")
@@ -54,6 +56,8 @@ func init() {
 	syncOptions.bCollection = cmdFilerSynchronize.Flag.String("b.collection", "", "collection on filer B")
 	syncOptions.aTtlSec = cmdFilerSynchronize.Flag.Int("a.ttlSec", 0, "ttl in seconds on filer A")
 	syncOptions.bTtlSec = cmdFilerSynchronize.Flag.Int("b.ttlSec", 0, "ttl in seconds on filer B")
+	syncOptions.aProxyByFiler = cmdFilerSynchronize.Flag.Bool("a.filerProxy", false, "read and write file chunks by filer A instead of volume servers")
+	syncOptions.bProxyByFiler = cmdFilerSynchronize.Flag.Bool("b.filerProxy", false, "read and write file chunks by filer B instead of volume servers")
 	syncOptions.aDebug = cmdFilerSynchronize.Flag.Bool("a.debug", false, "debug mode to print out filer A received files")
 	syncOptions.bDebug = cmdFilerSynchronize.Flag.Bool("b.debug", false, "debug mode to print out filer B received files")
 	syncCpuProfile = cmdFilerSynchronize.Flag.String("cpuprofile", "", "cpu profile output file")
@@ -62,8 +66,8 @@ func init() {

 var cmdFilerSynchronize = &Command{
 	UsageLine: "filer.sync -a=<oneFilerHost>:<oneFilerPort> -b=<otherFilerHost>:<otherFilerPort>",
-	Short:     "continuously synchronize between two active-active or active-passive SeaweedFS clusters",
-	Long: `continuously synchronize file changes between two active-active or active-passive filers
+	Short:     "resumeable continuous synchronization between two active-active or active-passive SeaweedFS clusters",
+	Long: `resumeable continuous synchronization for file changes between two active-active or active-passive filers

 	filer.sync listens on filer notifications. If any file is updated, it will fetch the updated content,
 	and write to the other destination. Different from filer.replicate:
@@ -86,8 +90,9 @@ func runFilerSynchronize(cmd *Command, args []string) bool {

 	go func() {
 		for {
-			err := doSubscribeFilerMetaChanges(grpcDialOption, *syncOptions.filerA, *syncOptions.aPath, *syncOptions.filerB,
-				*syncOptions.bPath, *syncOptions.bReplication, *syncOptions.bCollection, *syncOptions.bTtlSec, *syncOptions.bDebug)
+			err := doSubscribeFilerMetaChanges(grpcDialOption, *syncOptions.filerA, *syncOptions.aPath, *syncOptions.aProxyByFiler,
+				*syncOptions.filerB, *syncOptions.bPath, *syncOptions.bReplication, *syncOptions.bCollection, *syncOptions.bTtlSec, *syncOptions.bProxyByFiler,
+				*syncOptions.bDebug)
 			if err != nil {
 				glog.Errorf("sync from %s to %s: %v", *syncOptions.filerA, *syncOptions.filerB, err)
 				time.Sleep(1747 * time.Millisecond)
@@ -98,8 +103,9 @@ func runFilerSynchronize(cmd *Command, args []string) bool {
 	if !*syncOptions.isActivePassive {
 		go func() {
 			for {
-				err := doSubscribeFilerMetaChanges(grpcDialOption, *syncOptions.filerB, *syncOptions.bPath, *syncOptions.filerA,
-					*syncOptions.aPath, *syncOptions.aReplication, *syncOptions.aCollection, *syncOptions.aTtlSec, *syncOptions.aDebug)
+				err := doSubscribeFilerMetaChanges(grpcDialOption, *syncOptions.filerB, *syncOptions.bPath, *syncOptions.bProxyByFiler,
+					*syncOptions.filerA, *syncOptions.aPath, *syncOptions.aReplication, *syncOptions.aCollection, *syncOptions.aTtlSec, *syncOptions.aProxyByFiler,
+					*syncOptions.aDebug)
 				if err != nil {
 					glog.Errorf("sync from %s to %s: %v", *syncOptions.filerB, *syncOptions.filerA, err)
 					time.Sleep(2147 * time.Millisecond)
@@ -113,8 +119,8 @@ func runFilerSynchronize(cmd *Command, args []string) bool {
 	return true
 }

-func doSubscribeFilerMetaChanges(grpcDialOption grpc.DialOption, sourceFiler, sourcePath, targetFiler, targetPath string,
-	replicationStr, collection string, ttlSec int, debug bool) error {
+func doSubscribeFilerMetaChanges(grpcDialOption grpc.DialOption, sourceFiler, sourcePath string, sourceReadChunkFromFiler bool, targetFiler, targetPath string,
+	replicationStr, collection string, ttlSec int, sinkWriteChunkByFiler, debug bool) error {

 	// read source filer signature
 	sourceFilerSignature, sourceErr := replication.ReadFilerSignature(grpcDialOption, sourceFiler)
@@ -138,9 +144,9 @@ func doSubscribeFilerMetaChanges(grpcDialOption grpc.DialOption, sourceFiler, so

 	// create filer sink
 	filerSource := &source.FilerSource{}
-	filerSource.DoInitialize(pb.ServerToGrpcAddress(sourceFiler), sourcePath)
+	filerSource.DoInitialize(sourceFiler, pb.ServerToGrpcAddress(sourceFiler), sourcePath, sourceReadChunkFromFiler)
 	filerSink := &filersink.FilerSink{}
-	filerSink.DoInitialize(pb.ServerToGrpcAddress(targetFiler), targetPath, replicationStr, collection, ttlSec, grpcDialOption)
+	filerSink.DoInitialize(targetFiler, pb.ServerToGrpcAddress(targetFiler), targetPath, replicationStr, collection, ttlSec, grpcDialOption, sinkWriteChunkByFiler)
 	filerSink.SetSourceFiler(filerSource)

 	processEventFn := func(resp *filer_pb.SubscribeMetadataResponse) error {
@@ -6,24 +6,24 @@ import (
 )

 type MountOptions struct {
-	filer                       *string
-	filerMountRootPath          *string
-	dir                         *string
-	dirAutoCreate               *bool
-	collection                  *string
-	replication                 *string
-	ttlSec                      *int
-	chunkSizeLimitMB            *int
-	concurrentWriters           *int
-	cacheDir                    *string
-	cacheSizeMB                 *int64
-	dataCenter                  *string
-	allowOthers                 *bool
-	umaskString                 *string
-	nonempty                    *bool
-	outsideContainerClusterMode *bool
-	uidMap                      *string
-	gidMap                      *string
+	filer              *string
+	filerMountRootPath *string
+	dir                *string
+	dirAutoCreate      *bool
+	collection         *string
+	replication        *string
+	ttlSec             *int
+	chunkSizeLimitMB   *int
+	concurrentWriters  *int
+	cacheDir           *string
+	cacheSizeMB        *int64
+	dataCenter         *string
+	allowOthers        *bool
+	umaskString        *string
+	nonempty           *bool
+	volumeServerAccess *string
+	uidMap             *string
+	gidMap             *string
 }

 var (
@@ -50,7 +50,7 @@ func init() {
 	mountOptions.allowOthers = cmdMount.Flag.Bool("allowOthers", true, "allows other users to access the file system")
 	mountOptions.umaskString = cmdMount.Flag.String("umask", "022", "octal umask, e.g., 022, 0111")
 	mountOptions.nonempty = cmdMount.Flag.Bool("nonempty", false, "allows the mounting over a non-empty directory")
-	mountOptions.outsideContainerClusterMode = cmdMount.Flag.Bool("outsideContainerClusterMode", false, "allows other users to access volume servers with publicUrl")
+	mountOptions.volumeServerAccess = cmdMount.Flag.String("volumeServerAccess", "direct", "access volume servers by [direct|publicUrl|filerProxy]")
 	mountOptions.uidMap = cmdMount.Flag.String("map.uid", "", "map local uid to uid on filer, comma-separated <local_uid>:<filer_uid>")
 	mountOptions.gidMap = cmdMount.Flag.String("map.gid", "", "map local gid to gid on filer, comma-separated <local_gid>:<filer_gid>")

@@ -100,9 +100,9 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
 	uid, gid := uint32(0), uint32(0)
 	mountMode := os.ModeDir | 0777
 	if err == nil {
-		mountMode = os.ModeDir | fileInfo.Mode()
+		mountMode = os.ModeDir | os.FileMode(0777)&^umask
 		uid, gid = util.GetFileUidGid(fileInfo)
-		fmt.Printf("mount point owner uid=%d gid=%d mode=%s\n", uid, gid, fileInfo.Mode())
+		fmt.Printf("mount point owner uid=%d gid=%d mode=%s\n", uid, gid, mountMode)
 	} else {
 		fmt.Printf("can not stat %s\n", dir)
 		return false
@@ -169,27 +169,29 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
 	}

 	seaweedFileSystem := filesys.NewSeaweedFileSystem(&filesys.Option{
-		FilerGrpcAddress:            filerGrpcAddress,
-		GrpcDialOption:              grpcDialOption,
-		FilerMountRootPath:          mountRoot,
-		Collection:                  *option.collection,
-		Replication:                 *option.replication,
-		TtlSec:                      int32(*option.ttlSec),
-		ChunkSizeLimit:              int64(chunkSizeLimitMB) * 1024 * 1024,
-		ConcurrentWriters:           *option.concurrentWriters,
-		CacheDir:                    *option.cacheDir,
-		CacheSizeMB:                 *option.cacheSizeMB,
-		DataCenter:                  *option.dataCenter,
-		EntryCacheTtl:               3 * time.Second,
-		MountUid:                    uid,
-		MountGid:                    gid,
-		MountMode:                   mountMode,
-		MountCtime:                  fileInfo.ModTime(),
-		MountMtime:                  time.Now(),
-		Umask:                       umask,
-		OutsideContainerClusterMode: *mountOptions.outsideContainerClusterMode,
-		Cipher:                      cipher,
-		UidGidMapper:                uidGidMapper,
+		MountDirectory:     dir,
+		FilerAddress:       filer,
+		FilerGrpcAddress:   filerGrpcAddress,
+		GrpcDialOption:     grpcDialOption,
+		FilerMountRootPath: mountRoot,
+		Collection:         *option.collection,
+		Replication:        *option.replication,
+		TtlSec:             int32(*option.ttlSec),
+		ChunkSizeLimit:     int64(chunkSizeLimitMB) * 1024 * 1024,
+		ConcurrentWriters:  *option.concurrentWriters,
+		CacheDir:           *option.cacheDir,
+		CacheSizeMB:        *option.cacheSizeMB,
+		DataCenter:         *option.dataCenter,
+		EntryCacheTtl:      3 * time.Second,
+		MountUid:           uid,
+		MountGid:           gid,
+		MountMode:          mountMode,
+		MountCtime:         fileInfo.ModTime(),
+		MountMtime:         time.Now(),
+		Umask:              umask,
+		VolumeServerAccess: *mountOptions.volumeServerAccess,
+		Cipher:             cipher,
+		UidGidMapper:       uidGidMapper,
 	})

 	// mount
@@ -206,7 +208,9 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
 	})

 	glog.V(0).Infof("mounted %s%s to %s", filer, mountRoot, dir)
-	err = fs.Serve(c, seaweedFileSystem)
+	server := fs.New(c, nil)
+	seaweedFileSystem.Server = server
+	err = server.Serve(seaweedFileSystem)

 	// check if the mount process has an error to report
 	<-c.Ready
@@ -329,7 +329,8 @@ enabled = false
 # This URL will Dial the RabbitMQ server at the URL in the environment
 # variable RABBIT_SERVER_URL and open the exchange "myexchange".
 # The exchange must have already been created by some other means, like
-# the RabbitMQ management plugin.
+# the RabbitMQ management plugin. Сreate myexchange of type fanout and myqueue then
+# create binding myexchange => myqueue
 topic_url = "rabbit://myexchange"
 sub_url = "rabbit://myqueue"
 `
@@ -350,6 +351,16 @@ grpcAddress = "localhost:18888"
 # i.e., all files with this "prefix" are sent to notification message queue.
 directory = "/buckets"

+[sink.local]
+enabled = false
+directory = "/data"
+
+[sink.local_incremental]
+# all replicated files are under modified time as yyyy-mm-dd directories
+# so each date directory contains all new and updated files.
+enabled = false
+directory = "/backup"
+
 [sink.filer]
 enabled = false
 grpcAddress = "localhost:18888"
@@ -1,6 +1,7 @@
 package etcd

 import (
+	"bytes"
 	"context"
 	"fmt"
 	"strings"
@@ -145,14 +146,21 @@ func (store *EtcdStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPat

 func (store *EtcdStore) ListDirectoryEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
 	directoryPrefix := genDirectoryKeyPrefix(dirPath, "")
+	lastFileStart := directoryPrefix
+	if startFileName != "" {
+		lastFileStart = genDirectoryKeyPrefix(dirPath, startFileName)
+	}

-	resp, err := store.client.Get(ctx, string(directoryPrefix),
+	resp, err := store.client.Get(ctx, string(lastFileStart),
 		clientv3.WithPrefix(), clientv3.WithSort(clientv3.SortByKey, clientv3.SortDescend))
 	if err != nil {
 		return lastFileName, fmt.Errorf("list %s : %v", dirPath, err)
 	}

 	for _, kv := range resp.Kvs {
+		if !bytes.HasPrefix(kv.Key, directoryPrefix) {
+			break
+		}
 		fileName := getNameFromKey(kv.Key)
 		if fileName == "" {
 			continue
@@ -160,7 +168,6 @@ func (store *EtcdStore) ListDirectoryEntries(ctx context.Context, dirPath weed_u
 		if fileName == startFileName && !includeStartFile {
 			continue
 		}
-		lastFileName = fileName
 		limit--
 		if limit < 0 {
 			break
@@ -176,6 +183,7 @@ func (store *EtcdStore) ListDirectoryEntries(ctx context.Context, dirPath weed_u
 		if !eachEntryFunc(entry) {
 			break
 		}
+		lastFileName = fileName
 	}

 	return lastFileName, err
@@ -185,8 +185,6 @@ func (store *HbaseStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPa
 			continue
 		}

-		lastFileName = fileName
-
 		value := cell.Value

 		if fileName == startFileName && !includeStartFile {
@@ -197,6 +195,9 @@ func (store *HbaseStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPa
 		if limit < 0 {
 			break
 		}
+
+		lastFileName = fileName
+
 		entry := &filer.Entry{
 			FullPath: fullpath,
 		}
@@ -187,11 +187,11 @@ func (store *LevelDBStore) ListDirectoryPrefixedEntries(ctx context.Context, dir
 		if fileName == startFileName && !includeStartFile {
 			continue
 		}
-		lastFileName = fileName
 		limit--
 		if limit < 0 {
 			break
 		}
+		lastFileName = fileName
 		entry := &filer.Entry{
 			FullPath: weed_util.NewFullPath(string(dirPath), fileName),
 		}
@@ -196,11 +196,11 @@ func (store *LevelDB2Store) ListDirectoryPrefixedEntries(ctx context.Context, di
 		if fileName == startFileName && !includeStartFile {
 			continue
 		}
-		lastFileName = fileName
 		limit--
 		if limit < 0 {
 			break
 		}
+		lastFileName = fileName
 		entry := &filer.Entry{
 			FullPath: weed_util.NewFullPath(string(dirPath), fileName),
 		}
@@ -316,11 +316,11 @@ func (store *LevelDB3Store) ListDirectoryPrefixedEntries(ctx context.Context, di
 		if fileName == startFileName && !includeStartFile {
 			continue
 		}
-		lastFileName = fileName
 		limit--
 		if limit < 0 {
 			break
 		}
+		lastFileName = fileName
 		entry := &filer.Entry{
 			FullPath: weed_util.NewFullPath(string(dirPath), fileName),
 		}
@@ -85,11 +85,11 @@ func LookupFn(filerClient filer_pb.FilerClient) wdclient.LookupFileIdFunctionTyp
 	}
 }

-func NewChunkReaderAtFromClient(filerClient filer_pb.FilerClient, chunkViews []*ChunkView, chunkCache chunk_cache.ChunkCache, fileSize int64) *ChunkReadAt {
+func NewChunkReaderAtFromClient(lookupFn wdclient.LookupFileIdFunctionType, chunkViews []*ChunkView, chunkCache chunk_cache.ChunkCache, fileSize int64) *ChunkReadAt {

 	return &ChunkReadAt{
 		chunkViews:   chunkViews,
-		lookupFileId: LookupFn(filerClient),
+		lookupFileId: lookupFn,
 		chunkCache:   chunkCache,
 		fileSize:     fileSize,
 	}
@@ -58,7 +58,7 @@ func (dir *Dir) Attr(ctx context.Context, attr *fuse.Attr) error {
 		return err
 	}

-	attr.Inode = util.FullPath(dir.FullPath()).AsInode()
+	// attr.Inode = util.FullPath(dir.FullPath()).AsInode()
 	attr.Mode = os.FileMode(dir.entry.Attributes.FileMode) | os.ModeDir
 	attr.Mtime = time.Unix(dir.entry.Attributes.Mtime, 0)
 	attr.Crtime = time.Unix(dir.entry.Attributes.Crtime, 0)
@@ -82,8 +82,8 @@ func (dir *Dir) Getxattr(ctx context.Context, req *fuse.GetxattrRequest, resp *f
 }

 func (dir *Dir) setRootDirAttributes(attr *fuse.Attr) {
-	attr.Inode = 1 // filer2.FullPath(dir.Path).AsInode()
-	attr.Valid = time.Hour
+	// attr.Inode = 1 // filer2.FullPath(dir.Path).AsInode()
+	attr.Valid = time.Second
 	attr.Uid = dir.wfs.option.MountUid
 	attr.Gid = dir.wfs.option.MountGid
 	attr.Mode = dir.wfs.option.MountMode
@@ -91,7 +91,7 @@ func (dir *Dir) setRootDirAttributes(attr *fuse.Attr) {
 	attr.Ctime = dir.wfs.option.MountCtime
 	attr.Mtime = dir.wfs.option.MountMtime
 	attr.Atime = dir.wfs.option.MountMtime
-	attr.BlockSize = 1024 * 1024
+	attr.BlockSize = blockSize
 }

 func (dir *Dir) Fsync(ctx context.Context, req *fuse.FsyncRequest) error {
@@ -175,7 +175,7 @@ func (dir *Dir) doCreateEntry(name string, mode os.FileMode, uid, gid uint32, ex
 				TtlSec:      dir.wfs.option.TtlSec,
 			},
 		},
-		OExcl: exlusive,
+		OExcl:      exlusive,
 		Signatures: []int32{dir.wfs.signature},
 	}
 	glog.V(1).Infof("create %s/%s", dir.FullPath(), name)
@@ -285,7 +285,7 @@ func (dir *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.
 		}

 		// resp.EntryValid = time.Second
-		resp.Attr.Inode = fullFilePath.AsInode()
+		// resp.Attr.Inode = fullFilePath.AsInode()
 		resp.Attr.Valid = time.Second
 		resp.Attr.Mtime = time.Unix(entry.Attributes.Mtime, 0)
 		resp.Attr.Crtime = time.Unix(entry.Attributes.Crtime, 0)
@@ -308,13 +308,11 @@ func (dir *Dir) ReadDirAll(ctx context.Context) (ret []fuse.Dirent, err error) {
 	glog.V(4).Infof("dir ReadDirAll %s", dir.FullPath())

 	processEachEntryFn := func(entry *filer_pb.Entry, isLast bool) error {
-		fullpath := util.NewFullPath(dir.FullPath(), entry.Name)
-		inode := fullpath.AsInode()
 		if entry.IsDirectory {
-			dirent := fuse.Dirent{Inode: inode, Name: entry.Name, Type: fuse.DT_Dir}
+			dirent := fuse.Dirent{Name: entry.Name, Type: fuse.DT_Dir}
 			ret = append(ret, dirent)
 		} else {
-			dirent := fuse.Dirent{Inode: inode, Name: entry.Name, Type: findFileType(uint16(entry.Attributes.FileMode))}
+			dirent := fuse.Dirent{Name: entry.Name, Type: findFileType(uint16(entry.Attributes.FileMode))}
 			ret = append(ret, dirent)
 		}
 		return nil
@@ -404,11 +402,6 @@ func (dir *Dir) removeOneFile(req *fuse.RemoveRequest) error {
 	inodeId := util.NewFullPath(dir.FullPath(), req.Name).AsInode()
 	delete(dir.wfs.handles, inodeId)

-	// delete the chunks last
-	if isDeleteData {
-		dir.wfs.deleteFileChunks(entry.Chunks)
-	}
-
 	return nil

 }
@@ -3,7 +3,6 @@ package filesys
 import (
 	"bytes"
 	"io"
-	"runtime"
 	"sync"
 	"time"

@@ -12,30 +11,20 @@ import (
 )

 type ContinuousDirtyPages struct {
-	intervals              *ContinuousIntervals
-	f                      *File
-	writeWaitGroup         sync.WaitGroup
-	chunkAddLock           sync.Mutex
-	chunkSaveErrChan       chan error
-	chunkSaveErrChanClosed bool
-	lastErr                error
-	collection             string
-	replication            string
+	intervals      *ContinuousIntervals
+	f              *File
+	writeWaitGroup sync.WaitGroup
+	chunkAddLock   sync.Mutex
+	lastErr        error
+	collection     string
+	replication    string
 }

 func newDirtyPages(file *File) *ContinuousDirtyPages {
 	dirtyPages := &ContinuousDirtyPages{
-		intervals:        &ContinuousIntervals{},
-		f:                file,
-		chunkSaveErrChan: make(chan error, runtime.NumCPU()),
+		intervals: &ContinuousIntervals{},
+		f:         file,
 	}
-	go func() {
-		for t := range dirtyPages.chunkSaveErrChan {
-			if t != nil {
-				dirtyPages.lastErr = t
-			}
-		}
-	}()
 	return dirtyPages
 }

@@ -94,15 +83,6 @@ func (pages *ContinuousDirtyPages) saveExistingLargestPageToStorage() (hasSavedD

 func (pages *ContinuousDirtyPages) saveToStorage(reader io.Reader, offset int64, size int64) {

-	errChanSize := pages.f.wfs.option.ConcurrentWriters
-	if errChanSize == 0 {
-		errChanSize = runtime.NumCPU()
-	}
-	if pages.chunkSaveErrChanClosed {
-		pages.chunkSaveErrChan = make(chan error, errChanSize)
-		pages.chunkSaveErrChanClosed = false
-	}
-
 	mtime := time.Now().UnixNano()
 	pages.writeWaitGroup.Add(1)
 	writer := func() {
@@ -112,7 +92,7 @@ func (pages *ContinuousDirtyPages) saveToStorage(reader io.Reader, offset int64,
 		chunk, collection, replication, err := pages.f.wfs.saveDataAsChunk(pages.f.fullpath())(reader, pages.f.Name, offset)
 		if err != nil {
 			glog.V(0).Infof("%s saveToStorage [%d,%d): %v", pages.f.fullpath(), offset, offset+size, err)
-			pages.chunkSaveErrChan <- err
+			pages.lastErr = err
 			return
 		}
 		chunk.Mtime = mtime
@@ -45,7 +45,7 @@ func (file *File) fullpath() util.FullPath {

 func (file *File) Attr(ctx context.Context, attr *fuse.Attr) (err error) {

-	glog.V(4).Infof("file Attr %s, open:%v, existing attr: %+v", file.fullpath(), file.isOpen, attr)
+	glog.V(4).Infof("file Attr %s, open:%v existing:%v", file.fullpath(), file.isOpen, attr)

 	entry := file.entry
 	if file.isOpen <= 0 || entry == nil {
@@ -54,7 +54,7 @@ func (file *File) Attr(ctx context.Context, attr *fuse.Attr) (err error) {
 		}
 	}

-	attr.Inode = file.fullpath().AsInode()
+	// attr.Inode = file.fullpath().AsInode()
 	attr.Valid = time.Second
 	attr.Mode = os.FileMode(entry.Attributes.FileMode)
 	attr.Size = filer.FileSize(entry)
@@ -144,9 +144,8 @@ func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *f
 				}
 			}
 			file.entry.Chunks = chunks
-			file.entryViewCache, _ = filer.NonOverlappingVisibleIntervals(filer.LookupFn(file.wfs), chunks)
+			file.entryViewCache, _ = filer.NonOverlappingVisibleIntervals(file.wfs.LookupFn(), chunks)
 			file.reader = nil
-			file.wfs.deleteFileChunks(truncatedChunks)
 		}
 		file.entry.Attributes.FileSize = req.Size
 		file.dirtyMetadata = true
@@ -326,7 +325,7 @@ func (file *File) addChunks(chunks []*filer_pb.FileChunk) {

 func (file *File) setEntry(entry *filer_pb.Entry) {
 	file.entry = entry
-	file.entryViewCache, _ = filer.NonOverlappingVisibleIntervals(filer.LookupFn(file.wfs), entry.Chunks)
+	file.entryViewCache, _ = filer.NonOverlappingVisibleIntervals(file.wfs.LookupFn(), entry.Chunks)
 	file.reader = nil
 }

@@ -90,8 +90,9 @@ func (fh *FileHandle) Read(ctx context.Context, req *fuse.ReadRequest, resp *fus
 		glog.Warningf("%s FileHandle Read %d: [%d,%d) size %d totalRead %d", fh.f.fullpath(), fh.handle, req.Offset, req.Offset+int64(req.Size), req.Size, totalRead)
 		totalRead = min(int64(len(buff)), totalRead)
 	}
-	// resp.Data = buff[:totalRead]
-	resp.Data = buff
+	if err == nil {
+		resp.Data = buff[:totalRead]
+	}

 	return err
 }
@@ -118,19 +119,21 @@ func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, error) {

 	var chunkResolveErr error
 	if fh.f.entryViewCache == nil {
-		fh.f.entryViewCache, chunkResolveErr = filer.NonOverlappingVisibleIntervals(filer.LookupFn(fh.f.wfs), fh.f.entry.Chunks)
+		fh.f.entryViewCache, chunkResolveErr = filer.NonOverlappingVisibleIntervals(fh.f.wfs.LookupFn(), fh.f.entry.Chunks)
 		if chunkResolveErr != nil {
 			return 0, fmt.Errorf("fail to resolve chunk manifest: %v", chunkResolveErr)
 		}
 		fh.f.reader = nil
 	}

-	if fh.f.reader == nil {
+	reader := fh.f.reader
+	if reader == nil {
 		chunkViews := filer.ViewFromVisibleIntervals(fh.f.entryViewCache, 0, math.MaxInt64)
-		fh.f.reader = filer.NewChunkReaderAtFromClient(fh.f.wfs, chunkViews, fh.f.wfs.chunkCache, fileSize)
+		reader = filer.NewChunkReaderAtFromClient(fh.f.wfs.LookupFn(), chunkViews, fh.f.wfs.chunkCache, fileSize)
 	}
+	fh.f.reader = reader

-	totalRead, err := fh.f.reader.ReadAt(buff, offset)
+	totalRead, err := reader.ReadAt(buff, offset)

 	if err != nil && err != io.EOF {
 		glog.Errorf("file handle read %s: %v", fh.f.fullpath(), err)
@@ -181,25 +184,20 @@ func (fh *FileHandle) Release(ctx context.Context, req *fuse.ReleaseRequest) err
 	fh.Lock()
 	defer fh.Unlock()

-	fh.f.isOpen--
-
-	if fh.f.isOpen < 0 {
+	if fh.f.isOpen <= 0 {
 		glog.V(0).Infof("Release reset %s open count %d => %d", fh.f.Name, fh.f.isOpen, 0)
 		fh.f.isOpen = 0
 		return nil
 	}

-	if fh.f.isOpen == 0 {
+	if fh.f.isOpen == 1 {

 		if err := fh.doFlush(ctx, req.Header); err != nil {
 			glog.Errorf("Release doFlush %s: %v", fh.f.Name, err)
+			return err
 		}

-		// stop the goroutine
-		if !fh.dirtyPages.chunkSaveErrChanClosed {
-			fh.dirtyPages.chunkSaveErrChanClosed = true
-			close(fh.dirtyPages.chunkSaveErrChan)
-		}
+		fh.f.isOpen--

 		fh.f.wfs.ReleaseHandle(fh.f.fullpath(), fuse.HandleID(fh.handle))
 		if closer, ok := fh.f.reader.(io.Closer); ok {
@@ -213,10 +211,18 @@ func (fh *FileHandle) Release(ctx context.Context, req *fuse.ReleaseRequest) err

 func (fh *FileHandle) Flush(ctx context.Context, req *fuse.FlushRequest) error {

+	glog.V(4).Infof("Flush %v fh %d", fh.f.fullpath(), fh.handle)
+
 	fh.Lock()
 	defer fh.Unlock()

-	return fh.doFlush(ctx, req.Header)
+	if err := fh.doFlush(ctx, req.Header); err != nil {
+		glog.Errorf("Flush doFlush %s: %v", fh.f.Name, err)
+		return err
+	}
+
+	glog.V(4).Infof("Flush %v fh %d success", fh.f.fullpath(), fh.handle)
+	return nil
 }

 func (fh *FileHandle) doFlush(ctx context.Context, header fuse.Header) error {
@@ -229,7 +235,8 @@ func (fh *FileHandle) doFlush(ctx context.Context, header fuse.Header) error {
 	fh.dirtyPages.writeWaitGroup.Wait()

 	if fh.dirtyPages.lastErr != nil {
-		return fh.dirtyPages.lastErr
+		glog.Errorf("%v doFlush last err: %v", fh.f.fullpath(), fh.dirtyPages.lastErr)
+		return fuse.EIO
 	}

 	if !fh.f.dirtyMetadata {
@@ -268,7 +275,7 @@ func (fh *FileHandle) doFlush(ctx context.Context, header fuse.Header) error {

 		manifestChunks, nonManifestChunks := filer.SeparateManifestChunks(fh.f.entry.Chunks)

-		chunks, _ := filer.CompactFileChunks(filer.LookupFn(fh.f.wfs), nonManifestChunks)
+		chunks, _ := filer.CompactFileChunks(fh.f.wfs.LookupFn(), nonManifestChunks)
 		chunks, manifestErr := filer.MaybeManifestize(fh.f.wfs.saveDataAsChunk(fh.f.fullpath()), chunks)
 		if manifestErr != nil {
 			// not good, but should be ok
@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"strings"
 	"sync"

 	"github.com/chrislusf/seaweedfs/weed/filer"
@@ -29,7 +30,12 @@ func NewMetaCache(dbFolder string, baseDir util.FullPath, uidGidMapper *UidGidMa
 		localStore:      openMetaStore(dbFolder),
 		visitedBoundary: bounded_tree.NewBoundedTree(baseDir),
 		uidGidMapper:    uidGidMapper,
-		invalidateFunc:  invalidateFunc,
+		invalidateFunc: func(fullpath util.FullPath) {
+			if baseDir != "/" && strings.HasPrefix(string(fullpath), string(baseDir)) {
+				fullpath = fullpath[len(baseDir):]
+			}
+			invalidateFunc(fullpath)
+		},
 	}
 }

@@ -3,6 +3,8 @@ package filesys
 import (
 	"context"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/filer"
+	"github.com/chrislusf/seaweedfs/weed/wdclient"
 	"math"
 	"os"
 	"path"
@@ -24,6 +26,8 @@ import (
 )

 type Option struct {
+	MountDirectory     string
+	FilerAddress       string
 	FilerGrpcAddress   string
 	GrpcDialOption     grpc.DialOption
 	FilerMountRootPath string
@@ -44,9 +48,9 @@ type Option struct {
 	MountCtime time.Time
 	MountMtime time.Time

-	OutsideContainerClusterMode bool // whether the mount runs outside SeaweedFS containers
-	Cipher                      bool // whether encrypt data on volume server
-	UidGidMapper                *meta_cache.UidGidMapper
+	VolumeServerAccess string // how to access volume servers
+	Cipher             bool   // whether encrypt data on volume server
+	UidGidMapper       *meta_cache.UidGidMapper
 }

 var _ = fs.FS(&WFS{})
@@ -72,6 +76,7 @@ type WFS struct {

 	// throttle writers
 	concurrentWriters *util.LimitedConcurrentExecutor
+	Server            *fs.Server
 }
 type statsCache struct {
 	filer_pb.StatisticsResponse
@@ -89,7 +94,7 @@ func NewSeaweedFileSystem(option *Option) *WFS {
 		},
 		signature: util.RandomInt32(),
 	}
-	cacheUniqueId := util.Md5String([]byte(option.FilerGrpcAddress + option.FilerMountRootPath + util.Version()))[0:4]
+	cacheUniqueId := util.Md5String([]byte(option.MountDirectory + option.FilerGrpcAddress + option.FilerMountRootPath + util.Version()))[0:8]
 	cacheDir := path.Join(option.CacheDir, cacheUniqueId)
 	if option.CacheSizeMB > 0 {
 		os.MkdirAll(cacheDir, os.FileMode(0777)&^option.Umask)
@@ -100,9 +105,22 @@ func NewSeaweedFileSystem(option *Option) *WFS {
 		fsNode := wfs.fsNodeCache.GetFsNode(filePath)
 		if fsNode != nil {
 			if file, ok := fsNode.(*File); ok {
+				if err := wfs.Server.InvalidateNodeData(file); err != nil {
+					glog.V(4).Infof("InvalidateNodeData %s : %v", filePath, err)
+				}
 				file.clearEntry()
 			}
 		}
+		dir, name := filePath.DirAndName()
+		parent := wfs.root
+		if dir != "/" {
+			parent = wfs.fsNodeCache.GetFsNode(util.FullPath(dir))
+		}
+		if parent != nil {
+			if err := wfs.Server.InvalidateEntry(parent, name); err != nil {
+				glog.V(4).Infof("InvalidateEntry %s : %v", filePath, err)
+			}
+		}
 	})
 	startTime := time.Now()
 	go meta_cache.SubscribeMetaEvents(wfs.metaCache, wfs.signature, wfs, wfs.option.FilerMountRootPath, startTime.UnixNano())
@@ -237,3 +255,13 @@ func (wfs *WFS) mapPbIdFromLocalToFiler(entry *filer_pb.Entry) {
 	}
 	entry.Attributes.Uid, entry.Attributes.Gid = wfs.option.UidGidMapper.LocalToFiler(entry.Attributes.Uid, entry.Attributes.Gid)
 }
+
+func (wfs *WFS) LookupFn() wdclient.LookupFileIdFunctionType {
+	if wfs.option.VolumeServerAccess == "filerProxy" {
+		return func(fileId string) (targetUrls []string, err error) {
+			return []string{"http://" + wfs.option.FilerAddress + "/?proxyChunkId=" + fileId}, nil
+		}
+	}
+	return filer.LookupFn(wfs)
+
+}
@@ -1,84 +0,0 @@
-package filesys
-
-import (
-	"context"
-
-	"google.golang.org/grpc"
-
-	"github.com/chrislusf/seaweedfs/weed/filer"
-	"github.com/chrislusf/seaweedfs/weed/glog"
-	"github.com/chrislusf/seaweedfs/weed/operation"
-	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
-)
-
-func (wfs *WFS) deleteFileChunks(chunks []*filer_pb.FileChunk) {
-	if len(chunks) == 0 {
-		return
-	}
-
-	var fileIds []string
-	for _, chunk := range chunks {
-		if !chunk.IsChunkManifest {
-			fileIds = append(fileIds, chunk.GetFileIdString())
-			continue
-		}
-		dataChunks, manifestResolveErr := filer.ResolveOneChunkManifest(filer.LookupFn(wfs), chunk)
-		if manifestResolveErr != nil {
-			glog.V(0).Infof("failed to resolve manifest %s: %v", chunk.FileId, manifestResolveErr)
-		}
-		for _, dChunk := range dataChunks {
-			fileIds = append(fileIds, dChunk.GetFileIdString())
-		}
-		fileIds = append(fileIds, chunk.GetFileIdString())
-	}
-
-	wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
-		wfs.deleteFileIds(wfs.option.GrpcDialOption, client, fileIds)
-		return nil
-	})
-}
-
-func (wfs *WFS) deleteFileIds(grpcDialOption grpc.DialOption, client filer_pb.SeaweedFilerClient, fileIds []string) error {
-
-	var vids []string
-	for _, fileId := range fileIds {
-		vids = append(vids, filer.VolumeId(fileId))
-	}
-
-	lookupFunc := func(vids []string) (map[string]operation.LookupResult, error) {
-
-		m := make(map[string]operation.LookupResult)
-
-		glog.V(4).Infof("deleteFileIds lookup volume id locations: %v", vids)
-		resp, err := client.LookupVolume(context.Background(), &filer_pb.LookupVolumeRequest{
-			VolumeIds: vids,
-		})
-		if err != nil {
-			return m, err
-		}
-
-		for _, vid := range vids {
-			lr := operation.LookupResult{
-				VolumeId:  vid,
-				Locations: nil,
-			}
-			locations, found := resp.LocationsMap[vid]
-			if !found {
-				continue
-			}
-			for _, loc := range locations.Locations {
-				lr.Locations = append(lr.Locations, operation.Location{
-					Url:       wfs.AdjustedUrl(loc),
-					PublicUrl: loc.PublicUrl,
-				})
-			}
-			m[vid] = lr
-		}
-
-		return m, err
-	}
-
-	_, err := operation.DeleteFilesWithLookupVolumeId(grpcDialOption, fileIds, lookupFunc)
-
-	return err
-}
@@ -27,7 +27,7 @@ func (wfs *WFS) WithFilerClient(fn func(filer_pb.SeaweedFilerClient) error) erro
 }

 func (wfs *WFS) AdjustedUrl(location *filer_pb.Location) string {
-	if wfs.option.OutsideContainerClusterMode {
+	if wfs.option.VolumeServerAccess == "publicUrl" {
 		return location.PublicUrl
 	}
 	return location.Url
@@ -53,6 +53,9 @@ func (wfs *WFS) saveDataAsChunk(fullPath util.FullPath) filer.SaveDataAsChunkFun
 		}

 		fileUrl := fmt.Sprintf("http://%s/%s", host, fileId)
+		if wfs.option.VolumeServerAccess == "filerProxy" {
+			fileUrl = fmt.Sprintf("http://%s/?proxyChunkId=%s", wfs.option.FilerAddress, fileId)
+		}
 		uploadResult, err, data := operation.Upload(fileUrl, filename, wfs.option.Cipher, reader, false, "", nil, auth)
 		if err != nil {
 			glog.V(0).Infof("upload data %v to %s: %v", filename, fileUrl, err)
@@ -4,12 +4,11 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"google.golang.org/grpc"
 	"net/http"
 	"strings"
 	"sync"

-	"google.golang.org/grpc"
-
 	"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
 )

@@ -138,6 +138,22 @@ func ServerToGrpcAddress(server string) (serverGrpcAddress string) {
 	return fmt.Sprintf("%s:%d", hostnameAndPort[0], grpcPort)
 }

+func GrpcAddressToServerAddress(grpcAddress string) (serverAddress string) {
+	hostnameAndPort := strings.Split(grpcAddress, ":")
+	if len(hostnameAndPort) != 2 {
+		return fmt.Sprintf("unexpected grpcAddress: %s", grpcAddress)
+	}
+
+	grpcPort, parseErr := strconv.ParseUint(hostnameAndPort[1], 10, 64)
+	if parseErr != nil {
+		return fmt.Sprintf("failed to parse port for %s:%s", hostnameAndPort[0], hostnameAndPort[1])
+	}
+
+	port := int(grpcPort) - 10000
+
+	return fmt.Sprintf("%s:%d", hostnameAndPort[0], port)
+}
+
 func WithMasterClient(master string, grpcDialOption grpc.DialOption, fn func(client master_pb.SeaweedClient) error) error {

 	masterGrpcAddress, parseErr := ParseServerToGrpcAddress(master)
@@ -6,6 +6,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/pb"
 	"google.golang.org/grpc"
 	"strings"
+	"time"

 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
@@ -40,7 +41,17 @@ func (r *Replicator) Replicate(ctx context.Context, key string, message *filer_p
 		glog.V(4).Infof("skipping %v outside of %v", key, r.source.Dir)
 		return nil
 	}
-	newKey := util.Join(r.sink.GetSinkToDirectory(), key[len(r.source.Dir):])
+	var dateKey string
+	if r.sink.GetName() == "local_incremental" {
+		var mTime int64
+		if message.NewEntry != nil {
+			mTime = message.NewEntry.Attributes.Mtime
+		} else if message.OldEntry != nil {
+			mTime = message.OldEntry.Attributes.Mtime
+		}
+		dateKey = time.Unix(mTime, 0).Format("2006-01-02")
+	}
+	newKey := util.Join(r.sink.GetSinkToDirectory(), dateKey, key[len(r.source.Dir):])
 	glog.V(3).Infof("replicate %s => %s", key, newKey)
 	key = newKey
 	if message.OldEntry != nil && message.NewEntry == nil {
@@ -30,6 +30,7 @@ func (fs *FilerSink) replicateChunks(sourceChunks []*filer_pb.FileChunk, path st
 			replicatedChunk, e := fs.replicateOneChunk(chunk, path)
 			if e != nil {
 				err = e
+				return
 			}
 			replicatedChunks[index] = replicatedChunk
 		}(sourceChunk, chunkIndex)
@@ -97,6 +98,9 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string)
 	}

 	fileUrl := fmt.Sprintf("http://%s/%s", host, fileId)
+	if fs.writeChunkByFiler {
+		fileUrl = fmt.Sprintf("http://%s/?proxyChunkId=%s", fs.address, fileId)
+	}

 	glog.V(4).Infof("replicating %s to %s header:%+v", filename, fileUrl, header)

@@ -3,6 +3,7 @@ package filersink
 import (
 	"context"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/pb"
 	"github.com/chrislusf/seaweedfs/weed/wdclient"

 	"google.golang.org/grpc"
@@ -18,14 +19,16 @@ import (
 )

 type FilerSink struct {
-	filerSource    *source.FilerSource
-	grpcAddress    string
-	dir            string
-	replication    string
-	collection     string
-	ttlSec         int32
-	dataCenter     string
-	grpcDialOption grpc.DialOption
+	filerSource       *source.FilerSource
+	grpcAddress       string
+	dir               string
+	replication       string
+	collection        string
+	ttlSec            int32
+	dataCenter        string
+	grpcDialOption    grpc.DialOption
+	address           string
+	writeChunkByFiler bool
 }

 func init() {
@@ -42,26 +45,33 @@ func (fs *FilerSink) GetSinkToDirectory() string {

 func (fs *FilerSink) Initialize(configuration util.Configuration, prefix string) error {
 	return fs.DoInitialize(
+		"",
 		configuration.GetString(prefix+"grpcAddress"),
 		configuration.GetString(prefix+"directory"),
 		configuration.GetString(prefix+"replication"),
 		configuration.GetString(prefix+"collection"),
 		configuration.GetInt(prefix+"ttlSec"),
-		security.LoadClientTLS(util.GetViper(), "grpc.client"))
+		security.LoadClientTLS(util.GetViper(), "grpc.client"),
+		false)
 }

 func (fs *FilerSink) SetSourceFiler(s *source.FilerSource) {
 	fs.filerSource = s
 }

-func (fs *FilerSink) DoInitialize(grpcAddress string, dir string,
-	replication string, collection string, ttlSec int, grpcDialOption grpc.DialOption) (err error) {
+func (fs *FilerSink) DoInitialize(address, grpcAddress string, dir string,
+	replication string, collection string, ttlSec int, grpcDialOption grpc.DialOption, writeChunkByFiler bool) (err error) {
+	fs.address = address
+	if fs.address == "" {
+		fs.address = pb.GrpcAddressToServerAddress(grpcAddress)
+	}
 	fs.grpcAddress = grpcAddress
 	fs.dir = dir
 	fs.replication = replication
 	fs.collection = collection
 	fs.ttlSec = int32(ttlSec)
 	fs.grpcDialOption = grpcDialOption
+	fs.writeChunkByFiler = writeChunkByFiler
 	return nil
 }

@@ -0,0 +1,17 @@
+package localsink
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/replication/sink"
+)
+
+type LocalIncSink struct {
+	LocalSink
+}
+
+func (localincsink *LocalIncSink) GetName() string {
+	return "local_incremental"
+}
+
+func init() {
+	sink.Sinks = append(sink.Sinks, &LocalIncSink{})
+}
@@ -0,0 +1,101 @@
+package localsink
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/filer"
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/replication/repl_util"
+	"github.com/chrislusf/seaweedfs/weed/replication/sink"
+	"github.com/chrislusf/seaweedfs/weed/replication/source"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"strings"
+)
+
+type LocalSink struct {
+	Dir         string
+	filerSource *source.FilerSource
+}
+
+func init() {
+	sink.Sinks = append(sink.Sinks, &LocalSink{})
+}
+
+func (localsink *LocalSink) SetSourceFiler(s *source.FilerSource) {
+	localsink.filerSource = s
+}
+
+func (localsink *LocalSink) GetName() string {
+	return "local"
+}
+
+func (localsink *LocalSink) isMultiPartEntry(key string) bool {
+	return strings.HasSuffix(key, ".part") && strings.Contains(key, "/.uploads/")
+}
+
+func (localsink *LocalSink) initialize(dir string) error {
+	localsink.Dir = dir
+	return nil
+}
+
+func (localsink *LocalSink) Initialize(configuration util.Configuration, prefix string) error {
+	dir := configuration.GetString(prefix + "directory")
+	glog.V(4).Infof("sink.local.directory: %v", dir)
+	return localsink.initialize(dir)
+}
+
+func (localsink *LocalSink) GetSinkToDirectory() string {
+	return localsink.Dir
+}
+
+func (localsink *LocalSink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool, signatures []int32) error {
+	if localsink.isMultiPartEntry(key) {
+		return nil
+	}
+	glog.V(4).Infof("Delete Entry key: %s", key)
+	if err := os.Remove(key); err != nil {
+		glog.V(0).Infof("remove entry key %s: %s", key, err)
+	}
+	return nil
+}
+
+func (localsink *LocalSink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error {
+	if entry.IsDirectory || localsink.isMultiPartEntry(key) {
+		return nil
+	}
+	glog.V(4).Infof("Create Entry key: %s", key)
+
+	totalSize := filer.FileSize(entry)
+	chunkViews := filer.ViewFromChunks(localsink.filerSource.LookupFileId, entry.Chunks, 0, int64(totalSize))
+
+	dir := filepath.Dir(key)
+
+	if _, err := os.Stat(dir); os.IsNotExist(err) {
+		glog.V(4).Infof("Create Direcotry key: %s", dir)
+		if err = os.MkdirAll(dir, 0); err != nil {
+			return err
+		}
+	}
+
+	writeFunc := func(data []byte) error {
+		writeErr := ioutil.WriteFile(key, data, 0)
+		return writeErr
+	}
+
+	if err := repl_util.CopyFromChunkViews(chunkViews, localsink.filerSource, writeFunc); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (localsink *LocalSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool, signatures []int32) (foundExistingEntry bool, err error) {
+	if localsink.isMultiPartEntry(key) {
+		return true, nil
+	}
+	glog.V(4).Infof("Update Entry key: %s", key)
+	// do delete and create
+	return false, nil
+}
@@ -25,19 +25,28 @@ type FilerSource struct {
 	grpcAddress    string
 	grpcDialOption grpc.DialOption
 	Dir            string
+	address        string
+	proxyByFiler   bool
 }

 func (fs *FilerSource) Initialize(configuration util.Configuration, prefix string) error {
 	return fs.DoInitialize(
+		"",
 		configuration.GetString(prefix+"grpcAddress"),
 		configuration.GetString(prefix+"directory"),
+		false,
 	)
 }

-func (fs *FilerSource) DoInitialize(grpcAddress string, dir string) (err error) {
+func (fs *FilerSource) DoInitialize(address, grpcAddress string, dir string, readChunkFromFiler bool) (err error) {
+	fs.address = address
+	if fs.address == "" {
+		fs.address = pb.GrpcAddressToServerAddress(grpcAddress)
+	}
 	fs.grpcAddress = grpcAddress
 	fs.Dir = dir
 	fs.grpcDialOption = security.LoadClientTLS(util.GetViper(), "grpc.client")
+	fs.proxyByFiler = readChunkFromFiler
 	return nil
 }

@@ -81,9 +90,13 @@ func (fs *FilerSource) LookupFileId(part string) (fileUrls []string, err error)
 	return
 }

-func (fs *FilerSource) ReadPart(part string) (filename string, header http.Header, resp *http.Response, err error) {
+func (fs *FilerSource) ReadPart(fileId string) (filename string, header http.Header, resp *http.Response, err error) {

-	fileUrls, err := fs.LookupFileId(part)
+	if fs.proxyByFiler {
+		return util.DownloadFile("http://" + fs.address + "/?proxyChunkId=" + fileId)
+	}
+
+	fileUrls, err := fs.LookupFileId(fileId)
 	if err != nil {
 		return "", nil, nil, err
 	}
@@ -68,7 +68,7 @@ func (k *AwsSqsInput) initialize(awsAccessKeyId, awsSecretAccessKey, region, que
 	return nil
 }

-func (k *AwsSqsInput) ReceiveMessage() (key string, message *filer_pb.EventNotification, err error) {
+func (k *AwsSqsInput) ReceiveMessage() (key string, message *filer_pb.EventNotification, onSuccessFn func(), onFailureFn func(), err error) {

 	// receive message
 	result, err := k.svc.ReceiveMessage(&sqs.ReceiveMessageInput{
@@ -38,13 +38,24 @@ func (k *GoCDKPubSubInput) Initialize(configuration util.Configuration, prefix s
 	return nil
 }

-func (k *GoCDKPubSubInput) ReceiveMessage() (key string, message *filer_pb.EventNotification, err error) {
+func (k *GoCDKPubSubInput) ReceiveMessage() (key string, message *filer_pb.EventNotification, onSuccessFn func(), onFailureFn func(), err error) {
 	msg, err := k.sub.Receive(context.Background())
+	if err != nil {
+		return
+	}
+	onSuccessFn = func() {
+		msg.Ack()
+	}
+	onFailureFn = func() {
+		if msg.Nackable() {
+			msg.Nack()
+		}
+	}
 	key = msg.Metadata["key"]
 	message = &filer_pb.EventNotification{}
 	err = proto.Unmarshal(msg.Body, message)
 	if err != nil {
-		return "", nil, err
+		return "", nil, onSuccessFn, onFailureFn, err
 	}
-	return key, message, nil
+	return key, message, onSuccessFn, onFailureFn, nil
 }
@@ -85,16 +85,22 @@ func (k *GooglePubSubInput) initialize(google_application_credentials, projectId

 	go k.sub.Receive(ctx, func(ctx context.Context, m *pubsub.Message) {
 		k.messageChan <- m
-		m.Ack()
 	})

 	return err
 }

-func (k *GooglePubSubInput) ReceiveMessage() (key string, message *filer_pb.EventNotification, err error) {
+func (k *GooglePubSubInput) ReceiveMessage() (key string, message *filer_pb.EventNotification, onSuccessFn func(), onFailureFn func(), err error) {

 	m := <-k.messageChan

+	onSuccessFn = func() {
+		m.Ack()
+	}
+	onFailureFn = func() {
+		m.Nack()
+	}
+
 	// process the message
 	key = m.Attributes["key"]
 	message = &filer_pb.EventNotification{}
@@ -97,7 +97,7 @@ func (k *KafkaInput) initialize(hosts []string, topic string, offsetFile string,
 	return nil
 }

-func (k *KafkaInput) ReceiveMessage() (key string, message *filer_pb.EventNotification, err error) {
+func (k *KafkaInput) ReceiveMessage() (key string, message *filer_pb.EventNotification, onSuccessFn func(), onFailureFn func(), err error) {

 	msg := <-k.messageChan

@@ -10,7 +10,7 @@ type NotificationInput interface {
 	GetName() string
 	// Initialize initializes the file store
 	Initialize(configuration util.Configuration, prefix string) error
-	ReceiveMessage() (key string, message *filer_pb.EventNotification, err error)
+	ReceiveMessage() (key string, message *filer_pb.EventNotification, onSuccessFn func(), onFailureFn func(), err error)
 }

 var (
@@ -8,6 +8,7 @@ import (
 	"io"
 	"io/ioutil"
 	"net/http"
+	"sort"
 	"strings"

 	"github.com/chrislusf/seaweedfs/weed/s3api/s3err"
@@ -175,16 +176,15 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h
 	var deletedObjects []ObjectIdentifier
 	var deleteErrors []DeleteError

+	directoriesWithDeletion := make(map[string]int)
+
 	s3a.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {

+		// delete file entries
 		for _, object := range deleteObjects.Objects {
-			response, _ := s3a.listFilerEntries(bucket, object.ObjectName, 1, "", "/")
-			if len(response.Contents) != 0 && strings.HasSuffix(object.ObjectName, "/") {
-				continue
-			}

 			lastSeparator := strings.LastIndex(object.ObjectName, "/")
-			parentDirectoryPath, entryName, isDeleteData, isRecursive := "/", object.ObjectName, true, true
+			parentDirectoryPath, entryName, isDeleteData, isRecursive := "/", object.ObjectName, true, false
 			if lastSeparator > 0 && lastSeparator+1 < len(object.ObjectName) {
 				entryName = object.ObjectName[lastSeparator+1:]
 				parentDirectoryPath = "/" + object.ObjectName[:lastSeparator]
@@ -193,8 +193,10 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h

 			err := doDeleteEntry(client, parentDirectoryPath, entryName, isDeleteData, isRecursive)
 			if err == nil {
+				directoriesWithDeletion[parentDirectoryPath]++
 				deletedObjects = append(deletedObjects, object)
 			} else {
+				delete(directoriesWithDeletion, parentDirectoryPath)
 				deleteErrors = append(deleteErrors, DeleteError{
 					Code:    "",
 					Message: err.Error(),
@@ -202,6 +204,22 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h
 				})
 			}
 		}
+
+		// purge empty folders, only checking folders with deletions
+		var allDirs []string
+		for dir, _ := range directoriesWithDeletion {
+			allDirs = append(allDirs, dir)
+		}
+		sort.Slice(allDirs, func(i, j int) bool {
+			return len(allDirs[i]) > len(allDirs[j])
+		})
+		for _, dir := range allDirs {
+			parentDir, dirName := util.FullPath(dir).DirAndName()
+			if err := doDeleteEntry(client, parentDir, dirName, false, false); err != nil {
+				glog.V(4).Infof("directory %s has %d deletion but still not empty: %v", dir, directoriesWithDeletion[dir], err)
+			}
+		}
+
 		return nil
 	})

@@ -238,7 +238,7 @@ func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, d
 				return
 			}
 		}
-		if counter >= maxKeys + 1 {
+		if counter >= maxKeys {
 			isTruncated = true
 			return
 		}
@@ -3,18 +3,33 @@ package weed_server
 import (
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"net/http"
+	"strings"
 	"time"

 	"github.com/chrislusf/seaweedfs/weed/stats"
 )

 func (fs *FilerServer) filerHandler(w http.ResponseWriter, r *http.Request) {
+
+	start := time.Now()
+
+	// proxy to volume servers
+	var fileId string
+	if strings.HasPrefix(r.RequestURI, "/?proxyChunkId=") {
+		fileId = r.RequestURI[len("/?proxyChunkId="):]
+	}
+	if fileId != "" {
+		stats.FilerRequestCounter.WithLabelValues("proxy").Inc()
+		fs.proxyToVolumeServer(w, r, fileId)
+		stats.FilerRequestHistogram.WithLabelValues("proxy").Observe(time.Since(start).Seconds())
+		return
+	}
+
 	w.Header().Set("Server", "SeaweedFS Filer "+util.VERSION)
 	if r.Header.Get("Origin") != "" {
 		w.Header().Set("Access-Control-Allow-Origin", "*")
 		w.Header().Set("Access-Control-Allow-Credentials", "true")
 	}
-	start := time.Now()
 	switch r.Method {
 	case "GET":
 		stats.FilerRequestCounter.WithLabelValues("get").Inc()
@@ -0,0 +1,66 @@
+package weed_server
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"io"
+	"math/rand"
+	"net/http"
+)
+
+var (
+	client *http.Client
+)
+
+func init() {
+	client = &http.Client{Transport: &http.Transport{
+		MaxIdleConnsPerHost: 1024,
+	}}
+}
+
+func (fs *FilerServer) proxyToVolumeServer(w http.ResponseWriter, r *http.Request, fileId string) {
+
+	urlStrings, err := fs.filer.MasterClient.GetLookupFileIdFunction()(fileId)
+	if err != nil {
+		glog.Errorf("locate %s: %v", fileId, err)
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+
+	if len(urlStrings) == 0 {
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+
+	proxyReq, err := http.NewRequest(r.Method, urlStrings[rand.Intn(len(urlStrings))], r.Body)
+	if err != nil {
+		glog.Errorf("NewRequest %s: %v", urlStrings[0], err)
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+
+	proxyReq.Header.Set("Host", r.Host)
+	proxyReq.Header.Set("X-Forwarded-For", r.RemoteAddr)
+
+	for header, values := range r.Header {
+		for _, value := range values {
+			proxyReq.Header.Add(header, value)
+		}
+	}
+
+	proxyResponse, postErr := client.Do(proxyReq)
+
+	if postErr != nil {
+		glog.Errorf("post to filer: %v", postErr)
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+	defer util.CloseResponse(proxyResponse)
+
+	for k, v := range proxyResponse.Header {
+		w.Header()[k] = v
+	}
+	w.WriteHeader(proxyResponse.StatusCode)
+	io.Copy(w, proxyResponse.Body)
+
+}
@@ -159,10 +159,10 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request,
 	}

 	processRangeRequest(r, w, totalSize, mimeType, func(writer io.Writer, offset int64, size int64, httpStatusCode int) error {
+		if httpStatusCode != 0 {
+			w.WriteHeader(httpStatusCode)
+		}
 		if offset+size <= int64(len(entry.Content)) {
-			if httpStatusCode != 0 {
-				w.WriteHeader(httpStatusCode)
-			}
 			_, err := writer.Write(entry.Content[offset : offset+size])
 			return err
 		}
@@ -204,7 +204,7 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa

 	filerResult = &FilerPostResult{
 		Name: fileName,
-		Size: chunkOffset,
+		Size: int64(entry.FileSize),
 	}

 	if entry.Extended == nil {
@@ -523,7 +523,7 @@ func (f *WebDavFile) Read(p []byte) (readSize int, err error) {
 	}
 	if f.reader == nil {
 		chunkViews := filer.ViewFromVisibleIntervals(f.entryViewCache, 0, math.MaxInt64)
-		f.reader = filer.NewChunkReaderAtFromClient(f.fs, chunkViews, f.fs.chunkCache, fileSize)
+		f.reader = filer.NewChunkReaderAtFromClient(filer.LookupFn(f.fs), chunkViews, f.fs.chunkCache, fileSize)
 	}

 	readSize, err = f.reader.ReadAt(p, f.off)
@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/storage/needle"
 	"io"
+	"path/filepath"
 	"sort"

 	"github.com/chrislusf/seaweedfs/weed/operation"
@@ -19,6 +20,7 @@ func init() {
 }

 type commandVolumeFixReplication struct {
+	collectionPattern *string
 }

 func (c *commandVolumeFixReplication) Name() string {
@@ -33,12 +35,13 @@ func (c *commandVolumeFixReplication) Help() string {
 	This command also finds all under-replicated volumes, and finds volume servers with free slots.
 	If the free slots satisfy the replication requirement, the volume content is copied over and mounted.

-	volume.fix.replication -n # do not take action
-	volume.fix.replication    # actually deleting or copying the volume files and mount the volume
+	volume.fix.replication -n                             # do not take action
+	volume.fix.replication                                # actually deleting or copying the volume files and mount the volume
+	volume.fix.replication -collectionPattern=important*  # fix any collections with prefix "important"

 	Note:
-		* each time this will only add back one replica for one volume id. If there are multiple replicas
-		  are missing, e.g. multiple volume servers are new, you may need to run this multiple times.
+		* each time this will only add back one replica for each volume id that is under replicated.
+		  If there are multiple replicas are missing, e.g. replica count is > 2, you may need to run this multiple times.
 		* do not run this too quickly within seconds, since the new volume replica may take a few seconds 
 		  to register itself to the master.

@@ -52,6 +55,7 @@ func (c *commandVolumeFixReplication) Do(args []string, commandEnv *CommandEnv,
 	}

 	volFixReplicationCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	c.collectionPattern = volFixReplicationCommand.String("collectionPattern", "", "match with wildcard characters '*' and '?'")
 	skipChange := volFixReplicationCommand.Bool("n", false, "skip the changes")
 	if err = volFixReplicationCommand.Parse(args); err != nil {
 		return nil
@@ -127,6 +131,17 @@ func (c *commandVolumeFixReplication) fixOverReplicatedVolumes(commandEnv *Comma

 		replica := pickOneReplicaToDelete(replicas, replicaPlacement)

+		// check collection name pattern
+		if *c.collectionPattern != "" {
+			matched, err := filepath.Match(*c.collectionPattern, replica.info.Collection)
+			if err != nil {
+				return fmt.Errorf("match pattern %s with collection %s: %v", *c.collectionPattern, replica.info.Collection, err)
+			}
+			if !matched {
+				break
+			}
+		}
+
 		fmt.Fprintf(writer, "deleting volume %d from %s ...\n", replica.info.Id, replica.location.dataNode.Id)

 		if !takeAction {
@@ -147,9 +162,22 @@ func (c *commandVolumeFixReplication) fixUnderReplicatedVolumes(commandEnv *Comm
 		replica := pickOneReplicaToCopyFrom(replicas)
 		replicaPlacement, _ := super_block.NewReplicaPlacementFromByte(byte(replica.info.ReplicaPlacement))
 		foundNewLocation := false
+		hasSkippedCollection := false
 		for _, dst := range allLocations {
 			// check whether data nodes satisfy the constraints
 			if dst.dataNode.FreeVolumeCount > 0 && satisfyReplicaPlacement(replicaPlacement, replicas, dst) {
+				// check collection name pattern
+				if *c.collectionPattern != "" {
+					matched, err := filepath.Match(*c.collectionPattern, replica.info.Collection)
+					if err != nil {
+						return fmt.Errorf("match pattern %s with collection %s: %v", *c.collectionPattern, replica.info.Collection, err)
+					}
+					if !matched {
+						hasSkippedCollection = true
+						break
+					}
+				}
+
 				// ask the volume server to replicate the volume
 				foundNewLocation = true
 				fmt.Fprintf(writer, "replicating volume %d %s from %s to dataNode %s ...\n", replica.info.Id, replicaPlacement, replica.location.dataNode.Id, dst.dataNode.Id)
@@ -179,7 +207,7 @@ func (c *commandVolumeFixReplication) fixUnderReplicatedVolumes(commandEnv *Comm
 				break
 			}
 		}
-		if !foundNewLocation {
+		if !foundNewLocation && !hasSkippedCollection {
 			fmt.Fprintf(writer, "failed to place volume %d replica as %s, existing:%+v\n", replica.info.Id, replicaPlacement, len(replicas))
 		}

@@ -218,7 +218,7 @@ func (vl *VolumeLayout) ensureCorrectWritables(vid needle.VolumeId) {

 func (vl *VolumeLayout) isAllWritable(vid needle.VolumeId) bool {
 	for _, dn := range vl.vid2location[vid].list {
-		if v, found := dn.volumes[vid]; found {
+		if v, getError := dn.GetVolumesById(vid); getError == nil {
 			if v.ReadOnly {
 				return false
 			}
@@ -5,7 +5,7 @@ import (
 )

 var (
-	VERSION = fmt.Sprintf("%s %d.%02d", sizeLimit, 2, 21)
+	VERSION = fmt.Sprintf("%s %d.%02d", sizeLimit, 2, 23)
 	COMMIT  = ""
 )
Author	SHA1	Message	Date
Chris Lu	318a3d2efc	2.23	2021-01-31 20:18:06 -08:00
Chris Lu	4be51c0701	filer: leveldb and hbase may miss files when listing large directories more than 1024 fix https://github.com/chrislusf/seaweedfs/issues/1768	2021-01-31 20:11:44 -08:00
Chris Lu	1102ae32c4	fix concurrent map reads	2021-01-31 18:26:26 -08:00
Chris Lu	314dc1c957	filer: etcd store fix listing fix https://github.com/chrislusf/seaweedfs/issues/1767	2021-01-30 13:38:44 -08:00
Chris Lu	87d1bfa862	Hadoop Compatible FS: 1.5.7	2021-01-30 06:18:41 -08:00
Chris Lu	6f4aab51f9	refactoring SeaweedInputStream	2021-01-30 06:16:02 -08:00
Chris Lu	043c2d7960	refactoring SeaweedOutputStream	2021-01-30 05:39:09 -08:00
Chris Lu	678b9a60be	test different volumeServerAcccss	2021-01-30 05:38:31 -08:00
Chris Lu	d475c89fcc	go fmt	2021-01-28 15:23:46 -08:00
Chris Lu	19295600f9	mount: change option name to volumeServerAccess, with publicUrl and filerProxy modes	2021-01-28 15:23:16 -08:00
Chris Lu	990fa69bfe	add back AdjustedUrl() related code	2021-01-28 14:36:29 -08:00
Chris Lu	9292796ec2	fmt	2021-01-28 14:29:14 -08:00
Chris Lu	c2bf1a88ac	delete from the deepest directory first when checking empty folders	2021-01-28 14:28:40 -08:00
Chris Lu	e9d8201925	s3: batch purge empty folders	2021-01-28 13:20:06 -08:00
Chris Lu	cf252fc0cd	mount: report error when Flush() The error in Release() is not working. See https://github.com/jaderhs/libfuse/blob/master/FAQ related to https://github.com/chrislusf/seaweedfs/issues/1765	2021-01-28 04:46:37 -08:00
Chris Lu	9a06c35da4	replicate: incremental sink only contains new and updated files address https://github.com/chrislusf/seaweedfs/commit/da08402ba253fe5391f0152905f51f82c51fd527	2021-01-28 02:39:22 -08:00
Chris Lu	da08402ba2	replicate: use creation time for local incremental file sink related to https://github.com/chrislusf/seaweedfs/pull/1762	2021-01-28 02:17:41 -08:00
Chris LuandGitHub	822f1ade9d	Merge pull request #1762 from kmlebedev/backupsink replication to create time date directory	2021-01-28 02:03:33 -08:00
Konstantin Lebedev	be1062b7fc	rename in scaffold	2021-01-28 14:59:20 +05:00
Konstantin Lebedev	02fdc0a333	rename backup to local_incremental and use mtime	2021-01-28 14:56:13 +05:00
Chris LuandGitHub	5f72c388bc	Merge pull request #1764 from danlsgiga/handle-umask Use provided umask on mount	2021-01-28 01:09:20 -08:00
Daniel Santos	d38613c49a	match upstream	2021-01-27 20:56:41 -07:00
Chris Lu	22e167bc8a	mount: skip daemon timeout on FreeBSD	2021-01-27 15:34:43 -08:00
Chris Lu	7a96276f34	adjust dev local docker compose yaml	2021-01-27 10:29:09 -08:00
Chris Lu	b81956bcb5	mount: invalidate kernel cache when mounted to a filer path fix https://github.com/chrislusf/seaweedfs/issues/1752#issuecomment-768178422	2021-01-27 10:28:37 -08:00
Konstantin Lebedev	6b54ff9912	replication to create time date directory	2021-01-27 15:01:33 +05:00
Chris Lu	f20ec82a28	update help message see https://github.com/chrislusf/seaweedfs/issues/1758#issuecomment-768123224	2021-01-27 00:48:31 -08:00
Chris Lu	1045ab8b07	avoid error message	2021-01-27 00:21:59 -08:00
Chris Lu	0488c84bea	adjust help message	2021-01-27 00:17:52 -08:00
Chris LuandGitHub	2266b5c528	Merge pull request #1761 from kmlebedev/todaysDateFormat replication to todays date directory	2021-01-27 00:15:16 -08:00
Konstantin Lebedev	3634811408	replication to todays date directory	2021-01-27 12:45:58 +05:00
Chris LuandGitHub	5e07afb0f0	Merge pull request #1759 from kmlebedev/sink.local replication to local disk storage	2021-01-26 22:31:54 -08:00
Chris Lu	b3f66199db	shell: volume.fix.replication adds collectionPattern with wildcard characters fix https://github.com/chrislusf/seaweedfs/issues/1758	2021-01-26 22:30:39 -08:00
Chris LuandGitHub	2945a64fd6	Update issue templates	2021-01-26 22:03:22 -08:00
Daniel Santos	0c64d60fdb	Fix log message with correct mode	2021-01-26 21:45:52 -07:00
Daniel Santos	0aaaa9b470	Fix var assignment	2021-01-26 21:06:08 -07:00
Daniel Santos	5a761bb7a4	Enforce umask correctly	2021-01-26 20:42:10 -07:00
Chris Lu	ad2a20c8a5	notification add ack and nack	2021-01-26 11:08:44 -08:00
Konstantin Lebedev	612b7975a1	replication to local disk storage	2021-01-26 22:50:25 +05:00
Chris Lu	3a1d3d3413	mount: properly invalidate kernel node cache entry fix https://github.com/chrislusf/seaweedfs/issues/1752	2021-01-26 02:50:53 -08:00
Chris LuandGitHub	c41961d5cc	Update README.md	2021-01-25 10:44:16 -08:00
Chris LuandGitHub	ff35520a1a	Update README.md	2021-01-25 10:34:51 -08:00
Chris Lu	cc1f3907ff	filer.sync: default filerProxy to true	2021-01-25 10:14:42 -08:00
Chris Lu	62ebb917e0	mount: avoid the need to specify different cache directory for different mount fix https://github.com/seaweedfs/seaweedfs-csi-driver/issues/11	2021-01-25 09:10:20 -08:00
Chris Lu	17f7c1c43f	Merge branch 'mount_from_outside_cluster'	2021-01-24 19:03:02 -08:00
Chris Lu	00707ec00f	mount: outsideContainerClusterMode proxy through filer Running mount outside of the cluster would not need to expose all the volume servers to outside of the cluster. The chunk read and write will go through the filer.	2021-01-24 19:01:58 -08:00
Chris Lu	9bcb28a3ea	avoid possible nil reader fix https://github.com/chrislusf/seaweedfs/issues/1754	2021-01-24 17:19:35 -08:00
Chris Lu	2c5eac5705	2.22	2021-01-24 14:06:09 -08:00
Chris Lu	6ca10725b8	Revert "mount: when outside cluster network, use filer as proxy to access volume servers" This reverts commit `096e088d7b`.	2021-01-24 03:15:19 -08:00
Chris Lu	096e088d7b	mount: when outside cluster network, use filer as proxy to access volume servers	2021-01-24 01:41:38 -08:00
Chris Lu	ccbdb38c89	s3: fix for listing objects if more than 1000 in the folder many existing tools have max key set to 1000	2021-01-24 00:27:52 -08:00
Chris Lu	80b8692688	filer.sync: replicate outside of either cluster, only need to see filers	2021-01-24 00:01:44 -08:00
Chris Lu	6897f1bfbc	makefile: add env for 2 clusters	2021-01-24 00:00:42 -08:00
Chris Lu	f8dbb03bdd	filer: append operation returns final file size	2021-01-23 04:19:49 -08:00
Chris Lu	759482e2ac	filer: proxy all http methods to volume servers	2021-01-23 03:52:38 -08:00
Chris Lu	937cfacc01	filer: add "proxyToFileId" to reverse proxy to a volume server	2021-01-23 03:43:48 -08:00
Chris Lu	3ecd989500	add local mount for testing	2021-01-22 22:47:23 -08:00
Chris Lu	20ef3bb8d4	mount: use direct_io to avoid OS page cache fix https://github.com/chrislusf/seaweedfs/issues/1752	2021-01-22 22:39:46 -08:00
Chris Lu	711c3f3939	filer: fix http status 206 setting fix https://github.com/chrislusf/seaweedfs/issues/1753	2021-01-22 00:26:30 -08:00
Chris Lu	024c186a48	filer.meta.tail: output json format for easier parsing	2021-01-21 22:39:25 -08:00