reddycharan commented on a change in pull request #1201: ISSUE #570: Entrylog
per ledger
URL: https://github.com/apache/bookkeeper/pull/1201#discussion_r171040651
##########
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);
+ mainFlushThread.interrupt();
+ }
+ // return without countdowning the
latch since we got unexpected Exception
+ return;
+ }
+ }
+ latch.countDown();
+ }
+ });
+ }
+
+ try {
+ while (!latch.await(memtableFlushTimeoutInSeconds,
TimeUnit.SECONDS)) {
+ logger.error("Entrymemtable parallel flush has not
completed in {0} secs, so waiting again",
Review comment:
no it doesn't, that was mistake. will fix that
----------------------------------------------------------------
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
