You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by kt...@apache.org on 2019/01/10 15:18:22 UTC
[accumulo] branch 1.9 updated: During tablet recovery filter logs
out that do not define tablet. (#881)
This is an automated email from the ASF dual-hosted git repository.
kturner pushed a commit to branch 1.9
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/1.9 by this push:
new 577dd9a During tablet recovery filter logs out that do not define tablet. (#881)
577dd9a is described below
commit 577dd9a643574b79700ed3de076eb84f9d4d8124
Author: Keith Turner <kt...@apache.org>
AuthorDate: Thu Jan 10 10:18:17 2019 -0500
During tablet recovery filter logs out that do not define tablet. (#881)
In a situation where a tablet had lots of write ahead logs, log
recovery consumed a lot of memory. The reason for the memory
consumption was that all recovery logs are opened at once. This
means the index for each file and a few key/values from each file
are read into memory.
This patch modifies the code to first open logs one at time to see if
they define the tablet. Then only the logs that define the tablet are
opened at the same time. This has the potential to use less memory
when a subset of the logs define a tablet.
---
.../org/apache/accumulo/tserver/TabletServer.java | 2 +-
.../accumulo/tserver/log/RecoveryLogsIterator.java | 10 +++-
.../accumulo/tserver/log/SortedLogRecovery.java | 62 ++++++++++++++++++++--
3 files changed, 68 insertions(+), 6 deletions(-)
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
index dc4d349..c30b443 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
@@ -3279,7 +3279,7 @@ public class TabletServer extends AccumuloServerContext implements Runnable {
Path finished = RecoveryPath.getRecoveryPath(fs,
fs.getFullPath(FileType.WAL, entry.filename));
finished = SortedLogState.getFinishedMarkerPath(finished);
- TabletServer.log.info("Looking for " + finished);
+ TabletServer.log.debug("Looking for " + finished);
if (fs.exists(finished)) {
recovery = finished.getParent();
}
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java
index a68a4ec..8912aeb 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java
@@ -52,7 +52,15 @@ public class RecoveryLogsIterator implements CloseableIterator<Entry<LogFileKey,
try {
for (Path log : recoveryLogPaths) {
- iterators.add(new RecoveryLogReader(fs, log, start, end));
+ LOG.debug("Opening recovery log {}", log.getName());
+ RecoveryLogReader rlr = new RecoveryLogReader(fs, log, start, end);
+ if (rlr.hasNext()) {
+ LOG.debug("Write ahead log {} has data in range {} {}", log.getName(), start, end);
+ iterators.add(rlr);
+ } else {
+ LOG.debug("Write ahead log {} has no data in range {} {}", log.getName(), start, end);
+ rlr.close();
+ }
}
iter = Iterators.mergeSorted(iterators, new Comparator<Entry<LogFileKey,LogFileValue>>() {
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/SortedLogRecovery.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/SortedLogRecovery.java
index 642064f..211e0f2 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/SortedLogRecovery.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/SortedLogRecovery.java
@@ -24,10 +24,16 @@ import static org.apache.accumulo.tserver.logger.LogEvents.MANY_MUTATIONS;
import static org.apache.accumulo.tserver.logger.LogEvents.MUTATION;
import java.io.IOException;
+import java.util.AbstractMap;
+import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
+import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
@@ -122,6 +128,48 @@ public class SortedLogRecovery {
return tabletId;
}
+ /**
+ * This function opens recovery logs one at a time to see if they define the tablet. This is done
+ * so that later recovery steps that open all of the logs at once can possibly open a smaller set
+ * of logs. Opening a recovery log requires holding its index in memory and few key/values from
+ * it. For a lot of recovery logs this could possibly be a lot of memory.
+ *
+ * @return The maximum tablet ID observed AND the list of logs that contained the maximum tablet
+ * ID.
+ */
+ private Entry<Integer,List<Path>> findLogsThatDefineTablet(KeyExtent extent,
+ List<Path> recoveryLogs) throws IOException {
+ Map<Integer,List<Path>> logsThatDefineTablet = new HashMap<>();
+
+ for (Path wal : recoveryLogs) {
+ int tabletId = findMaxTabletId(extent, Collections.singletonList(wal));
+ if (tabletId != -1) {
+ List<Path> perIdList = logsThatDefineTablet.get(tabletId);
+ if (perIdList == null) {
+ perIdList = new ArrayList<>();
+ logsThatDefineTablet.put(tabletId, perIdList);
+
+ }
+ perIdList.add(wal);
+ log.debug("Found tablet {} with id {} in recovery log {}", extent, tabletId, wal.getName());
+ } else {
+ log.debug("Did not find tablet {} in recovery log {}", extent, wal.getName());
+ }
+ }
+
+ if (logsThatDefineTablet.isEmpty()) {
+ return new AbstractMap.SimpleEntry<Integer,List<Path>>(-1, Collections.<Path> emptyList());
+ } else {
+ return Collections.max(logsThatDefineTablet.entrySet(),
+ new Comparator<Entry<Integer,List<Path>>>() {
+ @Override
+ public int compare(Entry<Integer,List<Path>> o1, Entry<Integer,List<Path>> o2) {
+ return Integer.compare(o1.getKey(), o2.getKey());
+ }
+ });
+ }
+ }
+
private String getPathSuffix(String pathString) {
Path path = new Path(pathString);
if (path.depth() < 2)
@@ -258,23 +306,29 @@ public class SortedLogRecovery {
public void recover(KeyExtent extent, List<Path> recoveryLogs, Set<String> tabletFiles,
MutationReceiver mr) throws IOException {
+ Entry<Integer,List<Path>> maxEntry = findLogsThatDefineTablet(extent, recoveryLogs);
+
// A tablet may leave a tserver and then come back, in which case it would have a different and
// higher tablet id. Only want to consider events in the log related to the last time the tablet
// was loaded.
- int tabletId = findMaxTabletId(extent, recoveryLogs);
+ int tabletId = maxEntry.getKey();
+ List<Path> logsThatDefineTablet = maxEntry.getValue();
if (tabletId == -1) {
log.info("Tablet {} is not defined in recovery logs {} ", extent, asNames(recoveryLogs));
return;
+ } else {
+ log.info("Found {} of {} logs with max id {} for tablet {}", logsThatDefineTablet.size(),
+ recoveryLogs.size(), tabletId, extent);
}
// Find the seq # for the last compaction that started and finished
- long recoverySeq = findRecoverySeq(recoveryLogs, tabletFiles, tabletId);
+ long recoverySeq = findRecoverySeq(logsThatDefineTablet, tabletFiles, tabletId);
log.info("Recovering mutations, tablet:{} tabletId:{} seq:{} logs:{}", extent, tabletId,
- recoverySeq, asNames(recoveryLogs));
+ recoverySeq, asNames(logsThatDefineTablet));
// Replay all mutations that were written after the last successful compaction started.
- playbackMutations(recoveryLogs, mr, tabletId, recoverySeq);
+ playbackMutations(logsThatDefineTablet, mr, tabletId, recoverySeq);
}
}