You are viewing a plain text version of this content. The canonical link for it is here.
Posted to distributedlog-dev@bookkeeper.apache.org by si...@apache.org on 2017/06/21 17:20:49 UTC
[17/23] incubator-distributedlog git commit: DL-124: Use Java8 Future
rather than twitter Future
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())));
}
}