sijie closed pull request #240: Make distributedlog compiled with latest
bookkeeper version
URL: https://github.com/apache/distributedlog/pull/240
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/distributedlog-core/src/main/java/org/apache/bookkeeper/client/LedgerReader.java
b/distributedlog-core/src/main/java/org/apache/bookkeeper/client/LedgerReader.java
index ccdc52b4..6c6bd4a7 100644
---
a/distributedlog-core/src/main/java/org/apache/bookkeeper/client/LedgerReader.java
+++
b/distributedlog-core/src/main/java/org/apache/bookkeeper/client/LedgerReader.java
@@ -21,27 +21,27 @@
import io.netty.buffer.Unpooled;
import java.net.InetSocketAddress;
import java.util.ArrayList;
-import java.util.Enumeration;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.SortedMap;
import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.bookkeeper.client.BKException.BKNoSuchEntryException;
+import org.apache.bookkeeper.client.BKException.Code;
import org.apache.bookkeeper.client.DistributionSchedule.WriteSet;
+import org.apache.bookkeeper.client.api.LedgerEntries;
+import org.apache.bookkeeper.client.impl.LedgerEntryImpl;
+import org.apache.bookkeeper.common.concurrent.FutureEventListener;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.proto.BookieClient;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
import
org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.ReadEntryCallback;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* Reader used for DL tools to read entries.
*/
public class LedgerReader {
- private static final Logger logger =
LoggerFactory.getLogger(LedgerReader.class);
-
/**
* Read Result Holder.
*/
@@ -137,21 +137,42 @@ public void forwardReadEntriesFromLastConfirmed(final
LedgerHandle lh,
final
GenericCallback<List<LedgerEntry>> callback) {
final List<LedgerEntry> resultList = new ArrayList<LedgerEntry>();
- final AsyncCallback.ReadCallback readCallback = new
AsyncCallback.ReadCallback() {
+ final FutureEventListener<LedgerEntries> readListener = new
FutureEventListener<LedgerEntries>() {
+
+ private void readNext(long entryId) {
+ PendingReadOp op = new PendingReadOp(lh, lh.bk.scheduler,
entryId, entryId, false);
+ op.future().whenComplete(this);
+ op.submit();
+ }
+
@Override
- public void readComplete(int rc, LedgerHandle lh,
Enumeration<LedgerEntry> entries, Object ctx) {
- if (BKException.Code.NoSuchEntryException == rc) {
- callback.operationComplete(BKException.Code.OK,
resultList);
- } else if (BKException.Code.OK == rc) {
- while (entries.hasMoreElements()) {
- resultList.add(entries.nextElement());
- }
- long entryId = (Long) ctx;
- ++entryId;
- PendingReadOp readOp = new PendingReadOp(lh,
lh.bk.scheduler, entryId, entryId, this, entryId);
- readOp.initiate();
+ public void onSuccess(LedgerEntries ledgerEntries) {
+ long entryId = -1L;
+ for (org.apache.bookkeeper.client.api.LedgerEntry entry :
ledgerEntries) {
+ resultList.add(new LedgerEntry((LedgerEntryImpl) entry));
+ entryId = entry.getEntryId();
+ }
+ try {
+ ledgerEntries.close();
+ } catch (Exception e) {
+ // bk should not throw any exceptions here
+ }
+ ++entryId;
+ readNext(entryId);
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ if (throwable instanceof BKNoSuchEntryException) {
+ callback.operationComplete(Code.OK, resultList);
} else {
- callback.operationComplete(rc, resultList);
+ int retCode;
+ if (throwable instanceof BKException) {
+ retCode = ((BKException) throwable).getCode();
+ } else {
+ retCode = Code.UnexpectedConditionException;
+ }
+ callback.operationComplete(retCode, resultList);
}
}
};
@@ -169,13 +190,9 @@ public void readComplete(int rc, LedgerHandle lh,
Enumeration<LedgerEntry> entri
}
long entryId = recoveryData.lastAddConfirmed;
- PendingReadOp readOp = new PendingReadOp(lh, lh.bk.scheduler,
entryId, entryId, readCallback, entryId);
- try {
- readOp.initiate();
- } catch (Throwable t) {
- logger.error("Failed to initialize pending read entry {} for
ledger {} : ",
- new Object[] { entryId, lh.getLedgerMetadata(), t
});
- }
+ PendingReadOp op = new PendingReadOp(lh, lh.bk.scheduler, entryId,
entryId, false);
+ op.future().whenComplete(readListener);
+ op.submit();
};
// Read Last AddConfirmed
new ReadLastConfirmedOp(lh, readLACCallback).initiate();
diff --git a/pom.xml b/pom.xml
index 616d755c..738ef81e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -100,7 +100,7 @@
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<!-- dependencies -->
- <bookkeeper.version>4.6.0-SNAPSHOT</bookkeeper.version>
+ <bookkeeper.version>4.7.0-SNAPSHOT</bookkeeper.version>
<codahale.metrics.version>3.0.1</codahale.metrics.version>
<commons-cli.version>1.1</commons-cli.version>
<commons-codec.version>1.6</commons-codec.version>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
