distributedlog-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From si...@apache.org
Subject [17/23] incubator-distributedlog git commit: DL-124: Use Java8 Future rather than twitter Future
Date Wed, 21 Jun 2017 17:20:49 GMT
http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/53fca4ac/distributedlog-core/src/main/java/org/apache/distributedlog/BKLogHandler.java
----------------------------------------------------------------------
diff --git a/distributedlog-core/src/main/java/org/apache/distributedlog/BKLogHandler.java b/distributedlog-core/src/main/java/org/apache/distributedlog/BKLogHandler.java
index 3919e92..a7d0d25 100644
--- a/distributedlog-core/src/main/java/org/apache/distributedlog/BKLogHandler.java
+++ b/distributedlog-core/src/main/java/org/apache/distributedlog/BKLogHandler.java
@@ -18,6 +18,27 @@
 package org.apache.distributedlog;
 
 import com.google.common.base.Stopwatch;
+import com.google.common.collect.Lists;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+import org.apache.bookkeeper.stats.AlertStatsLogger;
+import org.apache.bookkeeper.stats.OpStatsLogger;
+import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.bookkeeper.versioning.Version;
+import org.apache.bookkeeper.versioning.Versioned;
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.distributedlog.callback.LogSegmentNamesListener;
 import org.apache.distributedlog.exceptions.LogEmptyException;
 import org.apache.distributedlog.exceptions.LogSegmentNotFoundException;
@@ -31,35 +52,12 @@ import org.apache.distributedlog.logsegment.PerStreamLogSegmentCache;
 import org.apache.distributedlog.logsegment.LogSegmentFilter;
 import org.apache.distributedlog.logsegment.LogSegmentMetadataStore;
 import org.apache.distributedlog.metadata.LogStreamMetadataStore;
-import org.apache.distributedlog.util.FutureUtils;
+import org.apache.distributedlog.common.concurrent.FutureEventListener;
+import org.apache.distributedlog.common.concurrent.FutureUtils;
 import org.apache.distributedlog.util.OrderedScheduler;
-import com.twitter.util.Function;
-import com.twitter.util.Future;
-import com.twitter.util.FutureEventListener;
-import com.twitter.util.Promise;
-import org.apache.bookkeeper.stats.AlertStatsLogger;
-import org.apache.bookkeeper.stats.OpStatsLogger;
-import org.apache.bookkeeper.stats.StatsLogger;
-import org.apache.bookkeeper.versioning.Version;
-import org.apache.bookkeeper.versioning.Versioned;
-import org.apache.commons.lang3.tuple.Pair;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReference;
-
 /**
  * The base class about log handler on managing log segments.
  *
@@ -171,25 +169,27 @@ abstract class BKLogHandler implements AsyncCloseable, AsyncAbortable {
         return lockClientId;
     }
 
-    public Future<LogRecordWithDLSN> asyncGetFirstLogRecord() {
-        final Promise<LogRecordWithDLSN> promise = new Promise<LogRecordWithDLSN>();
-        streamMetadataStore.logExists(logMetadata.getUri(), logMetadata.getLogName())
-                .addEventListener(new FutureEventListener<Void>() {
+    public CompletableFuture<LogRecordWithDLSN> asyncGetFirstLogRecord() {
+        final CompletableFuture<LogRecordWithDLSN> promise = new CompletableFuture<LogRecordWithDLSN>();
+        streamMetadataStore.logExists(
+            logMetadata.getUri(),
+            logMetadata.getLogName()
+        ).whenComplete(new FutureEventListener<Void>() {
             @Override
             public void onSuccess(Void value) {
                 readLogSegmentsFromStore(
                         LogSegmentMetadata.COMPARATOR,
                         LogSegmentFilter.DEFAULT_FILTER,
                         null
-                ).addEventListener(new FutureEventListener<Versioned<List<LogSegmentMetadata>>>() {
+                ).whenComplete(new FutureEventListener<Versioned<List<LogSegmentMetadata>>>() {
 
                     @Override
                     public void onSuccess(Versioned<List<LogSegmentMetadata>> ledgerList) {
                         if (ledgerList.getValue().isEmpty()) {
-                            promise.setException(new LogEmptyException("Log " + getFullyQualifiedName() + " has no records"));
+                            promise.completeExceptionally(new LogEmptyException("Log " + getFullyQualifiedName() + " has no records"));
                             return;
                         }
-                        Future<LogRecordWithDLSN> firstRecord = null;
+                        CompletableFuture<LogRecordWithDLSN> firstRecord = null;
                         for (LogSegmentMetadata ledger : ledgerList.getValue()) {
                             if (!ledger.isTruncated() && (ledger.getRecordCount() > 0 || ledger.isInProgress())) {
                                 firstRecord = asyncReadFirstUserRecord(ledger, DLSN.InitialDLSN);
@@ -197,43 +197,45 @@ abstract class BKLogHandler implements AsyncCloseable, AsyncAbortable {
                             }
                         }
                         if (null != firstRecord) {
-                            promise.become(firstRecord);
+                            FutureUtils.proxyTo(firstRecord, promise);
                         } else {
-                            promise.setException(new LogEmptyException("Log " + getFullyQualifiedName() + " has no records"));
+                            promise.completeExceptionally(new LogEmptyException("Log " + getFullyQualifiedName() + " has no records"));
                         }
                     }
 
                     @Override
                     public void onFailure(Throwable cause) {
-                        promise.setException(cause);
+                        promise.completeExceptionally(cause);
                     }
                 });
             }
 
             @Override
             public void onFailure(Throwable cause) {
-                promise.setException(cause);
+                promise.completeExceptionally(cause);
             }
         });
         return promise;
     }
 
-    public Future<LogRecordWithDLSN> getLastLogRecordAsync(final boolean recover, final boolean includeEndOfStream) {
-        final Promise<LogRecordWithDLSN> promise = new Promise<LogRecordWithDLSN>();
-        streamMetadataStore.logExists(logMetadata.getUri(), logMetadata.getLogName())
-                .addEventListener(new FutureEventListener<Void>() {
+    public CompletableFuture<LogRecordWithDLSN> getLastLogRecordAsync(final boolean recover, final boolean includeEndOfStream) {
+        final CompletableFuture<LogRecordWithDLSN> promise = new CompletableFuture<LogRecordWithDLSN>();
+        streamMetadataStore.logExists(
+            logMetadata.getUri(),
+            logMetadata.getLogName()
+        ).whenComplete(new FutureEventListener<Void>() {
             @Override
             public void onSuccess(Void value) {
                 readLogSegmentsFromStore(
                         LogSegmentMetadata.DESC_COMPARATOR,
                         LogSegmentFilter.DEFAULT_FILTER,
                         null
-                ).addEventListener(new FutureEventListener<Versioned<List<LogSegmentMetadata>>>() {
+                ).whenComplete(new FutureEventListener<Versioned<List<LogSegmentMetadata>>>() {
 
                     @Override
                     public void onSuccess(Versioned<List<LogSegmentMetadata>> ledgerList) {
                         if (ledgerList.getValue().isEmpty()) {
-                            promise.setException(
+                            promise.completeExceptionally(
                                     new LogEmptyException("Log " + getFullyQualifiedName() + " has no records"));
                             return;
                         }
@@ -247,49 +249,51 @@ abstract class BKLogHandler implements AsyncCloseable, AsyncAbortable {
 
                     @Override
                     public void onFailure(Throwable cause) {
-                        promise.setException(cause);
+                        promise.completeExceptionally(cause);
                     }
                 });
             }
 
             @Override
             public void onFailure(Throwable cause) {
-                promise.setException(cause);
+                promise.completeExceptionally(cause);
             }
         });
         return promise;
     }
 
     private void asyncGetLastLogRecord(final Iterator<LogSegmentMetadata> ledgerIter,
-                                       final Promise<LogRecordWithDLSN> promise,
+                                       final CompletableFuture<LogRecordWithDLSN> promise,
                                        final boolean fence,
                                        final boolean includeControlRecord,
                                        final boolean includeEndOfStream) {
         if (ledgerIter.hasNext()) {
             LogSegmentMetadata metadata = ledgerIter.next();
-            asyncReadLastRecord(metadata, fence, includeControlRecord, includeEndOfStream).addEventListener(
-                    new FutureEventListener<LogRecordWithDLSN>() {
-                        @Override
-                        public void onSuccess(LogRecordWithDLSN record) {
-                            if (null == record) {
-                                asyncGetLastLogRecord(ledgerIter, promise, fence, includeControlRecord, includeEndOfStream);
-                            } else {
-                                promise.setValue(record);
-                            }
-                        }
-
-                        @Override
-                        public void onFailure(Throwable cause) {
-                            promise.setException(cause);
+            asyncReadLastRecord(
+                metadata, fence, includeControlRecord, includeEndOfStream
+            ).whenComplete(
+                new FutureEventListener<LogRecordWithDLSN>() {
+                    @Override
+                    public void onSuccess(LogRecordWithDLSN record) {
+                        if (null == record) {
+                            asyncGetLastLogRecord(ledgerIter, promise, fence, includeControlRecord, includeEndOfStream);
+                        } else {
+                            promise.complete(record);
                         }
                     }
+
+                    @Override
+                    public void onFailure(Throwable cause) {
+                                                         promise.completeExceptionally(cause);
+                                                                                              }
+                }
             );
         } else {
-            promise.setException(new LogEmptyException("Log " + getFullyQualifiedName() + " has no records"));
+            promise.completeExceptionally(new LogEmptyException("Log " + getFullyQualifiedName() + " has no records"));
         }
     }
 
-    private Future<LogRecordWithDLSN> asyncReadFirstUserRecord(LogSegmentMetadata ledger, DLSN beginDLSN) {
+    private CompletableFuture<LogRecordWithDLSN> asyncReadFirstUserRecord(LogSegmentMetadata ledger, DLSN beginDLSN) {
         return ReadUtils.asyncReadFirstUserRecord(
                 getFullyQualifiedName(),
                 ledger,
@@ -307,15 +311,17 @@ abstract class BKLogHandler implements AsyncCloseable, AsyncAbortable {
      * beginDLSN and the second denoted by endPosition. Its up to the caller to ensure that endPosition refers to
      * position in the same ledger as beginDLSN.
      */
-    private Future<Long> asyncGetLogRecordCount(LogSegmentMetadata ledger, final DLSN beginDLSN, final long endPosition) {
-        return asyncReadFirstUserRecord(ledger, beginDLSN).map(new Function<LogRecordWithDLSN, Long>() {
-            public Long apply(final LogRecordWithDLSN beginRecord) {
-                long recordCount = 0;
-                if (null != beginRecord) {
-                    recordCount = endPosition + 1 - beginRecord.getLastPositionWithinLogSegment();
-                }
-                return recordCount;
+    private CompletableFuture<Long> asyncGetLogRecordCount(LogSegmentMetadata ledger,
+                                                           final DLSN beginDLSN,
+                                                           final long endPosition) {
+        return asyncReadFirstUserRecord(
+            ledger, beginDLSN
+        ).thenApply(beginRecord -> {
+            long recordCount = 0;
+            if (null != beginRecord) {
+                recordCount = endPosition + 1 - beginRecord.getLastPositionWithinLogSegment();
             }
+            return recordCount;
         });
     }
 
@@ -325,31 +331,29 @@ abstract class BKLogHandler implements AsyncCloseable, AsyncAbortable {
      * an interior entry. For the last entry, if it is inprogress, we need to recover it and find the last user
      * entry.
      */
-    private Future<Long> asyncGetLogRecordCount(final LogSegmentMetadata ledger, final DLSN beginDLSN) {
+    private CompletableFuture<Long> asyncGetLogRecordCount(final LogSegmentMetadata ledger, final DLSN beginDLSN) {
         if (ledger.isInProgress() && ledger.isDLSNinThisSegment(beginDLSN)) {
-            return asyncReadLastUserRecord(ledger).flatMap(new Function<LogRecordWithDLSN, Future<Long>>() {
-                public Future<Long> apply(final LogRecordWithDLSN endRecord) {
+            return asyncReadLastUserRecord(ledger).thenCompose(
+                (Function<LogRecordWithDLSN, CompletableFuture<Long>>) endRecord -> {
                     if (null != endRecord) {
-                        return asyncGetLogRecordCount(ledger, beginDLSN, endRecord.getLastPositionWithinLogSegment() /* end position */);
+                        return asyncGetLogRecordCount(
+                            ledger, beginDLSN, endRecord.getLastPositionWithinLogSegment() /* end position */);
                     } else {
-                        return Future.value((long) 0);
+                        return FutureUtils.value((long) 0);
                     }
-                }
-            });
+                });
         } else if (ledger.isInProgress()) {
-            return asyncReadLastUserRecord(ledger).map(new Function<LogRecordWithDLSN, Long>() {
-                public Long apply(final LogRecordWithDLSN endRecord) {
-                    if (null != endRecord) {
-                        return (long) endRecord.getLastPositionWithinLogSegment();
-                    } else {
-                        return (long) 0;
-                    }
+            return asyncReadLastUserRecord(ledger).thenApply(endRecord -> {
+                if (null != endRecord) {
+                    return (long) endRecord.getLastPositionWithinLogSegment();
+                } else {
+                    return (long) 0;
                 }
             });
         } else if (ledger.isDLSNinThisSegment(beginDLSN)) {
             return asyncGetLogRecordCount(ledger, beginDLSN, ledger.getRecordCount() /* end position */);
         } else {
-            return Future.value((long) ledger.getRecordCount());
+            return FutureUtils.value((long) ledger.getRecordCount());
         }
     }
 
@@ -359,29 +363,26 @@ abstract class BKLogHandler implements AsyncCloseable, AsyncAbortable {
      * @param beginDLSN dlsn marking the start of the range
      * @return the count of records present in the range
      */
-    public Future<Long> asyncGetLogRecordCount(final DLSN beginDLSN) {
+    public CompletableFuture<Long> asyncGetLogRecordCount(final DLSN beginDLSN) {
         return streamMetadataStore.logExists(logMetadata.getUri(), logMetadata.getLogName())
-                .flatMap(new Function<Void, Future<Long>>() {
-            public Future<Long> apply(Void done) {
+                .thenCompose(new Function<Void, CompletableFuture<Long>>() {
+            public CompletableFuture<Long> apply(Void done) {
 
                 return readLogSegmentsFromStore(
                         LogSegmentMetadata.COMPARATOR,
                         LogSegmentFilter.DEFAULT_FILTER,
                         null
-                ).flatMap(new Function<Versioned<List<LogSegmentMetadata>>, Future<Long>>() {
-                    public Future<Long> apply(Versioned<List<LogSegmentMetadata>> ledgerList) {
+                ).thenCompose(new Function<Versioned<List<LogSegmentMetadata>>, CompletableFuture<Long>>() {
+                    public CompletableFuture<Long> apply(Versioned<List<LogSegmentMetadata>> ledgerList) {
 
-                        List<Future<Long>> futureCounts = new ArrayList<Future<Long>>(ledgerList.getValue().size());
+                        List<CompletableFuture<Long>> futureCounts =
+                          Lists.newArrayListWithExpectedSize(ledgerList.getValue().size());
                         for (LogSegmentMetadata ledger : ledgerList.getValue()) {
                             if (ledger.getLogSegmentSequenceNumber() >= beginDLSN.getLogSegmentSequenceNo()) {
                                 futureCounts.add(asyncGetLogRecordCount(ledger, beginDLSN));
                             }
                         }
-                        return Future.collect(futureCounts).map(new Function<List<Long>, Long>() {
-                            public Long apply(List<Long> counts) {
-                                return sum(counts);
-                            }
-                        });
+                        return FutureUtils.collect(futureCounts).thenApply(counts -> sum(counts));
                     }
                 });
             }
@@ -397,15 +398,15 @@ abstract class BKLogHandler implements AsyncCloseable, AsyncAbortable {
     }
 
     @Override
-    public Future<Void> asyncAbort() {
+    public CompletableFuture<Void> asyncAbort() {
         return asyncClose();
     }
 
-    public Future<LogRecordWithDLSN> asyncReadLastUserRecord(final LogSegmentMetadata l) {
+    public CompletableFuture<LogRecordWithDLSN> asyncReadLastUserRecord(final LogSegmentMetadata l) {
         return asyncReadLastRecord(l, false, false, false);
     }
 
-    public Future<LogRecordWithDLSN> asyncReadLastRecord(final LogSegmentMetadata l,
+    public CompletableFuture<LogRecordWithDLSN> asyncReadLastRecord(final LogSegmentMetadata l,
                                                          final boolean fence,
                                                          final boolean includeControl,
                                                          final boolean includeEndOfStream) {
@@ -422,7 +423,7 @@ abstract class BKLogHandler implements AsyncCloseable, AsyncAbortable {
                 numRecordsScanned,
                 scheduler,
                 entryStore
-        ).addEventListener(new FutureEventListener<LogRecordWithDLSN>() {
+        ).whenComplete(new FutureEventListener<LogRecordWithDLSN>() {
             @Override
             public void onSuccess(LogRecordWithDLSN value) {
                 recoverLastEntryStats.registerSuccessfulEvent(stopwatch.stop().elapsed(TimeUnit.MICROSECONDS));
@@ -572,17 +573,17 @@ abstract class BKLogHandler implements AsyncCloseable, AsyncAbortable {
      * @param logSegmentNamesListener
      * @return future represents the result of log segments
      */
-    public Future<Versioned<List<LogSegmentMetadata>>> readLogSegmentsFromStore(
+    public CompletableFuture<Versioned<List<LogSegmentMetadata>>> readLogSegmentsFromStore(
             final Comparator<LogSegmentMetadata> comparator,
             final LogSegmentFilter segmentFilter,
             final LogSegmentNamesListener logSegmentNamesListener) {
-        final Promise<Versioned<List<LogSegmentMetadata>>> readResult =
-                new Promise<Versioned<List<LogSegmentMetadata>>>();
+        final CompletableFuture<Versioned<List<LogSegmentMetadata>>> readResult =
+                new CompletableFuture<Versioned<List<LogSegmentMetadata>>>();
         metadataStore.getLogSegmentNames(logMetadata.getLogSegmentsPath(), logSegmentNamesListener)
-                .addEventListener(new FutureEventListener<Versioned<List<String>>>() {
+                .whenComplete(new FutureEventListener<Versioned<List<String>>>() {
                     @Override
                     public void onFailure(Throwable cause) {
-                        FutureUtils.setException(readResult, cause);
+                        readResult.completeExceptionally(cause);
                     }
 
                     @Override
@@ -596,7 +597,7 @@ abstract class BKLogHandler implements AsyncCloseable, AsyncAbortable {
     protected void readLogSegmentsFromStore(final Versioned<List<String>> logSegmentNames,
                                             final Comparator<LogSegmentMetadata> comparator,
                                             final LogSegmentFilter segmentFilter,
-                                            final Promise<Versioned<List<LogSegmentMetadata>>> readResult) {
+                                            final CompletableFuture<Versioned<List<LogSegmentMetadata>>> readResult) {
         Set<String> segmentsReceived = new HashSet<String>();
         segmentsReceived.addAll(segmentFilter.filter(logSegmentNames.getValue()));
         Set<String> segmentsAdded;
@@ -619,12 +620,11 @@ abstract class BKLogHandler implements AsyncCloseable, AsyncAbortable {
             try {
                 segmentList = getCachedLogSegments(comparator);
             } catch (UnexpectedException e) {
-                FutureUtils.setException(readResult, e);
+                readResult.completeExceptionally(e);
                 return;
             }
 
-            FutureUtils.setValue(readResult,
-                    new Versioned<List<LogSegmentMetadata>>(segmentList, logSegmentNames.getVersion()));
+            readResult.complete(new Versioned<List<LogSegmentMetadata>>(segmentList, logSegmentNames.getVersion()));
             return;
         }
 
@@ -646,7 +646,7 @@ abstract class BKLogHandler implements AsyncCloseable, AsyncAbortable {
                 continue;
             }
             metadataStore.getLogSegment(logSegmentPath)
-                    .addEventListener(new FutureEventListener<LogSegmentMetadata>() {
+                    .whenComplete(new FutureEventListener<LogSegmentMetadata>() {
 
                         @Override
                         public void onSuccess(LogSegmentMetadata result) {
@@ -666,7 +666,7 @@ abstract class BKLogHandler implements AsyncCloseable, AsyncAbortable {
                             } else {
                                 // fail fast
                                 if (1 == numFailures.incrementAndGet()) {
-                                    FutureUtils.setException(readResult, cause);
+                                    readResult.completeExceptionally(cause);
                                     return;
                                 }
                             }
@@ -689,7 +689,7 @@ abstract class BKLogHandler implements AsyncCloseable, AsyncAbortable {
     private void completeReadLogSegmentsFromStore(final Set<String> removedSegments,
                                                   final Map<String, LogSegmentMetadata> addedSegments,
                                                   final Comparator<LogSegmentMetadata> comparator,
-                                                  final Promise<Versioned<List<LogSegmentMetadata>>> readResult,
+                                                  final CompletableFuture<Versioned<List<LogSegmentMetadata>>> readResult,
                                                   final Version logSegmentNamesVersion,
                                                   final AtomicInteger numChildren,
                                                   final AtomicInteger numFailures) {
@@ -705,11 +705,10 @@ abstract class BKLogHandler implements AsyncCloseable, AsyncAbortable {
         try {
             segmentList = getCachedLogSegments(comparator);
         } catch (UnexpectedException e) {
-            FutureUtils.setException(readResult, e);
+            readResult.completeExceptionally(e);
             return;
         }
-        FutureUtils.setValue(readResult,
-            new Versioned<List<LogSegmentMetadata>>(segmentList, logSegmentNamesVersion));
+        readResult.complete(new Versioned<List<LogSegmentMetadata>>(segmentList, logSegmentNamesVersion));
     }
 
 }

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/53fca4ac/distributedlog-core/src/main/java/org/apache/distributedlog/BKLogReadHandler.java
----------------------------------------------------------------------
diff --git a/distributedlog-core/src/main/java/org/apache/distributedlog/BKLogReadHandler.java b/distributedlog-core/src/main/java/org/apache/distributedlog/BKLogReadHandler.java
index c6e2e07..ff6b527 100644
--- a/distributedlog-core/src/main/java/org/apache/distributedlog/BKLogReadHandler.java
+++ b/distributedlog-core/src/main/java/org/apache/distributedlog/BKLogReadHandler.java
@@ -17,15 +17,22 @@
  */
 package org.apache.distributedlog;
 
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Optional;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.TimeUnit;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Optional;
+import javax.annotation.Nullable;
+import org.apache.bookkeeper.stats.AlertStatsLogger;
+import org.apache.bookkeeper.stats.NullStatsLogger;
+import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.bookkeeper.versioning.Version;
+import org.apache.bookkeeper.versioning.Versioned;
 import org.apache.distributedlog.callback.LogSegmentListener;
 import org.apache.distributedlog.callback.LogSegmentNamesListener;
 import org.apache.distributedlog.config.DynamicDistributedLogConfiguration;
@@ -40,29 +47,12 @@ import org.apache.distributedlog.lock.DistributedLock;
 import org.apache.distributedlog.logsegment.LogSegmentFilter;
 import org.apache.distributedlog.logsegment.LogSegmentMetadataCache;
 import org.apache.distributedlog.metadata.LogStreamMetadataStore;
-import org.apache.distributedlog.util.FutureUtils;
+import org.apache.distributedlog.common.concurrent.FutureEventListener;
+import org.apache.distributedlog.common.concurrent.FutureUtils;
 import org.apache.distributedlog.util.OrderedScheduler;
 import org.apache.distributedlog.util.Utils;
-import com.twitter.util.ExceptionalFunction;
-import com.twitter.util.Function;
-import com.twitter.util.Future;
-import com.twitter.util.FutureEventListener;
-import com.twitter.util.Promise;
-import com.twitter.util.Return;
-import com.twitter.util.Throw;
-import com.twitter.util.Try;
-import org.apache.bookkeeper.stats.AlertStatsLogger;
-import org.apache.bookkeeper.stats.NullStatsLogger;
-import org.apache.bookkeeper.stats.StatsLogger;
-import org.apache.bookkeeper.util.SafeRunnable;
-import org.apache.bookkeeper.versioning.Version;
-import org.apache.bookkeeper.versioning.Versioned;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import scala.runtime.AbstractFunction1;
-import scala.runtime.BoxedUnit;
-
-import javax.annotation.Nullable;
 
 /**
  * Log Handler for Readers.
@@ -112,7 +102,7 @@ class BKLogReadHandler extends BKLogHandler implements LogSegmentNamesListener {
 
     private final Optional<String> subscriberId;
     private DistributedLock readLock;
-    private Future<Void> lockAcquireFuture;
+    private CompletableFuture<Void> lockAcquireFuture;
 
     // notify the state change about the read handler
     protected final AsyncNotification readerStateNotification;
@@ -166,31 +156,23 @@ class BKLogReadHandler extends BKLogHandler implements LogSegmentNamesListener {
         return logMetadataForReader.getReadLockPath(subscriberId);
     }
 
-    <T> void satisfyPromiseAsync(final Promise<T> promise, final Try<T> result) {
-        scheduler.submit(new SafeRunnable() {
-            @Override
-            public void safeRun() {
-                promise.update(result);
-            }
-        });
-    }
-
-    Future<Void> checkLogStreamExists() {
+    CompletableFuture<Void> checkLogStreamExists() {
         return streamMetadataStore.logExists(logMetadata.getUri(), logMetadata.getLogName());
     }
 
     /**
      * Elective stream lock--readers are not required to acquire the lock before using the stream.
      */
-    synchronized Future<Void> lockStream() {
+    synchronized CompletableFuture<Void> lockStream() {
         if (null == lockAcquireFuture) {
             lockAcquireFuture = streamMetadataStore.createReadLock(logMetadataForReader, subscriberId)
-                    .flatMap(new ExceptionalFunction<DistributedLock, Future<Void>>() {
-                        @Override
-                        public Future<Void> applyE(DistributedLock lock) throws Throwable {
+                    .thenCompose(lock -> {
+                        try {
                             BKLogReadHandler.this.readLock = lock;
                             LOG.info("acquiring readlock {} at {}", getLockClientId(), getReadLockPath());
                             return acquireLockOnExecutorThread(lock);
+                        } catch (LockingException le) {
+                            return FutureUtils.exception(le);
                         }
                     });
         }
@@ -201,33 +183,31 @@ class BKLogReadHandler extends BKLogHandler implements LogSegmentNamesListener {
      * Begin asynchronous lock acquire, but ensure that the returned future is satisfied on an
      * executor service thread.
      */
-    Future<Void> acquireLockOnExecutorThread(DistributedLock lock) throws LockingException {
-        final Future<? extends DistributedLock> acquireFuture = lock.asyncAcquire();
+    CompletableFuture<Void> acquireLockOnExecutorThread(DistributedLock lock) throws LockingException {
+        final CompletableFuture<? extends DistributedLock> acquireFuture = lock.asyncAcquire();
 
         // The future we return must be satisfied on an executor service thread. If we simply
         // return the future returned by asyncAcquire, user callbacks may end up running in
         // the lock state executor thread, which will cause deadlocks and introduce latency
         // etc.
-        final Promise<Void> threadAcquirePromise = new Promise<Void>();
-        threadAcquirePromise.setInterruptHandler(new Function<Throwable, BoxedUnit>() {
-            @Override
-            public BoxedUnit apply(Throwable t) {
-                FutureUtils.cancel(acquireFuture);
-                return null;
+        final CompletableFuture<Void> threadAcquirePromise = new CompletableFuture<Void>();
+        threadAcquirePromise.whenComplete((value, cause) -> {
+            if (cause instanceof CancellationException) {
+                acquireFuture.cancel(true);
             }
         });
-        acquireFuture.addEventListener(new FutureEventListener<DistributedLock>() {
+        acquireFuture.whenCompleteAsync(new FutureEventListener<DistributedLock>() {
             @Override
             public void onSuccess(DistributedLock lock) {
                 LOG.info("acquired readlock {} at {}", getLockClientId(), getReadLockPath());
-                satisfyPromiseAsync(threadAcquirePromise, new Return<Void>(null));
+                threadAcquirePromise.complete(null);
             }
 
             @Override
             public void onFailure(Throwable cause) {
                 LOG.info("failed to acquire readlock {} at {}",
                         new Object[]{ getLockClientId(), getReadLockPath(), cause });
-                satisfyPromiseAsync(threadAcquirePromise, new Throw<Void>(cause));
+                threadAcquirePromise.completeExceptionally(cause);
             }
         });
         return threadAcquirePromise;
@@ -239,7 +219,7 @@ class BKLogReadHandler extends BKLogHandler implements LogSegmentNamesListener {
     void checkReadLock() throws DLIllegalStateException, LockingException {
         synchronized (this) {
             if ((null == lockAcquireFuture) ||
-                (!lockAcquireFuture.isDefined())) {
+                (!lockAcquireFuture.isDone())) {
                 throw new DLIllegalStateException("Attempt to check for lock before it has been acquired successfully");
             }
         }
@@ -247,27 +227,24 @@ class BKLogReadHandler extends BKLogHandler implements LogSegmentNamesListener {
         readLock.checkOwnership();
     }
 
-    public Future<Void> asyncClose() {
+    public CompletableFuture<Void> asyncClose() {
         DistributedLock lockToClose;
         synchronized (this) {
-            if (null != lockAcquireFuture && !lockAcquireFuture.isDefined()) {
-                FutureUtils.cancel(lockAcquireFuture);
+            if (null != lockAcquireFuture && !lockAcquireFuture.isDone()) {
+                lockAcquireFuture.cancel(true);
             }
             lockToClose = readLock;
         }
         return Utils.closeSequence(scheduler, lockToClose)
-                .flatMap(new AbstractFunction1<Void, Future<Void>>() {
-            @Override
-            public Future<Void> apply(Void result) {
+            .thenApply((value) -> {
                 // unregister the log segment listener
                 metadataStore.unregisterLogSegmentListener(logMetadata.getLogSegmentsPath(), BKLogReadHandler.this);
-                return Future.Void();
-            }
-        });
+                return null;
+            });
     }
 
     @Override
-    public Future<Void> asyncAbort() {
+    public CompletableFuture<Void> asyncAbort() {
         return asyncClose();
     }
 
@@ -277,18 +254,18 @@ class BKLogReadHandler extends BKLogHandler implements LogSegmentNamesListener {
      *
      * @return future represents the fetch result
      */
-    Future<Versioned<List<LogSegmentMetadata>>> asyncStartFetchLogSegments() {
-        Promise<Versioned<List<LogSegmentMetadata>>> promise =
-                new Promise<Versioned<List<LogSegmentMetadata>>>();
+    CompletableFuture<Versioned<List<LogSegmentMetadata>>> asyncStartFetchLogSegments() {
+        CompletableFuture<Versioned<List<LogSegmentMetadata>>> promise =
+                new CompletableFuture<Versioned<List<LogSegmentMetadata>>>();
         asyncStartFetchLogSegments(promise);
         return promise;
     }
 
-    void asyncStartFetchLogSegments(final Promise<Versioned<List<LogSegmentMetadata>>> promise) {
+    void asyncStartFetchLogSegments(final CompletableFuture<Versioned<List<LogSegmentMetadata>>> promise) {
         readLogSegmentsFromStore(
                 LogSegmentMetadata.COMPARATOR,
                 LogSegmentFilter.DEFAULT_FILTER,
-                this).addEventListener(new FutureEventListener<Versioned<List<LogSegmentMetadata>>>() {
+                this).whenComplete(new FutureEventListener<Versioned<List<LogSegmentMetadata>>>() {
             @Override
             public void onFailure(Throwable cause) {
                 if (cause instanceof LogNotFoundException ||
@@ -298,7 +275,7 @@ class BKLogReadHandler extends BKLogHandler implements LogSegmentNamesListener {
                     metadataException.compareAndSet(null, (IOException) cause);
                     // notify the reader that read handler is in error state
                     notifyReaderOnError(cause);
-                    FutureUtils.setException(promise, cause);
+                    FutureUtils.completeExceptionally(promise, cause);
                     return;
                 }
                 scheduler.schedule(new Runnable() {
@@ -312,7 +289,7 @@ class BKLogReadHandler extends BKLogHandler implements LogSegmentNamesListener {
             @Override
             public void onSuccess(Versioned<List<LogSegmentMetadata>> segments) {
                 // no-op
-                FutureUtils.setValue(promise, segments);
+                FutureUtils.complete(promise, segments);
             }
         });
     }
@@ -332,9 +309,9 @@ class BKLogReadHandler extends BKLogHandler implements LogSegmentNamesListener {
             }
         }
 
-        Promise<Versioned<List<LogSegmentMetadata>>> readLogSegmentsPromise =
-                new Promise<Versioned<List<LogSegmentMetadata>>>();
-        readLogSegmentsPromise.addEventListener(new FutureEventListener<Versioned<List<LogSegmentMetadata>>>() {
+        CompletableFuture<Versioned<List<LogSegmentMetadata>>> readLogSegmentsPromise =
+                new CompletableFuture<Versioned<List<LogSegmentMetadata>>>();
+        readLogSegmentsPromise.whenComplete(new FutureEventListener<Versioned<List<LogSegmentMetadata>>>() {
             @Override
             public void onFailure(Throwable cause) {
                 if (cause instanceof LogNotFoundException ||

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/53fca4ac/distributedlog-core/src/main/java/org/apache/distributedlog/BKLogSegmentWriter.java
----------------------------------------------------------------------
diff --git a/distributedlog-core/src/main/java/org/apache/distributedlog/BKLogSegmentWriter.java b/distributedlog-core/src/main/java/org/apache/distributedlog/BKLogSegmentWriter.java
index 6b60c77..a4016c8 100644
--- a/distributedlog-core/src/main/java/org/apache/distributedlog/BKLogSegmentWriter.java
+++ b/distributedlog-core/src/main/java/org/apache/distributedlog/BKLogSegmentWriter.java
@@ -20,6 +20,8 @@ package org.apache.distributedlog;
 import java.io.IOException;
 import java.util.List;
 import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -28,6 +30,7 @@ import java.util.concurrent.locks.ReentrantLock;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Stopwatch;
+import java.util.function.Function;
 import org.apache.distributedlog.config.DynamicDistributedLogConfiguration;
 import org.apache.distributedlog.exceptions.BKTransmitException;
 import org.apache.distributedlog.exceptions.EndOfStreamException;
@@ -47,20 +50,15 @@ import org.apache.distributedlog.io.CompressionUtils;
 import org.apache.distributedlog.lock.DistributedLock;
 import org.apache.distributedlog.logsegment.LogSegmentEntryWriter;
 import org.apache.distributedlog.logsegment.LogSegmentWriter;
-import org.apache.distributedlog.stats.BroadCastStatsLogger;
-import org.apache.distributedlog.stats.OpStatsListener;
+import org.apache.distributedlog.common.stats.BroadCastStatsLogger;
+import org.apache.distributedlog.common.stats.OpStatsListener;
 import org.apache.distributedlog.util.FailpointUtils;
-import org.apache.distributedlog.util.FutureUtils;
+import org.apache.distributedlog.common.concurrent.FutureEventListener;
+import org.apache.distributedlog.common.concurrent.FutureUtils;
 import org.apache.distributedlog.util.OrderedScheduler;
-import org.apache.distributedlog.util.PermitLimiter;
-import org.apache.distributedlog.util.SafeQueueingFuturePool;
+import org.apache.distributedlog.common.util.PermitLimiter;
 import org.apache.distributedlog.util.SimplePermitLimiter;
-import org.apache.distributedlog.util.Sizable;
-import com.twitter.util.Function0;
-import com.twitter.util.Future;
-import com.twitter.util.FutureEventListener;
-import com.twitter.util.FuturePool;
-import com.twitter.util.Promise;
+import org.apache.distributedlog.common.util.Sizable;
 import org.apache.bookkeeper.client.AsyncCallback.AddCallback;
 import org.apache.bookkeeper.client.AsyncCallback.CloseCallback;
 import org.apache.bookkeeper.client.BKException;
@@ -73,10 +71,9 @@ import org.apache.bookkeeper.stats.Gauge;
 import org.apache.bookkeeper.stats.OpStatsLogger;
 import org.apache.bookkeeper.stats.StatsLogger;
 import org.apache.bookkeeper.util.MathUtils;
+import org.apache.distributedlog.util.Utils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import scala.runtime.AbstractFunction1;
-import scala.runtime.BoxedUnit;
 
 import static com.google.common.base.Charsets.UTF_8;
 import static org.apache.distributedlog.LogRecord.MAX_LOGRECORD_SIZE;
@@ -146,7 +143,7 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
     final private AtomicReference<ScheduledFuture<?>> immFlushSchedFutureRef = new AtomicReference<ScheduledFuture<?>>(null);
     final private AtomicReference<Exception> scheduledFlushException = new AtomicReference<Exception>(null);
     private boolean enforceLock = true;
-    private Promise<Void> closeFuture = null;
+    private CompletableFuture<Void> closeFuture = null;
     private final boolean enableRecordCounts;
     private int positionWithinLogSegment = 0;
     private final long logSegmentSequenceNumber;
@@ -170,28 +167,17 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
     private final OpStatsLogger addCompleteDeferredTime;
     private final Counter pendingWrites;
 
-    // add complete processing
-    private final SafeQueueingFuturePool<Void> addCompleteFuturePool;
-
     // Functions
-    private final AbstractFunction1<Integer, Future<Long>> GET_LAST_TXID_ACKNOWLEDGED_AFTER_TRANSMIT_FUNC =
-            new AbstractFunction1<Integer, Future<Long>>() {
-                @Override
-                public Future<Long> apply(Integer transmitRc) {
-                    if (BKException.Code.OK == transmitRc) {
-                        return Future.value(getLastTxIdAcknowledged());
-                    } else {
-                        return Future.exception(new BKTransmitException("Failed to transmit entry", transmitRc));
-                    }
-                }
-            };
-    final AbstractFunction1<Long, Future<Long>> COMMIT_AFTER_FLUSH_FUNC =
-            new AbstractFunction1<Long, Future<Long>>() {
-                @Override
-                public Future<Long> apply(Long lastAckedTxId) {
-                    return commit();
-                }
-            };
+    private final Function<Integer, CompletableFuture<Long>> GET_LAST_TXID_ACKNOWLEDGED_AFTER_TRANSMIT_FUNC =
+        transmitRc -> {
+            if (BKException.Code.OK == transmitRc) {
+                return FutureUtils.value(getLastTxIdAcknowledged());
+            } else {
+                return FutureUtils.exception(new BKTransmitException("Failed to transmit entry", transmitRc));
+            }
+        };
+    final Function<Long, CompletableFuture<Long>> COMMIT_AFTER_FLUSH_FUNC =
+        lastAckedTxId -> commit();
 
     private final AlertStatsLogger alertStatsLogger;
     private final WriteLimiter writeLimiter;
@@ -341,11 +327,6 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
         }
 
         this.conf = conf;
-        if (null != scheduler) {
-            this.addCompleteFuturePool = new SafeQueueingFuturePool<Void>(scheduler.getFuturePool(streamName));
-        } else {
-            this.addCompleteFuturePool = null;
-        }
         assert(!this.immediateFlushEnabled || (null != this.scheduler));
         this.lastTransmit = Stopwatch.createStarted();
     }
@@ -360,11 +341,8 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
     }
 
     @VisibleForTesting
-    FuturePool getFuturePool() {
-        if (null == scheduler) {
-            return null;
-        }
-        return scheduler.getFuturePool(streamName);
+    ScheduledExecutorService getFuturePool() {
+        return scheduler.chooseExecutor(streamName);
     }
 
     @VisibleForTesting
@@ -471,21 +449,15 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
     }
 
     @Override
-    public Future<Void> asyncClose() {
+    public CompletableFuture<Void> asyncClose() {
         return closeInternal(false);
     }
 
     @Override
-    public Future<Void> asyncAbort() {
+    public CompletableFuture<Void> asyncAbort() {
         return closeInternal(true);
     }
 
-    private void flushAddCompletes() {
-        if (null != addCompleteFuturePool) {
-            addCompleteFuturePool.close();
-        }
-    }
-
     private synchronized void abortPacket(BKTransmitPacket packet) {
         long numRecords = 0;
         if (null != packet) {
@@ -495,7 +467,7 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
             if (BKException.Code.OK == rc) {
                 rc = BKException.Code.InterruptedException;
             }
-            Throwable reason = new WriteCancelledException(streamName, FutureUtils.transmitException(rc));
+            Throwable reason = new WriteCancelledException(streamName, Utils.transmitException(rc));
             recordSet.abortTransmit(reason);
         }
         LOG.info("Stream {} aborted {} writes", fullyQualifiedLogSegment, numRecords);
@@ -509,21 +481,13 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
         }
     }
 
-    private synchronized long getPendingAddCompleteCount() {
-        if (null != addCompleteFuturePool) {
-            return addCompleteFuturePool.size();
-        } else {
-            return 0;
-        }
-    }
-
-    private Future<Void> closeInternal(boolean abort) {
-        Promise<Void> closePromise;
+    private CompletableFuture<Void> closeInternal(boolean abort) {
+        CompletableFuture<Void> closePromise;
         synchronized (this) {
             if (null != closeFuture) {
                 return closeFuture;
             }
-            closePromise = closeFuture = new Promise<Void>();
+            closePromise = closeFuture = new CompletableFuture<Void>();
         }
 
         AtomicReference<Throwable> throwExc = new AtomicReference<Throwable>(null);
@@ -533,7 +497,7 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
 
     private void closeInternal(final boolean abort,
                                final AtomicReference<Throwable> throwExc,
-                               final Promise<Void> closePromise) {
+                               final CompletableFuture<Void> closePromise) {
         // clean stats resources
         this.transmitOutstandingLogger.unregisterGauge("requests", transmitOutstandingGauge);
         this.writeLimiter.close();
@@ -560,7 +524,7 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
         if (!abort && !isLogSegmentInError()) {
             this.enforceLock = false;
             LOG.info("Flushing before closing log segment {}", getFullyQualifiedLogSegment());
-            flushAndCommit().addEventListener(new FutureEventListener<Long>() {
+            flushAndCommit().whenComplete(new FutureEventListener<Long>() {
                 @Override
                 public void onSuccess(Long value) {
                     abortTransmitPacketOnClose(abort, throwExc, closePromise);
@@ -580,11 +544,11 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
 
     private void abortTransmitPacketOnClose(final boolean abort,
                                             final AtomicReference<Throwable> throwExc,
-                                            final Promise<Void> closePromise) {
+                                            final CompletableFuture<Void> closePromise) {
         LOG.info("Closing BKPerStreamLogWriter (abort={}) for {} :" +
-                        " lastDLSN = {} outstandingTransmits = {} writesPendingTransmit = {} addCompletesPending = {}",
+                        " lastDLSN = {} outstandingTransmits = {} writesPendingTransmit = {}",
                 new Object[]{abort, fullyQualifiedLogSegment, getLastDLSN(),
-                        outstandingTransmits.get(), getWritesPendingTransmit(), getPendingAddCompleteCount()});
+                        outstandingTransmits.get(), getWritesPendingTransmit()});
 
         // Save the current packet to reset, leave a new empty packet to avoid a race with
         // addCompleteDeferredProcessing.
@@ -602,7 +566,6 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
             packetPreviousSaved.addTransmitCompleteListener(new FutureEventListener<Integer>() {
                 @Override
                 public void onSuccess(Integer transmitResult) {
-                    flushAddCompletes();
                     abortPacket(packetCurrentSaved);
                 }
                 @Override
@@ -620,7 +583,7 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
 
     private void closeLedgerOnClose(final boolean abort,
                                     final AtomicReference<Throwable> throwExc,
-                                    final Promise<Void> closePromise) {
+                                    final CompletableFuture<Void> closePromise) {
         // close the log segment if it isn't in error state, so all the outstanding addEntry(s) will callback.
         if (null == throwExc.get() && !isLogSegmentInError()) {
             // Synchronous closing the ledger handle, if we couldn't close a ledger handle successfully.
@@ -644,16 +607,16 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
 
     private void completeClosePromise(final boolean abort,
                                       final AtomicReference<Throwable> throwExc,
-                                      final Promise<Void> closePromise) {
+                                      final CompletableFuture<Void> closePromise) {
         // If add entry failed because of closing ledger above, we don't need to fail the close operation
         if (!abort && null == throwExc.get() && shouldFailCompleteLogSegment()) {
             throwExc.set(new BKTransmitException("Closing an errored stream : ", transmitResult.get()));
         }
 
         if (null == throwExc.get()) {
-            FutureUtils.setValue(closePromise, null);
+            FutureUtils.complete(closePromise, null);
         } else {
-            FutureUtils.setException(closePromise, throwExc.get());
+            FutureUtils.completeExceptionally(closePromise, throwExc.get());
         }
     }
 
@@ -664,12 +627,12 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
     }
 
     @Override
-    synchronized public Future<DLSN> asyncWrite(LogRecord record) {
+    synchronized public CompletableFuture<DLSN> asyncWrite(LogRecord record) {
         return asyncWrite(record, true);
     }
 
-    synchronized public Future<DLSN> asyncWrite(LogRecord record, boolean flush) {
-        Future<DLSN> result = null;
+    synchronized public CompletableFuture<DLSN> asyncWrite(LogRecord record, boolean flush) {
+        CompletableFuture<DLSN> result = null;
         try {
             if (record.isControl()) {
                 // we don't pack control records with user records together
@@ -677,7 +640,7 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
                 try {
                     transmit();
                 } catch (IOException ioe) {
-                    return Future.exception(new WriteCancelledException(fullyQualifiedLogSegment, ioe));
+                    return FutureUtils.exception(new WriteCancelledException(fullyQualifiedLogSegment, ioe));
                 }
                 result = writeControlLogRecord(record);
                 transmit();
@@ -685,7 +648,7 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
                 result = writeUserRecord(record);
                 if (!isDurableWriteEnabled) {
                     // we have no idea about the DLSN if durability is turned off.
-                    result = Future.value(DLSN.InvalidDLSN);
+                    result = FutureUtils.value(DLSN.InvalidDLSN);
                 }
                 if (flush) {
                     flushIfNeeded();
@@ -697,7 +660,7 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
             if (null != result) {
                 LOG.error("Overriding first result with flush failure {}", result);
             }
-            result = Future.exception(ioe);
+            result = FutureUtils.exception(ioe);
 
             // Flush to ensure any prev. writes with flush=false are flushed despite failure.
             flushIfNeededNoThrow();
@@ -705,7 +668,7 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
         return result;
     }
 
-    synchronized private Future<DLSN> writeUserRecord(LogRecord record) throws IOException {
+    synchronized private CompletableFuture<DLSN> writeUserRecord(LogRecord record) throws IOException {
         if (null != closeFuture) {
             throw new WriteException(fullyQualifiedLogSegment, BKException.getMessage(BKException.Code.LedgerClosedException));
         }
@@ -737,7 +700,7 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
         // Internally generated log records don't increment the count
         // writeInternal will always set a count regardless of whether it was
         // incremented or not.
-        Future<DLSN> future = null;
+        CompletableFuture<DLSN> future = null;
         try {
             // increment the position for the record to write
             // if the record is failed to write, it would be decremented.
@@ -759,12 +722,9 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
         }
 
         // Track outstanding requests and return the future.
-        return future.ensure(new Function0<BoxedUnit>() {
-            public BoxedUnit apply() {
-                pendingWrites.dec();
-                writeLimiter.release();
-                return null;
-            }
+        return FutureUtils.ensure(future, () -> {
+            pendingWrites.dec();
+            writeLimiter.release();
         });
     }
 
@@ -777,7 +737,7 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
                 (transmitResult.get() != BKException.Code.LedgerClosedException);
     }
 
-    synchronized public Future<DLSN> writeInternal(LogRecord record)
+    synchronized public CompletableFuture<DLSN> writeInternal(LogRecord record)
             throws LogRecordTooLongException, LockingException, BKTransmitException,
                    WriteException, InvalidEnvelopedEntryException {
         int logRecordSize = record.getPersistentSize();
@@ -802,8 +762,8 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
             record.setPositionWithinLogSegment(positionWithinLogSegment);
         }
 
-        Promise<DLSN> writePromise = new Promise<DLSN>();
-        writePromise.addEventListener(new OpStatsListener<DLSN>(writeTime));
+        CompletableFuture<DLSN> writePromise = new CompletableFuture<DLSN>();
+        writePromise.whenComplete(new OpStatsListener<DLSN>(writeTime));
         recordSetWriter.writeRecord(record, writePromise);
 
         if (record.getTransactionId() < lastTxId) {
@@ -818,7 +778,7 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
         return writePromise;
     }
 
-    synchronized private Future<DLSN> writeControlLogRecord()
+    synchronized private CompletableFuture<DLSN> writeControlLogRecord()
             throws BKTransmitException, WriteException, InvalidEnvelopedEntryException,
                    LockingException, LogRecordTooLongException {
         LogRecord controlRec = new LogRecord(lastTxId, DistributedLogConstants.CONTROL_RECORD_CONTENT);
@@ -826,7 +786,7 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
         return writeControlLogRecord(controlRec);
     }
 
-    synchronized private Future<DLSN> writeControlLogRecord(LogRecord record)
+    synchronized private CompletableFuture<DLSN> writeControlLogRecord(LogRecord record)
             throws BKTransmitException, WriteException, InvalidEnvelopedEntryException,
                    LockingException, LogRecordTooLongException {
         return writeInternal(record);
@@ -851,12 +811,12 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
      * as read-only in the metadata. No appends to the
      * stream will be allowed after this point
      */
-    public Future<Long> markEndOfStream() {
+    public CompletableFuture<Long> markEndOfStream() {
         synchronized (this) {
             try {
                 writeEndOfStreamMarker();
             } catch (IOException e) {
-                return Future.exception(e);
+                return FutureUtils.exception(e);
             }
             streamEnded = true;
         }
@@ -905,60 +865,60 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
     }
 
     @Override
-    public synchronized Future<Long> flush() {
+    public synchronized CompletableFuture<Long> flush() {
         try {
             checkStateBeforeTransmit();
         } catch (WriteException e) {
-            return Future.exception(e);
+            return FutureUtils.exception(e);
         }
 
-        Future<Integer> transmitFuture;
+        CompletableFuture<Integer> transmitFuture;
         try {
             transmitFuture = transmit();
         } catch (BKTransmitException e) {
-            return Future.exception(e);
+            return FutureUtils.exception(e);
         } catch (LockingException e) {
-            return Future.exception(e);
+            return FutureUtils.exception(e);
         } catch (WriteException e) {
-            return Future.exception(e);
+            return FutureUtils.exception(e);
         } catch (InvalidEnvelopedEntryException e) {
-            return Future.exception(e);
+            return FutureUtils.exception(e);
         }
 
         if (null == transmitFuture) {
             if (null != packetPrevious) {
                 transmitFuture = packetPrevious.getTransmitFuture();
             }  else {
-                return Future.value(getLastTxIdAcknowledged());
+                return FutureUtils.value(getLastTxIdAcknowledged());
             }
         }
 
-        return transmitFuture.flatMap(GET_LAST_TXID_ACKNOWLEDGED_AFTER_TRANSMIT_FUNC);
+        return transmitFuture.thenCompose(GET_LAST_TXID_ACKNOWLEDGED_AFTER_TRANSMIT_FUNC);
     }
 
     @Override
-    public synchronized Future<Long> commit() {
+    public synchronized CompletableFuture<Long> commit() {
         // we don't pack control records with user records together
         // so transmit current output buffer if possible
-        Future<Integer> transmitFuture;
+        CompletableFuture<Integer> transmitFuture;
         try {
             try {
                 transmitFuture = transmit();
             } catch (IOException ioe) {
-                return Future.exception(ioe);
+                return FutureUtils.exception(ioe);
             }
             if (null == transmitFuture) {
                 writeControlLogRecord();
                 return flush();
             }
         } catch (IOException ioe) {
-            return Future.exception(ioe);
+            return FutureUtils.exception(ioe);
         }
-        return transmitFuture.flatMap(GET_LAST_TXID_ACKNOWLEDGED_AFTER_TRANSMIT_FUNC);
+        return transmitFuture.thenCompose(GET_LAST_TXID_ACKNOWLEDGED_AFTER_TRANSMIT_FUNC);
     }
 
-    Future<Long> flushAndCommit() {
-        return flush().flatMap(COMMIT_AFTER_FLUSH_FUNC);
+    CompletableFuture<Long> flushAndCommit() {
+        return flush().thenCompose(COMMIT_AFTER_FLUSH_FUNC);
     }
 
     void flushIfNeededNoThrow() {
@@ -1054,7 +1014,7 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
      * @throws WriteException if failed to create the envelope for the data to transmit
      * @throws InvalidEnvelopedEntryException when built an invalid enveloped entry
      */
-    private Future<Integer> transmit()
+    private CompletableFuture<Integer> transmit()
         throws BKTransmitException, LockingException, WriteException, InvalidEnvelopedEntryException {
         EntryBuffer recordSetToTransmit;
         transmitLock.lock();
@@ -1183,10 +1143,11 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
             }
         }
 
-        if (null != addCompleteFuturePool) {
+        if (null != scheduler) {
             final Stopwatch queuedTime = Stopwatch.createStarted();
-            addCompleteFuturePool.apply(new Function0<Void>() {
-                public Void apply() {
+            scheduler.submit(streamName, new Callable<Void>() {
+                @Override
+                public Void call() {
                     final Stopwatch deferredTime = Stopwatch.createStarted();
                     addCompleteQueuedTime.registerSuccessfulEvent(queuedTime.elapsed(TimeUnit.MICROSECONDS));
                     addCompleteDeferredProcessing(transmitPacket, entryId, effectiveRC.get());
@@ -1198,7 +1159,7 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
                     return String.format("AddComplete(Stream=%s, entryId=%d, rc=%d)",
                             fullyQualifiedLogSegment, entryId, rc);
                 }
-            }).addEventListener(new FutureEventListener<Void>() {
+            }).whenComplete(new FutureEventListener<Void>() {
                 @Override
                 public void onSuccess(Void done) {
                 }
@@ -1278,7 +1239,7 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
         if (BKException.Code.OK == transmitResult.get()) {
             recordSet.completeTransmit(logSegmentSequenceNumber, entryId);
         } else {
-            recordSet.abortTransmit(FutureUtils.transmitException(transmitResult.get()));
+            recordSet.abortTransmit(Utils.transmitException(transmitResult.get()));
         }
 
         if (cancelPendingPromises) {
@@ -1292,7 +1253,7 @@ class BKLogSegmentWriter implements LogSegmentWriter, AddCallback, Runnable, Siz
             }
             packetCurrentSaved.getRecordSet().abortTransmit(
                     new WriteCancelledException(streamName,
-                            FutureUtils.transmitException(transmitResult.get())));
+                            Utils.transmitException(transmitResult.get())));
         }
     }
 


Mime
View raw message