DL-124: Use Java8 Future rather than twitter Future #133

sijie · 2017-06-02T23:48:46Z

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.

- move proxy related class from protocol to proxy-protocol, changed client and service to proxy-client and proxy-service

…sCode

sijie · 2017-06-02T23:50:07Z

tag @fcuny @mgodave @leighst @jiazhai for reviews

asfbot · 2017-06-03T00:44:33Z

Build finished.
--none--

jiazhai · 2017-06-03T10:50:41Z

distributedlog-core/src/main/java/org/apache/distributedlog/BKAbstractLogWriter.java

                lastTruncationAttempt = writeHandler.purgeLogSegmentsOlderThanTimestamp(minTimestampToKeep);
            }
        }
    }

-    private Future<BKLogSegmentWriter> asyncStartNewLogSegment(final BKLogWriteHandler writeHandler,
+    private CompletableFuture<BKLogSegmentWriter> asyncStartNewLogSegment(final BKLogWriteHandler writeHandler,
                                                               final long startTxId,


Please also do the code alignment change here.

jiazhai · 2017-06-03T10:52:13Z

distributedlog-core/src/main/java/org/apache/distributedlog/BKAsyncLogReader.java

        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,


Please also do the code alignment change here.

jiazhai · 2017-06-03T10:52:31Z

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,


Please also do the code alignment change here.

jiazhai · 2017-06-03T10:56:05Z

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,


Please also do the code alignment change in this file, at line 201, 212, 242, 272,

jiazhai · 2017-06-03T11:01:41Z

distributedlog-core/src/main/java/org/apache/distributedlog/BKDistributedLogManager.java

    }

-    private Future<DLSN> getDLSNNotLessThanTxId(long fromTxnId,
+    private CompletableFuture<DLSN> getDLSNNotLessThanTxId(long fromTxnId,
                                                final List<LogSegmentMetadata> segments) {


Please also do the code alignment change in this file, also at line 565, 707, 838.

jiazhai · 2017-06-03T11:02:54Z

distributedlog-core/src/main/java/org/apache/distributedlog/BKLogHandler.java

        return asyncReadLastRecord(l, false, false, false);
    }

-    public Future<LogRecordWithDLSN> asyncReadLastRecord(final LogSegmentMetadata l,
+    public CompletableFuture<LogRecordWithDLSN> asyncReadLastRecord(final LogSegmentMetadata l,
                                                         final boolean fence,


Please also do the code alignment change here.

jiazhai · 2017-06-03T11:06:21Z

distributedlog-core/src/main/java/org/apache/distributedlog/BKLogWriteHandler.java

    }

-    protected Future<BKLogSegmentWriter> asyncStartLogSegment(final long txId,
+    protected CompletableFuture<BKLogSegmentWriter> asyncStartLogSegment(final long txId,
                                                              final boolean bestEffort,


Please also do the code alignment change here.

jiazhai · 2017-06-03T11:06:58Z

distributedlog-core/src/main/java/org/apache/distributedlog/BookKeeperClient.java

                        }
                    }
                }, null);
        return promise;
    }

-    public Future<Void> deleteLedger(long lid,
+    public CompletableFuture<Void> deleteLedger(long lid,
                                     final boolean ignoreNonExistentLedger) {


Please also do the code alignment change here.

jiazhai · 2017-06-03T11:10:04Z

distributedlog-core/src/main/java/org/apache/distributedlog/bk/SimpleLedgerAllocator.java


    // Ledger configuration
    private final QuorumConfigProvider quorumConfigProvider;

-    static Future<Versioned<byte[]>> getAndCreateAllocationData(final String allocatePath,
+    static CompletableFuture<Versioned<byte[]>> getAndCreateAllocationData(final String allocatePath,


Please also do the code alignment change here, and also at line 128, 159.

jiazhai · 2017-06-03T11:11:43Z

distributedlog-core/src/main/java/org/apache/distributedlog/impl/ZKLogSegmentMetadataStore.java

                    KeeperException.Code.get(rc)));
        }
    }

    @Override
-    public Future<Versioned<List<String>>> getLogSegmentNames(String logSegmentsPath,
+    public CompletableFuture<Versioned<List<String>>> getLogSegmentNames(String logSegmentsPath,
                                                              LogSegmentNamesListener listener) {


Please also do the code alignment change here.

jiazhai · 2017-06-03T11:14:04Z

...log-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) {


Please also do the code alignment change here, also at line 244.

jiazhai · 2017-06-03T11:15:56Z

...log-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) {


Please also do the code alignment change here, and line 269, 312, 521, 572.

jiazhai · 2017-06-03T11:18:41Z

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) {


Please also do the code alignment change here.

jiazhai · 2017-06-03T11:19:14Z

...ributedlog-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);


Please also do the code alignment change here, and line 71

jiazhai · 2017-06-03T11:19:46Z

...og-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) {


Please also do the code alignment change here.

jiazhai · 2017-06-03T11:20:17Z

...ributedlog-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);


Please also do the code alignment change here, and line 83.

jiazhai · 2017-06-03T11:20:53Z

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);


Please also do the code alignment change here, and line 102.

jiazhai · 2017-06-03T11:25:01Z

distributedlog-core/src/main/java/org/apache/distributedlog/util/Utils.java

                }
            }
        }, null);
        return promise;
    }

-    public static Future<Void> asyncClose(@Nullable AsyncCloseable closeable,
+    public static CompletableFuture<Void> asyncClose(@Nullable AsyncCloseable closeable,
                                          boolean swallowIOException) {


Please also do the code alignment change here, and line 573, 586.

jiazhai

+1, LGTM.
some small comments related to code alignment.
This changed so many places, thanks for the time.

leighst · 2017-06-09T04:19:10Z

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?

leighst · 2017-06-09T04:08:51Z

distributedlog-benchmark/pom.xml

@@ -27,12 +27,12 @@
  <dependencies>
    <dependency>
      <groupId>org.apache.distributedlog</groupId>
-      <artifactId>distributedlog-client</artifactId>
+      <artifactId>distributedlog-proxy-client</artifactId>


this change was based on #132. github doesn't allow pull request based on the other branch.

leighst · 2017-06-09T04:11:05Z

distributedlog-core/src/main/java/org/apache/distributedlog/BKLogReadHandler.java

-                FutureUtils.cancel(acquireFuture);
-                return null;
+        final CompletableFuture<Void> threadAcquirePromise = new CompletableFuture<Void>();
+        threadAcquirePromise.whenComplete((value, cause) -> {


do i understand correctly java8 future doesn't really have the concept of interrupt?

correctly, java8 future doesn't have the concept of interrupt. I try to simulate the similar thing.

leighst · 2017-06-09T04:16:41Z

distributedlog-core/src/main/java/org/apache/distributedlog/util/Utils.java

+            + BKException.getMessage(transmitResult), transmitResult);
+    }
+
+    public static <T> T ioResult(CompletableFuture<T> result) throws IOException {


why not have a java8 version of FutureUtils?
seems weird to throw all of this into one file

sijie · 2017-06-09T06:41:00Z

@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.

…re_to_java_future

sijie · 2017-06-13T20:46:38Z

@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.

sijie · 2017-06-13T20:47:01Z

Let me know if this approach works for you.

asfbot · 2017-06-13T21:08:44Z

Build finished.
--none--

asfbot · 2017-06-13T21:20:06Z

Build finished.
--none--

asfbot · 2017-06-13T23:35:19Z

Build finished.
--none--

leighst · 2017-06-14T20:32:50Z

distributedlog-benchmark/conf/log4j.properties

 log4j.logger.org.apache.bookkeeper.util.SafeRunnable=INFO, Executors
-log4j.additivity.org.apache.distributedlog.util.MonitoredFuturePool=false


i think this will result in duplicate messages?

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.

sijie added 4 commits May 26, 2017 15:05

Re-organize the distributedlog modules

54c2de0

- move proxy related class from protocol to proxy-protocol, changed client and service to proxy-client and proxy-service

Use integration for exception code rather than thrift generated Statu…

67e7615

…sCode

Use the latest thrift version for distributedlog-core and remove scrooge

6e58786

Change Twitter Future to Java8 CompletableFuture

ce0686e

jiazhai reviewed Jun 3, 2017

View reviewed changes

jiazhai approved these changes Jun 3, 2017

View reviewed changes

leighst reviewed Jun 9, 2017

View reviewed changes

sijie added 2 commits June 12, 2017 14:38

Merge branch 'master' into change_twitter_future_to_java_future

987c531

Move Java Futures to common

b952f34

sijie added 8 commits June 12, 2017 22:54

Move the log interfaces to org.apache.distributedlog.api

38ed9e7

Fix lines over 120 characters

75eecd5

new module

bc0c640

Merge branch 'new_module' into change_twitter_future_to_java_future

563e2b4

Add FutureUtils

e6c0b81

Merge branch 'new_module' into change_twitter_future_to_java_future

3b8b47c

Add distributedlog-core-twitter module for backward compatibility

d915c39

Merge remote-tracking branch 'apache/master' into change_twitter_futu…

2534a53

…re_to_java_future

Add the distributedlog-common test-jar

eca076d

leighst approved these changes Jun 18, 2017

View reviewed changes

asfgit closed this in 53fca4a Jun 21, 2017

		log4j.logger.org.apache.bookkeeper.util.SafeRunnable=INFO, Executors
		log4j.additivity.org.apache.distributedlog.util.MonitoredFuturePool=false

DL-124: Use Java8 Future rather than twitter Future #133

DL-124: Use Java8 Future rather than twitter Future #133

Conversation

sijie commented Jun 2, 2017

sijie commented Jun 2, 2017

asfbot commented Jun 3, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jiazhai left a comment • edited

Choose a reason for hiding this comment

leighst commented Jun 9, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sijie commented Jun 9, 2017

sijie commented Jun 13, 2017

sijie commented Jun 13, 2017

asfbot commented Jun 13, 2017

asfbot commented Jun 13, 2017

asfbot commented Jun 13, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jiazhai left a comment •

edited