You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@distributedlog.apache.org by sijie <gi...@git.apache.org> on 2017/06/02 23:48:54 UTC
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
GitHub user sijie opened a pull request:
https://github.com/apache/incubator-distributedlog/pull/133
DL-124: Use Java8 Future rather than twitter Future
Switch to use Java8 CompletableFuture, to reduce dependencies introduced by twitter future and make it more friendly to users (users don't think of using which version of scala).
This change is based on #132 . Gitsha ce0686e is the change to review.
The changes:
- Change Future to CompletableFuture
- Map to thenApply
- flatMap to thenCompose
- Added a FutureEventListener, and switch addEvenListener to whenComplete (or whenCompleteAsync)
- setValue to complete
- setException to completeExceptionally
- add rescue, ignore, ensure to FutureUtils as util functions.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/sijie/incubator-distributedlog change_twitter_future_to_java_future
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-distributedlog/pull/133.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #133
----
commit 54c2de047e1656e34ead7fb54070441afd9c140d
Author: Sijie Guo <si...@apache.org>
Date: 2017-05-26T22:05:43Z
Re-organize the distributedlog modules
- move proxy related class from protocol to proxy-protocol, changed client and service to proxy-client and proxy-service
commit 67e76150b103422f16883857a1c1345cf044fb45
Author: Sijie Guo <si...@apache.org>
Date: 2017-05-27T07:04:40Z
Use integration for exception code rather than thrift generated StatusCode
commit 6e587869f87cdce50ae93ba3d52767719d1ab5a6
Author: Sijie Guo <si...@apache.org>
Date: 2017-05-27T07:34:02Z
Use the latest thrift version for distributedlog-core and remove scrooge
commit ce0686e30e89c75ffce81473de5a0264d5d95f58
Author: Sijie Guo <si...@apache.org>
Date: 2017-05-29T23:06:19Z
Change Twitter Future to Java8 CompletableFuture
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by jiazhai <gi...@git.apache.org>.
Github user jiazhai commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r119984733
--- Diff: distributedlog-core/src/main/java/org/apache/distributedlog/bk/SimpleLedgerAllocator.java ---
@@ -105,73 +103,71 @@ public ConcurrentObtainException(Phase phase, String msg) {
// Allocated Ledger
LedgerHandle allocatedLh = null;
- Future<Void> closeFuture = null;
- final LinkedList<Future<Void>> ledgerDeletions =
- new LinkedList<Future<Void>>();
+ CompletableFuture<Void> closeFuture = null;
+ final LinkedList<CompletableFuture<Void>> ledgerDeletions =
+ new LinkedList<CompletableFuture<Void>>();
// Ledger configuration
private final QuorumConfigProvider quorumConfigProvider;
- static Future<Versioned<byte[]>> getAndCreateAllocationData(final String allocatePath,
+ static CompletableFuture<Versioned<byte[]>> getAndCreateAllocationData(final String allocatePath,
--- End diff --
Please also do the code alignment change here, and also at line 128, 159.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by jiazhai <gi...@git.apache.org>.
Github user jiazhai commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r119984462
--- Diff: distributedlog-core/src/main/java/org/apache/distributedlog/BKAsyncLogReader.java ---
@@ -404,7 +392,7 @@ public String getStreamName() {
* num entries to read
* @return A promise that satisfied with a non-empty list of log records with their DLSN.
*/
- private synchronized Future<List<LogRecordWithDLSN>> readInternal(int numEntries,
+ private synchronized CompletableFuture<List<LogRecordWithDLSN>> readInternal(int numEntries,
long deadlineTime,
--- End diff --
Please also do the code alignment change here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by jiazhai <gi...@git.apache.org>.
Github user jiazhai commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r119984686
--- Diff: distributedlog-core/src/main/java/org/apache/distributedlog/BookKeeperClient.java ---
@@ -198,52 +196,52 @@ public synchronized BookKeeper get() throws IOException {
}
// Util functions
- public Future<LedgerHandle> createLedger(int ensembleSize,
- int writeQuorumSize,
- int ackQuorumSize) {
+ public CompletableFuture<LedgerHandle> createLedger(int ensembleSize,
+ int writeQuorumSize,
+ int ackQuorumSize) {
BookKeeper bk;
try {
bk = get();
} catch (IOException ioe) {
- return Future.exception(ioe);
+ return FutureUtils.exception(ioe);
}
- final Promise<LedgerHandle> promise = new Promise<LedgerHandle>();
+ final CompletableFuture<LedgerHandle> promise = new CompletableFuture<LedgerHandle>();
bk.asyncCreateLedger(ensembleSize, writeQuorumSize, ackQuorumSize,
BookKeeper.DigestType.CRC32, passwd, new AsyncCallback.CreateCallback() {
@Override
public void createComplete(int rc, LedgerHandle lh, Object ctx) {
if (BKException.Code.OK == rc) {
- promise.updateIfEmpty(new Return<LedgerHandle>(lh));
+ promise.complete(lh);
} else {
- promise.updateIfEmpty(new Throw<LedgerHandle>(BKException.create(rc)));
+ promise.completeExceptionally(BKException.create(rc));
}
}
}, null);
return promise;
}
- public Future<Void> deleteLedger(long lid,
+ public CompletableFuture<Void> deleteLedger(long lid,
final boolean ignoreNonExistentLedger) {
--- End diff --
Please also do the code alignment change here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by jiazhai <gi...@git.apache.org>.
Github user jiazhai commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r119984837
--- Diff: distributedlog-core/src/main/java/org/apache/distributedlog/logsegment/LogSegmentEntryStore.java ---
@@ -58,7 +57,7 @@
* @param startEntryId the start entry id
* @return future represent the opened reader
*/
- Future<LogSegmentEntryReader> openReader(LogSegmentMetadata segment,
+ CompletableFuture<LogSegmentEntryReader> openReader(LogSegmentMetadata segment,
long startEntryId);
--- End diff --
Please also do the code alignment change here, and line 71
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by jiazhai <gi...@git.apache.org>.
Github user jiazhai commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r119984743
--- Diff: distributedlog-core/src/main/java/org/apache/distributedlog/impl/ZKLogSegmentMetadataStore.java ---
@@ -350,40 +349,40 @@ public void process(WatchedEvent event) {
}
@Override
- public Future<LogSegmentMetadata> getLogSegment(String logSegmentPath) {
+ public CompletableFuture<LogSegmentMetadata> getLogSegment(String logSegmentPath) {
return LogSegmentMetadata.read(zkc, logSegmentPath, skipMinVersionCheck);
}
- Future<Versioned<List<String>>> zkGetLogSegmentNames(String logSegmentsPath, Watcher watcher) {
- Promise<Versioned<List<String>>> result = new Promise<Versioned<List<String>>>();
+ CompletableFuture<Versioned<List<String>>> zkGetLogSegmentNames(String logSegmentsPath, Watcher watcher) {
+ CompletableFuture<Versioned<List<String>>> result = new CompletableFuture<Versioned<List<String>>>();
try {
zkc.get().getChildren(logSegmentsPath, watcher, this, result);
} catch (ZooKeeperClient.ZooKeeperConnectionException e) {
- result.setException(FutureUtils.zkException(e, logSegmentsPath));
+ result.completeExceptionally(Utils.zkException(e, logSegmentsPath));
} catch (InterruptedException e) {
- result.setException(FutureUtils.zkException(e, logSegmentsPath));
+ result.completeExceptionally(Utils.zkException(e, logSegmentsPath));
}
return result;
}
@Override
@SuppressWarnings("unchecked")
public void processResult(int rc, String path, Object ctx, List<String> children, Stat stat) {
- Promise<Versioned<List<String>>> result = ((Promise<Versioned<List<String>>>) ctx);
+ CompletableFuture<Versioned<List<String>>> result = ((CompletableFuture<Versioned<List<String>>>) ctx);
if (KeeperException.Code.OK.intValue() == rc) {
/** cversion: the number of changes to the children of this znode **/
ZkVersion zkVersion = new ZkVersion(stat.getCversion());
- result.setValue(new Versioned(children, zkVersion));
+ result.complete(new Versioned(children, zkVersion));
} else if (KeeperException.Code.NONODE.intValue() == rc) {
- result.setException(new LogNotFoundException("Log " + path + " not found"));
+ result.completeExceptionally(new LogNotFoundException("Log " + path + " not found"));
} else {
- result.setException(new ZKException("Failed to get log segments from " + path,
+ result.completeExceptionally(new ZKException("Failed to get log segments from " + path,
KeeperException.Code.get(rc)));
}
}
@Override
- public Future<Versioned<List<String>>> getLogSegmentNames(String logSegmentsPath,
+ public CompletableFuture<Versioned<List<String>>> getLogSegmentNames(String logSegmentsPath,
LogSegmentNamesListener listener) {
--- End diff --
Please also do the code alignment change here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by jiazhai <gi...@git.apache.org>.
Github user jiazhai commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r119984610
--- Diff: distributedlog-core/src/main/java/org/apache/distributedlog/BKLogHandler.java ---
@@ -397,15 +398,15 @@ private Long sum(List<Long> values) {
}
@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,
--- End diff --
Please also do the code alignment change here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by jiazhai <gi...@git.apache.org>.
Github user jiazhai commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r119984515
--- Diff: distributedlog-core/src/main/java/org/apache/distributedlog/BKAsyncLogWriter.java ---
@@ -206,7 +197,7 @@ private BKLogSegmentWriter getCachedLogSegmentWriter() throws WriteException {
}
}
- private Future<BKLogSegmentWriter> getLogSegmentWriter(long firstTxid,
+ private CompletableFuture<BKLogSegmentWriter> getLogSegmentWriter(long firstTxid,
boolean bestEffort,
--- End diff --
Please also do the code alignment change in this file, at line 201, 212, 242, 272,
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by sijie <gi...@git.apache.org>.
Github user sijie commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r121057575
--- Diff: distributedlog-core/src/main/java/org/apache/distributedlog/BKLogReadHandler.java ---
@@ -201,33 +183,31 @@ public void safeRun() {
* 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) -> {
--- End diff --
correctly, java8 future doesn't have the concept of interrupt. I try to simulate the similar thing.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog issue #133: DL-124: Use Java8 Future rather than tw...
Posted by sijie <gi...@git.apache.org>.
Github user sijie commented on the issue:
https://github.com/apache/incubator-distributedlog/pull/133
tag @fcuny @mgodave @leighst @jiazhai for reviews
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by leighst <gi...@git.apache.org>.
Github user leighst commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r121046188
--- Diff: distributedlog-core/src/main/java/org/apache/distributedlog/BKLogReadHandler.java ---
@@ -201,33 +183,31 @@ public void safeRun() {
* 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) -> {
--- End diff --
do i understand correctly java8 future doesn't really have the concept of interrupt?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by jiazhai <gi...@git.apache.org>.
Github user jiazhai commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r119984925
--- Diff: distributedlog-core/src/main/java/org/apache/distributedlog/util/Utils.java ---
@@ -445,35 +440,35 @@ public void processResult(int rc, String path, Object ctx) {
* false if the node doesn't exist, otherwise future will throw exception
*
*/
- public static Future<Boolean> zkDeleteIfNotExist(ZooKeeperClient zkc, String path, ZkVersion version) {
+ public static CompletableFuture<Boolean> zkDeleteIfNotExist(ZooKeeperClient zkc, String path, ZkVersion version) {
ZooKeeper zk;
try {
zk = zkc.get();
} catch (ZooKeeperClient.ZooKeeperConnectionException e) {
- return Future.exception(FutureUtils.zkException(e, path));
+ return FutureUtils.exception(zkException(e, path));
} catch (InterruptedException e) {
- return Future.exception(FutureUtils.zkException(e, path));
+ return FutureUtils.exception(zkException(e, path));
}
- final Promise<Boolean> promise = new Promise<Boolean>();
+ final CompletableFuture<Boolean> promise = new CompletableFuture<Boolean>();
zk.delete(path, version.getZnodeVersion(), new AsyncCallback.VoidCallback() {
@Override
public void processResult(int rc, String path, Object ctx) {
if (KeeperException.Code.OK.intValue() == rc ) {
- promise.setValue(true);
+ promise.complete(true);
} else if (KeeperException.Code.NONODE.intValue() == rc) {
- promise.setValue(false);
+ promise.complete(false);
} else {
- promise.setException(KeeperException.create(KeeperException.Code.get(rc)));
+ promise.completeExceptionally(KeeperException.create(KeeperException.Code.get(rc)));
}
}
}, null);
return promise;
}
- public static Future<Void> asyncClose(@Nullable AsyncCloseable closeable,
+ public static CompletableFuture<Void> asyncClose(@Nullable AsyncCloseable closeable,
boolean swallowIOException) {
--- End diff --
Please also do the code alignment change here, and line 573, 586.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by jiazhai <gi...@git.apache.org>.
Github user jiazhai commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r119984456
--- Diff: distributedlog-core/src/main/java/org/apache/distributedlog/BKAsyncLogReader.java ---
@@ -381,16 +369,16 @@ public String getStreamName() {
* @return A promise that when satisfied will contain the Log Record with its DLSN.
*/
@Override
- public synchronized Future<LogRecordWithDLSN> readNext() {
- return readInternal(1, 0, TimeUnit.MILLISECONDS).map(READ_NEXT_MAP_FUNCTION);
+ public synchronized CompletableFuture<LogRecordWithDLSN> readNext() {
+ return readInternal(1, 0, TimeUnit.MILLISECONDS).thenApply(READ_NEXT_MAP_FUNCTION);
}
- public synchronized Future<List<LogRecordWithDLSN>> readBulk(int numEntries) {
+ public synchronized CompletableFuture<List<LogRecordWithDLSN>> readBulk(int numEntries) {
return readInternal(numEntries, 0, TimeUnit.MILLISECONDS);
}
@Override
- public synchronized Future<List<LogRecordWithDLSN>> readBulk(int numEntries,
+ public synchronized CompletableFuture<List<LogRecordWithDLSN>> readBulk(int numEntries,
long waitTime,
--- End diff --
Please also do the code alignment change here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog issue #133: DL-124: Use Java8 Future rather than tw...
Posted by sijie <gi...@git.apache.org>.
Github user sijie commented on the issue:
https://github.com/apache/incubator-distributedlog/pull/133
@leighst : let's discuss this tomorrow face to face. I am thinking of a new api with java8 future that works for both core-library and proxy. that means the way to distinguish core and library is just the uri. we can have the old twitter-future based api wrapper over the new java8 future api. so it will be backward compatible.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog issue #133: DL-124: Use Java8 Future rather than tw...
Posted by sijie <gi...@git.apache.org>.
Github user sijie commented on the issue:
https://github.com/apache/incubator-distributedlog/pull/133
@leighst - I pushed a new change:
- moved the interfaces to org.apache.distributedlog.api, which they are using Java8 interface.
- moved the old interfaces to distributedlog-core-twitter module, they are are backward compatible to old application.
So the old application can use distributedlog-core-twitter and keep the binary compatible.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by jiazhai <gi...@git.apache.org>.
Github user jiazhai commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r119984844
--- Diff: distributedlog-core/src/main/java/org/apache/distributedlog/metadata/LogSegmentMetadataStoreUpdater.java ---
@@ -73,7 +72,7 @@ private String formatLogSegmentSequenceNumber(long logSegmentSeqNo) {
}
@Override
- public Future<LogSegmentMetadata> changeSequenceNumber(LogSegmentMetadata segment,
+ public CompletableFuture<LogSegmentMetadata> changeSequenceNumber(LogSegmentMetadata segment,
long logSegmentSeqNo) {
--- End diff --
Please also do the code alignment change here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by leighst <gi...@git.apache.org>.
Github user leighst commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r122056689
--- Diff: distributedlog-benchmark/conf/log4j.properties ---
@@ -30,11 +30,7 @@ log4j.logger.org.apache.zookeeper=INFO
log4j.logger.org.apache.bookkeeper=INFO
# redirect executor output to executors.log since slow op warnings can be quite verbose
-log4j.logger.org.apache.distributedlog.util.MonitoredFuturePool=INFO, Executors
-log4j.logger.org.apache.distributedlog.util.MonitoredScheduledThreadPoolExecutor=INFO, Executors
log4j.logger.org.apache.bookkeeper.util.SafeRunnable=INFO, Executors
-log4j.additivity.org.apache.distributedlog.util.MonitoredFuturePool=false
--- End diff --
i think this will result in duplicate messages?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by leighst <gi...@git.apache.org>.
Github user leighst commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r121046009
--- Diff: distributedlog-benchmark/pom.xml ---
@@ -27,12 +27,12 @@
<dependencies>
<dependency>
<groupId>org.apache.distributedlog</groupId>
- <artifactId>distributedlog-client</artifactId>
+ <artifactId>distributedlog-proxy-client</artifactId>
--- End diff --
why?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by leighst <gi...@git.apache.org>.
Github user leighst commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r121046543
--- Diff: distributedlog-core/src/main/java/org/apache/distributedlog/util/Utils.java ---
@@ -635,4 +631,80 @@ public static String getParent(final String path) {
return path.substring(0, lastIndex);
}
+ /**
+ * Convert the <i>throwable</i> to zookeeper related exceptions.
+ *
+ * @param throwable cause
+ * @param path zookeeper path
+ * @return zookeeper related exceptions
+ */
+ public static Throwable zkException(Throwable throwable, String path) {
+ if (throwable instanceof KeeperException) {
+ return new ZKException("Encountered zookeeper exception on " + path, (KeeperException) throwable);
+ } else if (throwable instanceof ZooKeeperClient.ZooKeeperConnectionException) {
+ return new ZKException("Encountered zookeeper connection loss on " + path,
+ KeeperException.Code.CONNECTIONLOSS);
+ } else if (throwable instanceof InterruptedException) {
+ return new DLInterruptedException("Interrupted on operating " + path, throwable);
+ } else {
+ return new UnexpectedException("Encountered unexpected exception on operatiing " + path, throwable);
+ }
+ }
+
+ /**
+ * Create transmit exception from transmit result.
+ *
+ * @param transmitResult
+ * transmit result (basically bk exception code)
+ * @return transmit exception
+ */
+ public static BKTransmitException transmitException(int transmitResult) {
+ return new BKTransmitException("Failed to write to bookkeeper; Error is ("
+ + transmitResult + ") "
+ + BKException.getMessage(transmitResult), transmitResult);
+ }
+
+ public static <T> T ioResult(CompletableFuture<T> result) throws IOException {
--- End diff --
why not have a java8 version of FutureUtils?
seems weird to throw all of this into one file
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by jiazhai <gi...@git.apache.org>.
Github user jiazhai commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r119984799
--- Diff: distributedlog-core/src/main/java/org/apache/distributedlog/impl/metadata/ZKLogStreamMetadataStore.java ---
@@ -237,15 +227,13 @@ public DistributedLock createWriteLock(LogMetadataForWriter metadata) {
// Create Read Lock
//
- private Future<Void> ensureReadLockPathExist(final LogMetadata logMetadata,
+ private CompletableFuture<Void> ensureReadLockPathExist(final LogMetadata logMetadata,
final String readLockPath) {
--- End diff --
Please also do the code alignment change here, and line 269, 312, 521, 572.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by sijie <gi...@git.apache.org>.
Github user sijie commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r121057495
--- Diff: distributedlog-benchmark/pom.xml ---
@@ -27,12 +27,12 @@
<dependencies>
<dependency>
<groupId>org.apache.distributedlog</groupId>
- <artifactId>distributedlog-client</artifactId>
+ <artifactId>distributedlog-proxy-client</artifactId>
--- End diff --
this change was based on #132. github doesn't allow pull request based on the other branch.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by jiazhai <gi...@git.apache.org>.
Github user jiazhai commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r119984432
--- Diff: distributedlog-core/src/main/java/org/apache/distributedlog/BKAbstractLogWriter.java ---
@@ -357,80 +354,72 @@ private void truncateLogSegmentsIfNecessary(BKLogWriteHandler writeHandler) {
// skip scheduling if there is task that's already running
//
synchronized (this) {
- if (truncationEnabled && ((lastTruncationAttempt == null) || lastTruncationAttempt.isDefined())) {
+ if (truncationEnabled && ((lastTruncationAttempt == null) || lastTruncationAttempt.isDone())) {
lastTruncationAttempt = writeHandler.purgeLogSegmentsOlderThanTimestamp(minTimestampToKeep);
}
}
}
- private Future<BKLogSegmentWriter> asyncStartNewLogSegment(final BKLogWriteHandler writeHandler,
+ private CompletableFuture<BKLogSegmentWriter> asyncStartNewLogSegment(final BKLogWriteHandler writeHandler,
final long startTxId,
--- End diff --
Please also do the code alignment change here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by jiazhai <gi...@git.apache.org>.
Github user jiazhai commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r119984863
--- Diff: distributedlog-core/src/main/java/org/apache/distributedlog/metadata/MetadataUpdater.java ---
@@ -56,7 +56,7 @@
* ledger sequence number to change.
* @return new log segment
*/
- Future<LogSegmentMetadata> changeSequenceNumber(LogSegmentMetadata segment,
+ CompletableFuture<LogSegmentMetadata> changeSequenceNumber(LogSegmentMetadata segment,
long logSegmentSeqNo);
--- End diff --
Please also do the code alignment change here, and line 102.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by jiazhai <gi...@git.apache.org>.
Github user jiazhai commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r119984828
--- Diff: distributedlog-core/src/main/java/org/apache/distributedlog/lock/ZKSessionLock.java ---
@@ -1088,9 +1063,9 @@ public String getActionName() {
});
}
- private Future<String> checkLockOwnerAndWaitIfPossible(final LockWatcher lockWatcher,
+ private CompletableFuture<String> checkLockOwnerAndWaitIfPossible(final LockWatcher lockWatcher,
final boolean wait) {
--- End diff --
Please also do the code alignment change here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by jiazhai <gi...@git.apache.org>.
Github user jiazhai commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r119984676
--- Diff: distributedlog-core/src/main/java/org/apache/distributedlog/BKLogWriteHandler.java ---
@@ -483,23 +482,23 @@ protected long assignLogSegmentSequenceNumber() throws IOException {
}
protected BKLogSegmentWriter doStartLogSegment(long txId, boolean bestEffort, boolean allowMaxTxID) throws IOException {
- return FutureUtils.result(asyncStartLogSegment(txId, bestEffort, allowMaxTxID));
+ return Utils.ioResult(asyncStartLogSegment(txId, bestEffort, allowMaxTxID));
}
- protected Future<BKLogSegmentWriter> asyncStartLogSegment(final long txId,
+ protected CompletableFuture<BKLogSegmentWriter> asyncStartLogSegment(final long txId,
final boolean bestEffort,
--- End diff --
Please also do the code alignment change here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by jiazhai <gi...@git.apache.org>.
Github user jiazhai commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r119984852
--- Diff: distributedlog-core/src/main/java/org/apache/distributedlog/metadata/LogStreamMetadataStore.java ---
@@ -59,7 +59,7 @@
* @param readerId the reader id used for lock
* @return the read lock
*/
- Future<DistributedLock> createReadLock(LogMetadataForReader metadata,
+ CompletableFuture<DistributedLock> createReadLock(LogMetadataForReader metadata,
Optional<String> readerId);
--- End diff --
Please also do the code alignment change here, and line 83.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by jiazhai <gi...@git.apache.org>.
Github user jiazhai commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r119984594
--- Diff: distributedlog-core/src/main/java/org/apache/distributedlog/BKDistributedLogManager.java ---
@@ -525,75 +495,63 @@ public BKSyncLogWriter startLogSegmentNonPartitioned() throws IOException {
*/
@Override
public BKAsyncLogWriter startAsyncLogSegmentNonPartitioned() throws IOException {
- return (BKAsyncLogWriter) FutureUtils.result(openAsyncLogWriter());
+ return (BKAsyncLogWriter) Utils.ioResult(openAsyncLogWriter());
}
@Override
- public Future<AsyncLogWriter> openAsyncLogWriter() {
+ public CompletableFuture<AsyncLogWriter> openAsyncLogWriter() {
try {
checkClosedOrInError("startLogSegmentNonPartitioned");
} catch (AlreadyClosedException e) {
- return Future.exception(e);
+ return FutureUtils.exception(e);
}
- Future<BKLogWriteHandler> createWriteHandleFuture;
+ CompletableFuture<BKLogWriteHandler> createWriteHandleFuture;
synchronized (this) {
// 1. create the locked write handler
createWriteHandleFuture = asyncCreateWriteHandler(true);
}
- return createWriteHandleFuture.flatMap(new AbstractFunction1<BKLogWriteHandler, Future<AsyncLogWriter>>() {
- @Override
- public Future<AsyncLogWriter> apply(final BKLogWriteHandler writeHandler) {
- final BKAsyncLogWriter writer;
- synchronized (BKDistributedLogManager.this) {
- // 2. create the writer with the handler
- writer = new BKAsyncLogWriter(
- conf,
- dynConf,
- BKDistributedLogManager.this,
- writeHandler,
- featureProvider,
- statsLogger);
- }
- // 3. recover the incomplete log segments
- return writeHandler.recoverIncompleteLogSegments()
- .map(new AbstractFunction1<Long, AsyncLogWriter>() {
- @Override
- public AsyncLogWriter apply(Long lastTxId) {
- // 4. update last tx id if successfully recovered
- writer.setLastTxId(lastTxId);
- return writer;
- }
- }).onFailure(new AbstractFunction1<Throwable, BoxedUnit>() {
- @Override
- public BoxedUnit apply(Throwable cause) {
- // 5. close the writer if recovery failed
- writer.asyncAbort();
- return BoxedUnit.UNIT;
- }
- });
+ return createWriteHandleFuture.thenCompose(writeHandler -> {
+ final BKAsyncLogWriter writer;
+ synchronized (BKDistributedLogManager.this) {
+ // 2. create the writer with the handler
+ writer = new BKAsyncLogWriter(
+ conf,
+ dynConf,
+ BKDistributedLogManager.this,
+ writeHandler,
+ featureProvider,
+ statsLogger);
}
+ // 3. recover the incomplete log segments
+ return writeHandler.recoverIncompleteLogSegments()
+ .thenApply(lastTxId -> {
+ // 4. update last tx id if successfully recovered
+ writer.setLastTxId(lastTxId);
+ return (AsyncLogWriter) writer;
+ })
+ .whenComplete((lastTxId, cause) -> {
+ if (null != cause) {
+ // 5. close the writer if recovery failed
+ writer.asyncAbort();
+ }
+ });
});
}
@Override
- public Future<DLSN> getDLSNNotLessThanTxId(final long fromTxnId) {
- return getLogSegmentsAsync().flatMap(new AbstractFunction1<List<LogSegmentMetadata>, Future<DLSN>>() {
- @Override
- public Future<DLSN> apply(List<LogSegmentMetadata> segments) {
- return getDLSNNotLessThanTxId(fromTxnId, segments);
- }
- });
+ public CompletableFuture<DLSN> getDLSNNotLessThanTxId(final long fromTxnId) {
+ return getLogSegmentsAsync().thenCompose(segments -> getDLSNNotLessThanTxId(fromTxnId, segments));
}
- private Future<DLSN> getDLSNNotLessThanTxId(long fromTxnId,
+ private CompletableFuture<DLSN> getDLSNNotLessThanTxId(long fromTxnId,
final List<LogSegmentMetadata> segments) {
--- End diff --
Please also do the code alignment change in this file, also at line 565, 707, 838.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by jiazhai <gi...@git.apache.org>.
Github user jiazhai commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r119984772
--- Diff: distributedlog-core/src/main/java/org/apache/distributedlog/impl/logsegment/BKLogSegmentEntryStore.java ---
@@ -186,13 +185,13 @@ LedgerAllocator createLedgerAllocator(LogMetadataForWriter logMetadata,
//
@Override
- public Future<LogSegmentEntryReader> openReader(LogSegmentMetadata segment,
+ public CompletableFuture<LogSegmentEntryReader> openReader(LogSegmentMetadata segment,
long startEntryId) {
--- End diff --
Please also do the code alignment change here, also at line 244.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog issue #133: DL-124: Use Java8 Future rather than tw...
Posted by sijie <gi...@git.apache.org>.
Github user sijie commented on the issue:
https://github.com/apache/incubator-distributedlog/pull/133
Let me know if this approach works for you.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by sijie <gi...@git.apache.org>.
Github user sijie commented on a diff in the pull request:
https://github.com/apache/incubator-distributedlog/pull/133#discussion_r122609664
--- Diff: distributedlog-benchmark/conf/log4j.properties ---
@@ -30,11 +30,7 @@ log4j.logger.org.apache.zookeeper=INFO
log4j.logger.org.apache.bookkeeper=INFO
# redirect executor output to executors.log since slow op warnings can be quite verbose
-log4j.logger.org.apache.distributedlog.util.MonitoredFuturePool=INFO, Executors
-log4j.logger.org.apache.distributedlog.util.MonitoredScheduledThreadPoolExecutor=INFO, Executors
log4j.logger.org.apache.bookkeeper.util.SafeRunnable=INFO, Executors
-log4j.additivity.org.apache.distributedlog.util.MonitoredFuturePool=false
--- End diff --
MonitoredFuturePool and MonitoredScheduledThreadPoolExecutor are already removed in this patch. because in Java8, you can configure a thread where the callbacks will be executed, there is no need for additional future pool, we just use the ordered scheduler.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog pull request #133: DL-124: Use Java8 Future rather ...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/incubator-distributedlog/pull/133
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-distributedlog issue #133: DL-124: Use Java8 Future rather than tw...
Posted by leighst <gi...@git.apache.org>.
Github user leighst commented on the issue:
https://github.com/apache/incubator-distributedlog/pull/133
Is there some way we can make migration easier? We're going to have to do a lot of translation to twitter futures when we integrate. Should we be thinking about some kind of adapter layer?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---