reddycharan commented on a change in pull request #1201: ISSUE #570: Entrylog
per ledger
URL: https://github.com/apache/bookkeeper/pull/1201#discussion_r171113647
##########
File path:
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryMemTable.java
##########
@@ -263,6 +285,81 @@ private long flushSnapshot(final SkipListFlusher flusher,
Checkpoint checkpoint)
return size;
}
+ long flushSnapshotParallelly(final SkipListFlusher flusher, Checkpoint
checkpoint) throws IOException {
+ AtomicLong flushedSize = new AtomicLong();
+ if (this.snapshot.compareTo(checkpoint) < 0) {
+ synchronized (this) {
+ EntrySkipList keyValues = this.snapshot;
+ if (keyValues.compareTo(checkpoint) < 0) {
+ NavigableSet<EntryKey> keyValuesSet = keyValues.keySet();
+ Map<Long, List<EntryKeyValue>> entryKeyValuesMap = new
HashMap<Long, List<EntryKeyValue>>();
+
+ for (EntryKey key : keyValuesSet) {
+ EntryKeyValue kv = (EntryKeyValue) key;
+ Long ledger = kv.getLedgerId();
+ if (!entryKeyValuesMap.containsKey(ledger)) {
+ entryKeyValuesMap.put(ledger, new
LinkedList<EntryKeyValue>());
+ }
+ entryKeyValuesMap.get(ledger).add(kv);
+ }
+
+ CountDownLatch latch = new
CountDownLatch(entryKeyValuesMap.size());
+ AtomicBoolean isFlushThreadInterrupted = new
AtomicBoolean(false);
+ AtomicReference<Exception>
exceptionWhileFlushingParallelly = new AtomicReference<Exception>();
+ Thread mainFlushThread = Thread.currentThread();
+
+ for (Long ledgerId : entryKeyValuesMap.keySet()) {
+ List<EntryKeyValue> entryKeyValuesOfALedger =
entryKeyValuesMap.get(ledgerId);
+ flushExecutor.submitOrdered(ledgerId, new
SafeRunnable() {
+ @Override
+ public void safeRun() {
+ for (EntryKeyValue entryKeyValue :
entryKeyValuesOfALedger) {
+ try {
+ flusher.process(ledgerId,
entryKeyValue.getEntryId(),
+
entryKeyValue.getValueAsByteBuffer());
+
flushedSize.addAndGet(entryKeyValue.getLength());
+ } catch (NoLedgerException exception) {
+ logger.info("Got NoLedgerException
while flushing entry: {}. The ledger "
+ + "must be deleted " + "after
this entry is added to the Memtable",
+ entryKeyValue);
+ break;
+ } catch (Exception exc) {
+ logger.error(
+ "Got Exception while trying to
flush process entry: " + entryKeyValue,
+ exc);
+ if
(isFlushThreadInterrupted.compareAndSet(false, true)) {
+
exceptionWhileFlushingParallelly.set(exc);
Review comment:
OrderedSafeExecutor doesn't provides cancel API. Also, I don't think it as
an issue if we don't cancel the pending writes. This flush operation is
considered failure and it will be tried. Even if we try to cancel/remove the
pending tasks there isn't any guarantee that it would be cancelled/removed.
----------------------------------------------------------------
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
