This is an automated email from the ASF dual-hosted git repository.
kturner pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/master by this push:
new 0acb7f8 fix #936 store root tablet file list in Zookeeper (#1313)
0acb7f8 is described below
commit 0acb7f898460c482612678db9585ba1a4fa8389c
Author: Keith Turner <ktur...@apache.org>
AuthorDate: Wed Aug 14 15:18:04 2019 -0400
fix #936 store root tablet file list in Zookeeper (#1313)
---
.../apache/accumulo/core/metadata/RootTable.java | 5 +
.../accumulo/core/metadata/schema/Ample.java | 34 ++++-
.../core/metadata/schema/RootTabletMetadata.java | 13 +-
.../core/metadata/schema/TabletsMetadata.java | 39 +++---
server/base/pom.xml | 4 +
.../org/apache/accumulo/server/fs/VolumeUtil.java | 81 ++----------
.../apache/accumulo/server/init/Initialize.java | 20 ++-
.../accumulo/server/metadata/RootGcCandidates.java | 117 +++++++++++++++++
.../accumulo/server/metadata/ServerAmpleImpl.java | 99 ++++++++++++++-
.../accumulo/server/util/MasterMetadataUtil.java | 29 -----
.../accumulo/server/util/MetadataTableUtil.java | 50 ++------
.../apache/accumulo/gc/SimpleGarbageCollector.java | 95 ++++++--------
.../accumulo/master/upgrade/Upgrader9to10.java | 132 +++++++++++++++++++
.../master/state/RootTabletStateStoreTest.java | 3 +-
.../master/upgrade/RootFilesUpgradeTest.java} | 48 +++++--
.../org/apache/accumulo/tserver/TabletServer.java | 7 +-
.../accumulo/tserver/tablet/DatafileManager.java | 76 +++--------
.../apache/accumulo/tserver/tablet/RootFiles.java | 139 ---------------------
.../org/apache/accumulo/tserver/tablet/Tablet.java | 32 +----
.../apache/accumulo/tserver/tablet/TabletData.java | 53 --------
20 files changed, 555 insertions(+), 521 deletions(-)
diff --git
a/core/src/main/java/org/apache/accumulo/core/metadata/RootTable.java
b/core/src/main/java/org/apache/accumulo/core/metadata/RootTable.java
index 4e08c46..d943f17 100644
--- a/core/src/main/java/org/apache/accumulo/core/metadata/RootTable.java
+++ b/core/src/main/java/org/apache/accumulo/core/metadata/RootTable.java
@@ -37,6 +37,11 @@ public class RootTable {
*/
public static final String ZROOT_TABLET = ROOT_TABLET_LOCATION;
+ /**
+ * ZK path relative to the zookeeper node where the root tablet gc
candidates are stored.
+ */
+ public static final String ZROOT_TABLET_GC_CANDIDATES = ZROOT_TABLET +
"/gc_candidates";
+
public static final KeyExtent EXTENT = new KeyExtent(ID, null, null);
public static final KeyExtent OLD_EXTENT =
new KeyExtent(MetadataTable.ID, TabletsSection.getRow(MetadataTable.ID,
null), null);
diff --git
a/core/src/main/java/org/apache/accumulo/core/metadata/schema/Ample.java
b/core/src/main/java/org/apache/accumulo/core/metadata/schema/Ample.java
index 88fefa1..9b8a692 100644
--- a/core/src/main/java/org/apache/accumulo/core/metadata/schema/Ample.java
+++ b/core/src/main/java/org/apache/accumulo/core/metadata/schema/Ample.java
@@ -18,9 +18,12 @@
package org.apache.accumulo.core.metadata.schema;
import java.util.Collection;
+import java.util.Iterator;
import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.metadata.MetadataTable;
+import org.apache.accumulo.core.metadata.RootTable;
import org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType;
import org.apache.accumulo.core.metadata.schema.TabletMetadata.LocationType;
import org.apache.accumulo.core.tabletserver.log.LogEntry;
@@ -56,6 +59,31 @@ import org.apache.hadoop.io.Text;
public interface Ample {
/**
+ * Accumulo is a distributed tree with three levels. This enum is used to
communicate to Ample
+ * that code is interested in operating on the metadata of a data level.
Sometimes tables ids or
+ * key extents are passed to Ample in lieu of a data level, in these cases
the data level is
+ * derived from the table id.
+ */
+ public enum DataLevel {
+ ROOT(null), METADATA(RootTable.NAME), USER(MetadataTable.NAME);
+
+ private final String table;
+
+ private DataLevel(String table) {
+ this.table = table;
+ }
+
+ /**
+ * @return The name of the Accumulo table in which this data level stores
its metadata.
+ */
+ public String metaTable() {
+ if (table == null)
+ throw new UnsupportedOperationException();
+ return table;
+ }
+ }
+
+ /**
* Read a single tablets metadata. No checking is done for prev row, so it
could differ.
*
* @param extent
@@ -90,7 +118,11 @@ public interface Ample {
throw new UnsupportedOperationException();
}
- default void deleteGcCandidates(TableId tableId, Collection<String> paths) {
+ default void deleteGcCandidates(DataLevel level, Collection<String> paths) {
+ throw new UnsupportedOperationException();
+ }
+
+ default Iterator<String> getGcCandidates(DataLevel level, String
continuePoint) {
throw new UnsupportedOperationException();
}
diff --git
a/core/src/main/java/org/apache/accumulo/core/metadata/schema/RootTabletMetadata.java
b/core/src/main/java/org/apache/accumulo/core/metadata/schema/RootTabletMetadata.java
index 59b4244..33790ec 100644
---
a/core/src/main/java/org/apache/accumulo/core/metadata/schema/RootTabletMetadata.java
+++
b/core/src/main/java/org/apache/accumulo/core/metadata/schema/RootTabletMetadata.java
@@ -26,6 +26,7 @@ import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeMap;
+import org.apache.accumulo.core.client.admin.TimeType;
import org.apache.accumulo.core.data.ArrayByteSequence;
import org.apache.accumulo.core.data.ByteSequence;
import org.apache.accumulo.core.data.ColumnUpdate;
@@ -34,6 +35,8 @@ import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.metadata.RootTable;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
+import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.DataFileColumnFamily;
+import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily;
import org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType;
import org.apache.hadoop.io.Text;
@@ -175,10 +178,14 @@ public class RootTabletMetadata {
/**
* Generate initial json for the root tablet metadata.
*/
- public static byte[] getInitialJson(String dir) {
+ public static byte[] getInitialJson(String dir, String file) {
Mutation mutation = RootTable.EXTENT.getPrevRowUpdateMutation();
- TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(mutation,
- new Value(dir.getBytes(UTF_8)));
+ ServerColumnFamily.DIRECTORY_COLUMN.put(mutation, new
Value(dir.getBytes(UTF_8)));
+
+ mutation.put(DataFileColumnFamily.STR_NAME, file, new DataFileValue(0,
0).encodeAsValue());
+
+ ServerColumnFamily.TIME_COLUMN.put(mutation,
+ new Value(new MetadataTime(0, TimeType.LOGICAL).encode()));
RootTabletMetadata rtm = new RootTabletMetadata();
diff --git
a/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletsMetadata.java
b/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletsMetadata.java
index 3af28b3..1ae3229 100644
---
a/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletsMetadata.java
+++
b/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletsMetadata.java
@@ -42,6 +42,7 @@ import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.dataImpl.KeyExtent;
import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.RootTable;
+import org.apache.accumulo.core.metadata.schema.Ample.DataLevel;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.BulkFileColumnFamily;
import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ClonedColumnFamily;
@@ -69,7 +70,8 @@ public class TabletsMetadata implements
Iterable<TabletMetadata>, AutoCloseable
private List<Text> families = new ArrayList<>();
private List<ColumnFQ> qualifiers = new ArrayList<>();
- private String table = MetadataTable.NAME;
+ private Ample.DataLevel level;
+ private String table;
private Range range;
private EnumSet<ColumnType> fetchedCols = EnumSet.noneOf(ColumnType.class);
private Text endRow;
@@ -77,30 +79,26 @@ public class TabletsMetadata implements
Iterable<TabletMetadata>, AutoCloseable
private boolean saveKeyValues;
private TableId tableId;
- // An internal constant that represents a fictional table where the root
tablet stores its
- // metadata
- private static String SEED_TABLE = "accumulo.seed";
-
@Override
public TabletsMetadata build(AccumuloClient client) {
- if (table.equals(SEED_TABLE)) {
- return buildSeed(client);
+ Preconditions.checkState(level == null ^ table == null);
+ if (level == DataLevel.ROOT) {
+ ClientContext ctx = ((ClientContext) client);
+ ZooCache zc = ctx.getZooCache();
+ String zkRoot = ctx.getZooKeeperRoot();
+ return new TabletsMetadata(getRootMetadata(zkRoot, zc));
} else {
- return buildNonSeed(client);
+ return buildNonRoot(client);
}
}
- private TabletsMetadata buildSeed(AccumuloClient client) {
- ClientContext ctx = ((ClientContext) client);
- ZooCache zc = ctx.getZooCache();
- String zkRoot = ctx.getZooKeeperRoot();
+ private TabletsMetadata buildNonRoot(AccumuloClient client) {
+ try {
- return new TabletsMetadata(getRootMetadata(zkRoot, zc));
- }
+ String resolvedTable = table == null ? level.metaTable() : table;
- private TabletsMetadata buildNonSeed(AccumuloClient client) {
- try {
- Scanner scanner = new IsolatedScanner(client.createScanner(table,
Authorizations.EMPTY));
+ Scanner scanner =
+ new IsolatedScanner(client.createScanner(resolvedTable,
Authorizations.EMPTY));
scanner.setRange(range);
if (checkConsistency && !fetchedCols.contains(ColumnType.PREV_ROW)) {
@@ -198,11 +196,11 @@ public class TabletsMetadata implements
Iterable<TabletMetadata>, AutoCloseable
@Override
public TableRangeOptions forTable(TableId tableId) {
if (tableId.equals(RootTable.ID)) {
- this.table = SEED_TABLE;
+ this.level = DataLevel.ROOT;
} else if (tableId.equals(MetadataTable.ID)) {
- this.table = RootTable.NAME;
+ this.level = DataLevel.METADATA;
} else {
- this.table = MetadataTable.NAME;
+ this.level = DataLevel.USER;
}
this.tableId = tableId;
@@ -239,7 +237,6 @@ public class TabletsMetadata implements
Iterable<TabletMetadata>, AutoCloseable
@Override
public RangeOptions scanTable(String tableName) {
- Preconditions.checkArgument(!tableName.equals(SEED_TABLE));
this.table = tableName;
this.range = TabletsSection.getRange();
return this;
diff --git a/server/base/pom.xml b/server/base/pom.xml
index a963813..c7b544d 100644
--- a/server/base/pom.xml
+++ b/server/base/pom.xml
@@ -37,6 +37,10 @@
<optional>true</optional>
</dependency>
<dependency>
+ <groupId>com.google.code.gson</groupId>
+ <artifactId>gson</artifactId>
+ </dependency>
+ <dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
</dependency>
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java
b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java
index 04d1d1f..4f0b3ab 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java
@@ -18,7 +18,6 @@ package org.apache.accumulo.server.fs;
import java.io.FileNotFoundException;
import java.io.IOException;
-import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
@@ -44,7 +43,6 @@ import org.apache.commons.codec.digest.DigestUtils;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -55,7 +53,6 @@ import org.slf4j.LoggerFactory;
public class VolumeUtil {
private static final Logger log = LoggerFactory.getLogger(VolumeUtil.class);
- private static final SecureRandom rand = new SecureRandom();
private static boolean isActiveVolume(ServerContext context, Path dir) {
@@ -203,21 +200,17 @@ public class VolumeUtil {
}
}
- if (extent.isRootTablet()) {
- ret.datafiles = tabletFiles.datafiles;
- } else {
- for (Entry<FileRef,DataFileValue> entry :
tabletFiles.datafiles.entrySet()) {
- String metaPath = entry.getKey().meta().toString();
- String switchedPath = switchVolume(metaPath, FileType.TABLE,
replacements);
- if (switchedPath != null) {
- filesToRemove.add(entry.getKey());
- FileRef switchedRef = new FileRef(switchedPath, new
Path(switchedPath));
- filesToAdd.put(switchedRef, entry.getValue());
- ret.datafiles.put(switchedRef, entry.getValue());
- log.debug("Replacing volume {} : {} -> {}", extent, metaPath,
switchedPath);
- } else {
- ret.datafiles.put(entry.getKey(), entry.getValue());
- }
+ for (Entry<FileRef,DataFileValue> entry :
tabletFiles.datafiles.entrySet()) {
+ String metaPath = entry.getKey().meta().toString();
+ String switchedPath = switchVolume(metaPath, FileType.TABLE,
replacements);
+ if (switchedPath != null) {
+ filesToRemove.add(entry.getKey());
+ FileRef switchedRef = new FileRef(switchedPath, new
Path(switchedPath));
+ filesToAdd.put(switchedRef, entry.getValue());
+ ret.datafiles.put(switchedRef, entry.getValue());
+ log.debug("Replacing volume {} : {} -> {}", extent, metaPath,
switchedPath);
+ } else {
+ ret.datafiles.put(entry.getKey(), entry.getValue());
}
}
@@ -276,52 +269,10 @@ public class VolumeUtil {
+ Path.SEPARATOR + dir.getName());
log.info("Updating directory for {} from {} to {}", extent, dir, newDir);
- if (extent.isRootTablet()) {
- // the root tablet is special case, its files need to be copied if its
dir is changed
-
- // this code needs to be idempotent
-
- FileSystem fs1 = vm.getVolumeByPath(dir).getFileSystem();
- FileSystem fs2 = vm.getVolumeByPath(newDir).getFileSystem();
-
- if (!same(fs1, dir, fs2, newDir)) {
- if (fs2.exists(newDir)) {
- Path newDirBackup = getBackupName(newDir);
- // never delete anything because were dealing with the root tablet
- // one reason this dir may exist is because this method failed
previously
- log.info("renaming {} to {}", newDir, newDirBackup);
- if (!fs2.rename(newDir, newDirBackup)) {
- throw new IOException("Failed to rename " + newDir + " to " +
newDirBackup);
- }
- }
-
- // do a lot of logging since this is the root tablet
- log.info("copying {} to {}", dir, newDir);
- if (!FileUtil.copy(fs1, dir, fs2, newDir, false,
context.getHadoopConf())) {
- throw new IOException("Failed to copy " + dir + " to " + newDir);
- }
-
- // only set the new location in zookeeper after a successful copy
- log.info("setting root tablet location to {}", newDir);
-
context.getAmple().mutateTablet(RootTable.EXTENT).putDir(newDir.toString()).mutate();
- // rename the old dir to avoid confusion when someone looks at
filesystem... its ok if we
- // fail here and this does not happen because the location in
- // zookeeper is the authority
- Path dirBackup = getBackupName(dir);
- log.info("renaming {} to {}", dir, dirBackup);
- fs1.rename(dir, dirBackup);
+ MetadataTableUtil.updateTabletDir(extent, newDir.toString(), context,
zooLock);
+ return newDir.toString();
- } else {
- log.info("setting root tablet location to {}", newDir);
-
context.getAmple().mutateTablet(RootTable.EXTENT).putDir(newDir.toString()).mutate();
- }
-
- return newDir.toString();
- } else {
- MetadataTableUtil.updateTabletDir(extent, newDir.toString(), context,
zooLock);
- return newDir.toString();
- }
}
static boolean same(FileSystem fs1, Path dir, FileSystem fs2, Path newDir)
@@ -372,10 +323,4 @@ public class VolumeUtil {
}
}
-
- private static Path getBackupName(Path path) {
- return new Path(path.getParent(), path.getName() + "_" +
System.currentTimeMillis() + "_"
- + (rand.nextInt(Integer.MAX_VALUE) + 1) + ".bak");
- }
-
}
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/init/Initialize.java
b/server/base/src/main/java/org/apache/accumulo/server/init/Initialize.java
index a19b47b..236637b 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/init/Initialize.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/init/Initialize.java
@@ -96,6 +96,7 @@ import org.apache.accumulo.server.fs.VolumeManager;
import org.apache.accumulo.server.fs.VolumeManagerImpl;
import org.apache.accumulo.server.iterators.MetadataBulkLoadFilter;
import org.apache.accumulo.server.log.WalStateManager;
+import org.apache.accumulo.server.metadata.RootGcCandidates;
import org.apache.accumulo.server.replication.ReplicationUtil;
import org.apache.accumulo.server.replication.StatusCombiner;
import org.apache.accumulo.server.security.AuditedSecurityOperation;
@@ -365,15 +366,18 @@ public class Initialize implements KeywordExecutable {
fs.choose(chooserEnv, configuredVolumes) + Path.SEPARATOR +
ServerConstants.TABLE_DIR
+ Path.SEPARATOR + RootTable.ID +
RootTable.ROOT_TABLET_LOCATION).toString();
+ String ext =
FileOperations.getNewFileExtension(DefaultConfiguration.getInstance());
+ String rootTabletFileName = rootTabletDir + Path.SEPARATOR +
"00000_00000." + ext;
+
try {
- initZooKeeper(opts, uuid.toString(), instanceNamePath, rootTabletDir);
+ initZooKeeper(opts, uuid.toString(), instanceNamePath, rootTabletDir,
rootTabletFileName);
} catch (Exception e) {
log.error("FATAL: Failed to initialize zookeeper", e);
return false;
}
try {
- initFileSystem(siteConfig, hadoopConf, fs, uuid, rootTabletDir);
+ initFileSystem(siteConfig, hadoopConf, fs, uuid, rootTabletDir,
rootTabletFileName);
} catch (Exception e) {
log.error("FATAL Failed to initialize filesystem", e);
@@ -489,7 +493,8 @@ public class Initialize implements KeywordExecutable {
}
private void initFileSystem(SiteConfiguration siteConfig, Configuration
hadoopConf,
- VolumeManager fs, UUID uuid, String rootTabletDir) throws IOException {
+ VolumeManager fs, UUID uuid, String rootTabletDir, String
rootTabletFileName)
+ throws IOException {
initDirs(fs, uuid, VolumeConfiguration.getVolumeUris(siteConfig,
hadoopConf), false);
// initialize initial system tables config in zookeeper
@@ -523,7 +528,6 @@ public class Initialize implements KeywordExecutable {
createMetadataFile(fs, metadataFileName, siteConfig, replicationTablet);
// populate the root tablet with info about the metadata table's two
initial tablets
- String rootTabletFileName = rootTabletDir + Path.SEPARATOR +
"00000_00000." + ext;
Text splitPoint = TabletsSection.getRange().getEndKey().getRow();
Tablet tablesTablet =
new Tablet(MetadataTable.ID, tableMetadataTabletDir, null, splitPoint,
metadataFileName);
@@ -603,7 +607,8 @@ public class Initialize implements KeywordExecutable {
}
private static void initZooKeeper(Opts opts, String uuid, String
instanceNamePath,
- String rootTabletDir) throws KeeperException, InterruptedException {
+ String rootTabletDir, String rootTabletFileName)
+ throws KeeperException, InterruptedException {
// setup basic data in zookeeper
zoo.putPersistentData(Constants.ZROOT, new byte[0], -1,
NodeExistsPolicy.SKIP,
Ids.OPEN_ACL_UNSAFE);
@@ -641,7 +646,10 @@ public class Initialize implements KeywordExecutable {
zoo.putPersistentData(zkInstanceRoot + Constants.ZPROBLEMS,
EMPTY_BYTE_ARRAY,
NodeExistsPolicy.FAIL);
zoo.putPersistentData(zkInstanceRoot + RootTable.ZROOT_TABLET,
- RootTabletMetadata.getInitialJson(rootTabletDir),
NodeExistsPolicy.FAIL);
+ RootTabletMetadata.getInitialJson(rootTabletDir, rootTabletFileName),
+ NodeExistsPolicy.FAIL);
+ zoo.putPersistentData(zkInstanceRoot +
RootTable.ZROOT_TABLET_GC_CANDIDATES,
+ new RootGcCandidates().toJson().getBytes(UTF_8),
NodeExistsPolicy.FAIL);
zoo.putPersistentData(zkInstanceRoot + Constants.ZMASTERS,
EMPTY_BYTE_ARRAY,
NodeExistsPolicy.FAIL);
zoo.putPersistentData(zkInstanceRoot + Constants.ZMASTER_LOCK,
EMPTY_BYTE_ARRAY,
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/metadata/RootGcCandidates.java
b/server/base/src/main/java/org/apache/accumulo/server/metadata/RootGcCandidates.java
new file mode 100644
index 0000000..decde0c
--- /dev/null
+++
b/server/base/src/main/java/org/apache/accumulo/server/metadata/RootGcCandidates.java
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.accumulo.server.metadata;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import java.util.Collection;
+import java.util.SortedMap;
+import java.util.SortedSet;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.stream.Stream;
+
+import org.apache.accumulo.core.metadata.schema.Ample.FileMeta;
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.base.Preconditions;
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+
+public class RootGcCandidates {
+ private static final Gson GSON = new
GsonBuilder().setPrettyPrinting().create();
+
+ // This class is used to serialize and deserialize root tablet metadata
using GSon. Any changes to
+ // this class must consider persisted data.
+ private static class GSonData {
+ int version = 1;
+
+ // SortedMap<dir path, SortedSet<file name>>
+ SortedMap<String,SortedSet<String>> candidates;
+ }
+
+ /*
+ * The root tablet will only have a single dir on each volume. Therefore
root file paths will have
+ * a small set of unique prefixes. The following map is structured to avoid
storing the same dir
+ * prefix over and over in JSon and java.
+ *
+ * SortedMap<dir path, SortedSet<file name>>
+ */
+ private SortedMap<String,SortedSet<String>> candidates;
+
+ public RootGcCandidates() {
+ this.candidates = new TreeMap<>();
+ }
+
+ private RootGcCandidates(SortedMap<String,SortedSet<String>> candidates) {
+ this.candidates = candidates;
+ }
+
+ public void add(Collection<? extends FileMeta> refs) {
+ refs.forEach(ref -> {
+ Path path = ref.path();
+
+ String parent = path.getParent().toString();
+ String name = path.getName();
+
+ candidates.computeIfAbsent(parent, k -> new TreeSet<>()).add(name);
+ });
+ }
+
+ public void remove(Collection<String> refs) {
+ refs.forEach(ref -> {
+ Path path = new Path(ref);
+ String parent = path.getParent().toString();
+ String name = path.getName();
+
+ SortedSet<String> names = candidates.get(parent);
+ if (names != null) {
+ names.remove(name);
+ if (names.isEmpty()) {
+ candidates.remove(parent);
+ }
+ }
+ });
+ }
+
+ public Stream<String> stream() {
+ return candidates.entrySet().stream().flatMap(entry -> {
+ String parent = entry.getKey();
+ SortedSet<String> names = entry.getValue();
+ return names.stream().map(name -> new Path(parent, name).toString());
+ });
+ }
+
+ public String toJson() {
+ GSonData gd = new GSonData();
+ gd.candidates = candidates;
+ return GSON.toJson(gd);
+ }
+
+ public static RootGcCandidates fromJson(String json) {
+ GSonData gd = GSON.fromJson(json, GSonData.class);
+
+ Preconditions.checkArgument(gd.version == 1);
+
+ return new RootGcCandidates(gd.candidates);
+ }
+
+ public static RootGcCandidates fromJson(byte[] json) {
+ return fromJson(new String(json, UTF_8));
+ }
+}
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/metadata/ServerAmpleImpl.java
b/server/base/src/main/java/org/apache/accumulo/server/metadata/ServerAmpleImpl.java
index 2f5d9da..ce5f4f3 100644
---
a/server/base/src/main/java/org/apache/accumulo/server/metadata/ServerAmpleImpl.java
+++
b/server/base/src/main/java/org/apache/accumulo/server/metadata/ServerAmpleImpl.java
@@ -17,14 +17,23 @@
package org.apache.accumulo.server.metadata;
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static
org.apache.accumulo.core.metadata.RootTable.ZROOT_TABLET_GC_CANDIDATES;
import static org.apache.accumulo.server.util.MetadataTableUtil.EMPTY_TEXT;
import java.util.Collection;
+import java.util.Iterator;
+import java.util.function.Consumer;
+import java.util.stream.Stream;
import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.MutationsRejectedException;
+import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.PartialKey;
+import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.dataImpl.KeyExtent;
@@ -33,14 +42,21 @@ import org.apache.accumulo.core.metadata.RootTable;
import org.apache.accumulo.core.metadata.schema.Ample;
import org.apache.accumulo.core.metadata.schema.AmpleImpl;
import org.apache.accumulo.core.metadata.schema.MetadataSchema;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.fate.zookeeper.ZooUtil;
import org.apache.accumulo.server.ServerContext;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import com.google.common.base.Preconditions;
+import com.google.common.collect.Iterators;
public class ServerAmpleImpl extends AmpleImpl implements Ample {
+ private static Logger log = LoggerFactory.getLogger(ServerAmpleImpl.class);
+
private ServerContext context;
public ServerAmpleImpl(ServerContext ctx) {
@@ -61,8 +77,44 @@ public class ServerAmpleImpl extends AmpleImpl implements
Ample {
return new TabletsMutatorImpl(context);
}
+ private void mutateRootGcCandidates(Consumer<RootGcCandidates> mutator) {
+ String zpath = context.getZooKeeperRoot() + ZROOT_TABLET_GC_CANDIDATES;
+ try {
+ context.getZooReaderWriter().mutate(zpath, new byte[0], ZooUtil.PUBLIC,
currVal -> {
+ String currJson = new String(currVal, UTF_8);
+
+ RootGcCandidates rgcc = RootGcCandidates.fromJson(currJson);
+
+ log.debug("Root GC candidates before change : {}", currJson);
+
+ mutator.accept(rgcc);
+
+ String newJson = rgcc.toJson();
+
+ log.debug("Root GC candidates after change : {}", newJson);
+
+ if (newJson.length() > 262_144) {
+ log.warn(
+ "Root tablet deletion candidates stored in ZK at {} are getting
large ({} bytes), is"
+ + " Accumulo GC process running? Large nodes may cause
problems for Zookeeper!",
+ zpath, newJson.length());
+ }
+
+ return newJson.getBytes(UTF_8);
+ });
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
@Override
public void putGcCandidates(TableId tableId, Collection<? extends
Ample.FileMeta> candidates) {
+
+ if (RootTable.ID.equals(tableId)) {
+ mutateRootGcCandidates(rgcc -> rgcc.add(candidates));
+ return;
+ }
+
try (BatchWriter writer = createWriter(tableId)) {
for (Ample.FileMeta file : candidates) {
writer.addMutation(createDeleteMutation(context, tableId,
file.path().toString()));
@@ -73,18 +125,59 @@ public class ServerAmpleImpl extends AmpleImpl implements
Ample {
}
@Override
- public void deleteGcCandidates(TableId tableId, Collection<String> paths) {
- try (BatchWriter writer = createWriter(tableId)) {
+ public void deleteGcCandidates(DataLevel level, Collection<String> paths) {
+
+ if (level == DataLevel.ROOT) {
+ mutateRootGcCandidates(rgcc -> rgcc.remove(paths));
+ return;
+ }
+
+ try (BatchWriter writer = context.createBatchWriter(level.metaTable())) {
for (String path : paths) {
Mutation m = new Mutation(MetadataSchema.DeletesSection.getRowPrefix()
+ path);
m.putDelete(EMPTY_TEXT, EMPTY_TEXT);
writer.addMutation(m);
}
- } catch (MutationsRejectedException e) {
+ } catch (MutationsRejectedException | TableNotFoundException e) {
throw new RuntimeException(e);
}
}
+ public Iterator<String> getGcCandidates(DataLevel level, String
continuePoint) {
+ if (level == DataLevel.ROOT) {
+ byte[] json = context.getZooCache()
+ .get(context.getZooKeeperRoot() +
RootTable.ZROOT_TABLET_GC_CANDIDATES);
+ Stream<String> candidates =
RootGcCandidates.fromJson(json).stream().sorted();
+
+ if (continuePoint != null && !continuePoint.isEmpty()) {
+ candidates = candidates.dropWhile(candidate ->
candidate.compareTo(continuePoint) <= 0);
+ }
+
+ return candidates.iterator();
+ } else if (level == DataLevel.METADATA || level == DataLevel.USER) {
+ Range range = MetadataSchema.DeletesSection.getRange();
+ if (continuePoint != null && !continuePoint.isEmpty()) {
+ String continueRow = MetadataSchema.DeletesSection.getRowPrefix() +
continuePoint;
+ range = new Range(new Key(continueRow).followingKey(PartialKey.ROW),
true,
+ range.getEndKey(), range.isEndKeyInclusive());
+ }
+
+ Scanner scanner;
+ try {
+ scanner = context.createScanner(level.metaTable(),
Authorizations.EMPTY);
+ } catch (TableNotFoundException e) {
+ throw new RuntimeException(e);
+ }
+ scanner.setRange(range);
+
+ return Iterators.transform(scanner.iterator(), entry ->
entry.getKey().getRow().toString()
+ .substring(MetadataSchema.DeletesSection.getRowPrefix().length()));
+
+ } else {
+ throw new IllegalArgumentException();
+ }
+ }
+
private BatchWriter createWriter(TableId tableId) {
Preconditions.checkArgument(!RootTable.ID.equals(tableId));
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/util/MasterMetadataUtil.java
b/server/base/src/main/java/org/apache/accumulo/server/util/MasterMetadataUtil.java
index 1c50b0e..55cc43b 100644
---
a/server/base/src/main/java/org/apache/accumulo/server/util/MasterMetadataUtil.java
+++
b/server/base/src/main/java/org/apache/accumulo/server/util/MasterMetadataUtil.java
@@ -39,7 +39,6 @@ import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.dataImpl.KeyExtent;
import org.apache.accumulo.core.metadata.MetadataTable;
-import org.apache.accumulo.core.metadata.RootTable;
import org.apache.accumulo.core.metadata.schema.Ample.TabletMutator;
import org.apache.accumulo.core.metadata.schema.DataFileValue;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
@@ -224,34 +223,6 @@ public class MasterMetadataUtil {
FileRef mergeFile, DataFileValue dfv, MetadataTime time, Set<FileRef>
filesInUseByScans,
String address, ZooLock zooLock, Set<String> unusedWalLogs,
TServerInstance lastLocation,
long flushId) {
- if (extent.isRootTablet()) {
- updateRootTabletDataFile(context, unusedWalLogs);
- } else {
- updateForTabletDataFile(context, extent, path, mergeFile, dfv, time,
filesInUseByScans,
- address, zooLock, unusedWalLogs, lastLocation, flushId);
- }
-
- }
-
- /**
- * Update the data file for the root tablet
- */
- private static void updateRootTabletDataFile(ServerContext context,
Set<String> unusedWalLogs) {
- if (unusedWalLogs != null) {
- TabletMutator tablet = context.getAmple().mutateTablet(RootTable.EXTENT);
- unusedWalLogs.forEach(tablet::deleteWal);
- tablet.mutate();
- }
- }
-
- /**
- * Create an update that updates a tablet
- *
- */
- private static void updateForTabletDataFile(ServerContext context, KeyExtent
extent, FileRef path,
- FileRef mergeFile, DataFileValue dfv, MetadataTime time, Set<FileRef>
filesInUseByScans,
- String address, ZooLock zooLock, Set<String> unusedWalLogs,
TServerInstance lastLocation,
- long flushId) {
TabletMutator tablet = context.getAmple().mutateTablet(extent);
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
b/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
index 57edd93..937e686 100644
---
a/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
+++
b/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
@@ -91,14 +91,12 @@ import
org.apache.accumulo.server.fs.VolumeChooserEnvironment;
import org.apache.accumulo.server.fs.VolumeChooserEnvironmentImpl;
import org.apache.accumulo.server.fs.VolumeManager;
import org.apache.accumulo.server.metadata.ServerAmpleImpl;
-import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
/**
* provides a reference to the metadata table for updates by tablet servers
@@ -166,22 +164,18 @@ public class MetadataTableUtil {
public static void updateTabletFlushID(KeyExtent extent, long flushID,
ServerContext context,
ZooLock zooLock) {
- if (!extent.isRootTablet()) {
- TabletMutator tablet = context.getAmple().mutateTablet(extent);
- tablet.putFlushId(flushID);
- tablet.putZooLock(zooLock);
- tablet.mutate();
- }
+ TabletMutator tablet = context.getAmple().mutateTablet(extent);
+ tablet.putFlushId(flushID);
+ tablet.putZooLock(zooLock);
+ tablet.mutate();
}
public static void updateTabletCompactID(KeyExtent extent, long compactID,
ServerContext context,
ZooLock zooLock) {
- if (!extent.isRootTablet()) {
- TabletMutator tablet = context.getAmple().mutateTablet(extent);
- tablet.putCompactionId(compactID);
- tablet.putZooLock(zooLock);
- tablet.mutate();
- }
+ TabletMutator tablet = context.getAmple().mutateTablet(extent);
+ tablet.putCompactionId(compactID);
+ tablet.putZooLock(zooLock);
+ tablet.mutate();
}
public static void updateTabletDataFile(long tid, KeyExtent extent,
@@ -222,14 +216,6 @@ public class MetadataTableUtil {
SortedMap<FileRef,DataFileValue> filesToAdd, String newDir, ZooLock
zooLock,
ServerContext context) {
- if (extent.isRootTablet()) {
- if (newDir != null)
- throw new IllegalArgumentException("newDir not expected for " +
extent);
-
- if (filesToRemove.size() != 0 || filesToAdd.size() != 0)
- throw new IllegalArgumentException("files not expected for " + extent);
- }
-
TabletMutator tabletMutator = context.getAmple().mutateTablet(extent);
logsToRemove.forEach(tabletMutator::deleteWal);
logsToAdd.forEach(tabletMutator::putWal);
@@ -440,23 +426,9 @@ public class MetadataTableUtil {
result.addAll(tablet.getLogs());
- if (extent.isRootTablet()) {
- Preconditions.checkState(tablet.getFiles().isEmpty(),
- "Saw unexpected files in root tablet metadata %s",
tablet.getFiles());
-
- FileStatus[] files = fs.listStatus(new Path(tablet.getDir()));
- for (FileStatus fileStatus : files) {
- if (fileStatus.getPath().toString().endsWith("_tmp")) {
- continue;
- }
- DataFileValue dfv = new DataFileValue(0, 0);
- sizes.put(new FileRef(fileStatus.getPath().toString(),
fileStatus.getPath()), dfv);
- }
- } else {
- tablet.getFilesMap().forEach((k, v) -> {
- sizes.put(new FileRef(k, fs.getFullPath(tablet.getTableId(), k)), v);
- });
- }
+ tablet.getFilesMap().forEach((k, v) -> {
+ sizes.put(new FileRef(k, fs.getFullPath(tablet.getTableId(), k)), v);
+ });
return new Pair<>(result, sizes);
}
diff --git
a/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
b/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
index 51e0817..54db447 100644
--- a/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
+++ b/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
@@ -23,6 +23,7 @@ import static
org.apache.accumulo.fate.util.UtilWaitThread.sleepUninterruptibly;
import java.io.FileNotFoundException;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
@@ -36,20 +37,12 @@ import java.util.stream.Stream;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.AccumuloClient;
-import org.apache.accumulo.core.client.BatchWriter;
-import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.IsolatedScanner;
-import org.apache.accumulo.core.client.MutationsRejectedException;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.clientImpl.Tables;
import org.apache.accumulo.core.conf.Property;
-import org.apache.accumulo.core.data.Key;
-import org.apache.accumulo.core.data.Mutation;
-import org.apache.accumulo.core.data.PartialKey;
-import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.TableId;
-import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.gc.thrift.GCMonitorService.Iface;
import org.apache.accumulo.core.gc.thrift.GCMonitorService.Processor;
import org.apache.accumulo.core.gc.thrift.GCStatus;
@@ -57,6 +50,8 @@ import org.apache.accumulo.core.gc.thrift.GcCycleStats;
import org.apache.accumulo.core.master.state.tables.TableState;
import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.RootTable;
+import org.apache.accumulo.core.metadata.schema.Ample;
+import org.apache.accumulo.core.metadata.schema.Ample.DataLevel;
import org.apache.accumulo.core.metadata.schema.MetadataSchema;
import org.apache.accumulo.core.metadata.schema.TabletMetadata;
import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
@@ -90,7 +85,6 @@ import org.apache.accumulo.server.rpc.ThriftServerType;
import org.apache.accumulo.server.util.Halt;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.Text;
import org.apache.htrace.Trace;
import org.apache.htrace.TraceScope;
import org.apache.htrace.impl.ProbabilitySampler;
@@ -109,7 +103,6 @@ import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
// the ZK lock is acquired. The server is only for metrics, there are no
concerns about clients
// using the service before the lock is acquired.
public class SimpleGarbageCollector extends AbstractServer implements Iface {
- private static final Text EMPTY_TEXT = new Text();
/**
* Options for the garbage collector.
@@ -185,32 +178,23 @@ public class SimpleGarbageCollector extends
AbstractServer implements Iface {
private class GCEnv implements GarbageCollectionEnvironment {
- private String tableName;
+ private DataLevel level;
- GCEnv(String tableName) {
- this.tableName = tableName;
+ GCEnv(Ample.DataLevel level) {
+ this.level = level;
}
@Override
public boolean getCandidates(String continuePoint, List<String> result)
throws TableNotFoundException {
- // want to ensure GC makes progress... if the 1st N deletes are stable
and we keep processing
- // them,
- // then will never inspect deletes after N
- Range range = MetadataSchema.DeletesSection.getRange();
- if (continuePoint != null && !continuePoint.isEmpty()) {
- String continueRow = MetadataSchema.DeletesSection.getRowPrefix() +
continuePoint;
- range = new Range(new Key(continueRow).followingKey(PartialKey.ROW),
true,
- range.getEndKey(), range.isEndKeyInclusive());
- }
- Scanner scanner = getContext().createScanner(tableName,
Authorizations.EMPTY);
- scanner.setRange(range);
+ Iterator<String> candidates =
getContext().getAmple().getGcCandidates(level, continuePoint);
+
result.clear();
- // find candidates for deletion; chop off the prefix
- for (Entry<Key,Value> entry : scanner) {
- String cand = entry.getKey().getRow().toString()
- .substring(MetadataSchema.DeletesSection.getRowPrefix().length());
+
+ while (candidates.hasNext()) {
+ String cand = candidates.next();
+
result.add(cand);
if (almostOutOfMemory(Runtime.getRuntime())) {
log.info("List of delete candidates has exceeded the memory"
@@ -224,9 +208,14 @@ public class SimpleGarbageCollector extends AbstractServer
implements Iface {
@Override
public Iterator<String> getBlipIterator() throws TableNotFoundException {
+
+ if (level == DataLevel.ROOT) {
+ return Collections.<String>emptySet().iterator();
+ }
+
@SuppressWarnings("resource")
IsolatedScanner scanner =
- new IsolatedScanner(getContext().createScanner(tableName,
Authorizations.EMPTY));
+ new IsolatedScanner(getContext().createScanner(level.metaTable(),
Authorizations.EMPTY));
scanner.setRange(MetadataSchema.BlipSection.getRange());
@@ -237,8 +226,15 @@ public class SimpleGarbageCollector extends AbstractServer
implements Iface {
@Override
public Stream<Reference> getReferences() {
- Stream<TabletMetadata> tabletStream =
TabletsMetadata.builder().scanTable(tableName)
- .checkConsistency().fetch(DIR, FILES,
SCANS).build(getContext()).stream();
+ Stream<TabletMetadata> tabletStream;
+
+ if (level == DataLevel.ROOT) {
+ tabletStream =
+ Stream.of(getContext().getAmple().readTablet(RootTable.EXTENT,
DIR, FILES, SCANS));
+ } else {
+ tabletStream =
TabletsMetadata.builder().scanTable(level.metaTable()).checkConsistency()
+ .fetch(DIR, FILES, SCANS).build(getContext()).stream();
+ }
Stream<Reference> refStream = tabletStream.flatMap(tm -> {
Stream<Reference> refs = Stream.concat(tm.getFiles().stream(),
tm.getScans().stream())
@@ -275,14 +271,13 @@ public class SimpleGarbageCollector extends
AbstractServer implements Iface {
return;
}
- AccumuloClient c = getContext();
- BatchWriter writer = c.createBatchWriter(tableName, new
BatchWriterConfig());
-
// when deleting a dir and all files in that dir, only need to delete
the dir
// the dir will sort right before the files... so remove the files in
this case
// to minimize namenode ops
Iterator<Entry<String,String>> cdIter =
confirmedDeletes.entrySet().iterator();
+ List<String> processedDeletes = Collections.synchronizedList(new
ArrayList<String>());
+
String lastDir = null;
while (cdIter.hasNext()) {
Entry<String,String> entry = cdIter.next();
@@ -294,11 +289,7 @@ public class SimpleGarbageCollector extends AbstractServer
implements Iface {
} else if (lastDir != null) {
if (absPath.startsWith(lastDir)) {
log.debug("Ignoring {} because {} exist", entry.getValue(),
lastDir);
- try {
- putMarkerDeleteMutation(entry.getValue(), writer);
- } catch (MutationsRejectedException e) {
- throw new RuntimeException(e);
- }
+ processedDeletes.add(entry.getValue());
cdIter.remove();
} else {
lastDir = null;
@@ -306,8 +297,6 @@ public class SimpleGarbageCollector extends AbstractServer
implements Iface {
}
}
- final BatchWriter finalWriter = writer;
-
ExecutorService deleteThreadPool =
Executors.newFixedThreadPool(getNumDeleteThreads(), new
NamingThreadFactory("deleting"));
@@ -377,8 +366,8 @@ public class SimpleGarbageCollector extends AbstractServer
implements Iface {
// proceed to clearing out the flags for successful deletes and
// non-existent files
- if (removeFlag && finalWriter != null) {
- putMarkerDeleteMutation(delete, finalWriter);
+ if (removeFlag) {
+ processedDeletes.add(delete);
}
} catch (Exception e) {
log.error("{}", e.getMessage(), e);
@@ -397,13 +386,7 @@ public class SimpleGarbageCollector extends AbstractServer
implements Iface {
log.error("{}", e1.getMessage(), e1);
}
- if (writer != null) {
- try {
- writer.close();
- } catch (MutationsRejectedException e) {
- log.error("Problem removing entries from the metadata table: ", e);
- }
- }
+ getContext().getAmple().deleteGcCandidates(level, processedDeletes);
}
@Override
@@ -501,8 +484,9 @@ public class SimpleGarbageCollector extends AbstractServer
implements Iface {
status.current.started = System.currentTimeMillis();
- new GarbageCollectionAlgorithm().collect(new
GCEnv(RootTable.NAME));
- new GarbageCollectionAlgorithm().collect(new
GCEnv(MetadataTable.NAME));
+ new GarbageCollectionAlgorithm().collect(new
GCEnv(DataLevel.ROOT));
+ new GarbageCollectionAlgorithm().collect(new
GCEnv(DataLevel.METADATA));
+ new GarbageCollectionAlgorithm().collect(new
GCEnv(DataLevel.USER));
log.info("Number of data file candidates for deletion: {}",
status.current.candidates);
log.info("Number of data file candidates still in use: {}",
status.current.inUse);
@@ -654,13 +638,6 @@ public class SimpleGarbageCollector extends AbstractServer
implements Iface {
> CANDIDATE_MEMORY_PERCENTAGE * runtime.maxMemory();
}
- private static void putMarkerDeleteMutation(final String delete, final
BatchWriter writer)
- throws MutationsRejectedException {
- Mutation m = new Mutation(MetadataSchema.DeletesSection.getRowPrefix() +
delete);
- m.putDelete(EMPTY_TEXT, EMPTY_TEXT);
- writer.addMutation(m);
- }
-
/**
* Checks if the given string is a directory.
*
diff --git
a/server/master/src/main/java/org/apache/accumulo/master/upgrade/Upgrader9to10.java
b/server/master/src/main/java/org/apache/accumulo/master/upgrade/Upgrader9to10.java
index 2fcc46b..e2ce6c4 100644
---
a/server/master/src/main/java/org/apache/accumulo/master/upgrade/Upgrader9to10.java
+++
b/server/master/src/main/java/org/apache/accumulo/master/upgrade/Upgrader9to10.java
@@ -19,24 +19,42 @@ package org.apache.accumulo.master.upgrade;
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.apache.accumulo.core.metadata.RootTable.ZROOT_TABLET;
+import static
org.apache.accumulo.core.metadata.RootTable.ZROOT_TABLET_GC_CANDIDATES;
import java.io.IOException;
+import java.io.UncheckedIOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.admin.TimeType;
import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.file.FileOperations;
+import org.apache.accumulo.core.file.FileSKVIterator;
import org.apache.accumulo.core.metadata.RootTable;
+import org.apache.accumulo.core.metadata.schema.DataFileValue;
+import org.apache.accumulo.core.metadata.schema.MetadataTime;
import org.apache.accumulo.core.metadata.schema.RootTabletMetadata;
import org.apache.accumulo.core.metadata.schema.TabletMetadata.LocationType;
import org.apache.accumulo.core.tabletserver.log.LogEntry;
import org.apache.accumulo.core.util.HostAndPort;
import org.apache.accumulo.fate.zookeeper.IZooReaderWriter;
import org.apache.accumulo.fate.zookeeper.ZooUtil;
+import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeExistsPolicy;
import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeMissingPolicy;
import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.server.fs.FileRef;
+import org.apache.accumulo.server.fs.VolumeManager;
import org.apache.accumulo.server.master.state.TServerInstance;
+import org.apache.accumulo.server.metadata.RootGcCandidates;
import org.apache.accumulo.server.metadata.TabletMutatorBase;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -95,9 +113,22 @@ public class Upgrader9to10 implements Upgrader {
logs.forEach(tabletMutator::putWal);
+ Map<String,DataFileValue> files =
cleanupRootTabletFiles(ctx.getVolumeManager(), dir);
+ files.forEach((path, dfv) -> tabletMutator.putFile(new FileRef(path),
dfv));
+
+ tabletMutator.putTime(computeRootTabletTime(ctx, files.keySet()));
+
tabletMutator.mutate();
}
+ try {
+ ctx.getZooReaderWriter().putPersistentData(
+ ctx.getZooKeeperRoot() + ZROOT_TABLET_GC_CANDIDATES,
+ new RootGcCandidates().toJson().getBytes(UTF_8),
NodeExistsPolicy.SKIP);
+ } catch (KeeperException | InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+
// this operation must be idempotent, so deleting after updating is very
important
delete(ctx, ZROOT_TABLET_CURRENT_LOGS);
@@ -220,4 +251,105 @@ public class Upgrader9to10 implements Upgrader {
}
}
+ MetadataTime computeRootTabletTime(ServerContext context, Collection<String>
goodPaths) {
+
+ try {
+ long rtime = Long.MIN_VALUE;
+ for (String good : goodPaths) {
+ Path path = new Path(good);
+
+ FileSystem ns =
context.getVolumeManager().getVolumeByPath(path).getFileSystem();
+ long maxTime = -1;
+ try (FileSKVIterator reader =
FileOperations.getInstance().newReaderBuilder()
+ .forFile(path.toString(), ns, ns.getConf(),
context.getCryptoService())
+ .withTableConfiguration(
+
context.getServerConfFactory().getTableConfiguration(RootTable.ID))
+ .seekToBeginning().build()) {
+ while (reader.hasTop()) {
+ maxTime = Math.max(maxTime, reader.getTopKey().getTimestamp());
+ reader.next();
+ }
+ }
+ if (maxTime > rtime) {
+
+ rtime = maxTime;
+ }
+ }
+
+ if (rtime < 0) {
+ throw new IllegalStateException("Unexpected root tablet logical time "
+ rtime);
+ }
+
+ return new MetadataTime(rtime, TimeType.LOGICAL);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ static Map<String,DataFileValue> cleanupRootTabletFiles(VolumeManager fs,
String dir) {
+
+ try {
+ FileStatus[] files = fs.listStatus(new Path(dir));
+
+ Map<String,DataFileValue> goodFiles = new HashMap<>(files.length);
+
+ for (FileStatus file : files) {
+
+ String path = file.getPath().toString();
+ if (file.getPath().toUri().getScheme() == null) {
+ // depending on the behavior of HDFS, if list status does not return
fully qualified
+ // volumes
+ // then could switch to the default volume
+ throw new IllegalArgumentException("Require fully qualified paths "
+ file.getPath());
+ }
+
+ String filename = file.getPath().getName();
+
+ // check for incomplete major compaction, this should only occur
+ // for root tablet
+ if (filename.startsWith("delete+")) {
+ String expectedCompactedFile =
+ path.substring(0, path.lastIndexOf("/delete+")) + "/" +
filename.split("\\+")[1];
+ if (fs.exists(new Path(expectedCompactedFile))) {
+ // compaction finished, but did not finish deleting compacted
files.. so delete it
+ if (!fs.deleteRecursively(file.getPath()))
+ log.warn("Delete of file: {} return false", file.getPath());
+ continue;
+ }
+ // compaction did not finish, so put files back
+
+ // reset path and filename for rest of loop
+ filename = filename.split("\\+", 3)[2];
+ path = path.substring(0, path.lastIndexOf("/delete+")) + "/" +
filename;
+ Path src = file.getPath();
+ Path dst = new Path(path);
+
+ if (!fs.rename(src, dst)) {
+ throw new IOException("Rename " + src + " to " + dst + " returned
false ");
+ }
+ }
+
+ if (filename.endsWith("_tmp")) {
+ log.warn("cleaning up old tmp file: {}", path);
+ if (!fs.deleteRecursively(file.getPath()))
+ log.warn("Delete of tmp file: {} return false", file.getPath());
+
+ continue;
+ }
+
+ if (!filename.startsWith(Constants.MAPFILE_EXTENSION + "_")
+ &&
!FileOperations.getValidExtensions().contains(filename.split("\\.")[1])) {
+ log.error("unknown file in tablet: {}", path);
+ continue;
+ }
+
+ goodFiles.put(path, new DataFileValue(file.getLen(), 0));
+ }
+
+ return goodFiles;
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
}
diff --git
a/server/master/src/test/java/org/apache/accumulo/master/state/RootTabletStateStoreTest.java
b/server/master/src/test/java/org/apache/accumulo/master/state/RootTabletStateStoreTest.java
index 4e5ca47..5743676 100644
---
a/server/master/src/test/java/org/apache/accumulo/master/state/RootTabletStateStoreTest.java
+++
b/server/master/src/test/java/org/apache/accumulo/master/state/RootTabletStateStoreTest.java
@@ -49,7 +49,8 @@ public class RootTabletStateStoreTest {
private static class TestAmple implements Ample {
private String json =
- new String(RootTabletMetadata.getInitialJson("/some/dir"),
StandardCharsets.UTF_8);
+ new String(RootTabletMetadata.getInitialJson("/some/dir",
"/some/dir/0000.rf"),
+ StandardCharsets.UTF_8);
@Override
public TabletMetadata readTablet(KeyExtent extent, ColumnType...
colsToFetch) {
diff --git
a/server/tserver/src/test/java/org/apache/accumulo/tserver/tablet/RootFilesTest.java
b/server/master/src/test/java/org/apache/accumulo/master/upgrade/RootFilesUpgradeTest.java
similarity index 74%
rename from
server/tserver/src/test/java/org/apache/accumulo/tserver/tablet/RootFilesTest.java
rename to
server/master/src/test/java/org/apache/accumulo/master/upgrade/RootFilesUpgradeTest.java
index 3730306..686a12e 100644
---
a/server/tserver/src/test/java/org/apache/accumulo/tserver/tablet/RootFilesTest.java
+++
b/server/master/src/test/java/org/apache/accumulo/master/upgrade/RootFilesUpgradeTest.java
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.accumulo.tserver.tablet;
+package org.apache.accumulo.master.upgrade;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -42,12 +42,18 @@ import org.junit.rules.TemporaryFolder;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
@SuppressFBWarnings(value = "PATH_TRAVERSAL_IN", justification = "paths not
set by user input")
-public class RootFilesTest {
+public class RootFilesUpgradeTest {
@Rule
public TemporaryFolder tempFolder =
new TemporaryFolder(new File(System.getProperty("user.dir") +
"/target"));
+ static void rename(VolumeManager fs, Path src, Path dst) throws IOException {
+ if (!fs.rename(src, dst)) {
+ throw new IOException("Rename " + src + " to " + dst + " returned false
");
+ }
+ }
+
private class TestWrapper {
File rootTabletDir;
Set<FileRef> oldDatafiles;
@@ -57,6 +63,35 @@ public class RootFilesTest {
VolumeManager vm;
AccumuloConfiguration conf;
+ public void prepareReplacement(VolumeManager fs, Path location,
Set<FileRef> oldDatafiles,
+ String compactName) throws IOException {
+ for (FileRef ref : oldDatafiles) {
+ Path path = ref.path();
+ rename(fs, path, new Path(location + "/delete+" + compactName + "+" +
path.getName()));
+ }
+ }
+
+ public void renameReplacement(VolumeManager fs, FileRef tmpDatafile,
FileRef newDatafile)
+ throws IOException {
+ if (fs.exists(newDatafile.path())) {
+ throw new IllegalStateException("Target map file already exist " +
newDatafile);
+ }
+
+ rename(fs, tmpDatafile.path(), newDatafile.path());
+ }
+
+ public void finishReplacement(AccumuloConfiguration acuTableConf,
VolumeManager fs,
+ Path location, Set<FileRef> oldDatafiles, String compactName) throws
IOException {
+ // start deleting files, if we do not finish they will be cleaned
+ // up later
+ for (FileRef ref : oldDatafiles) {
+ Path path = ref.path();
+ Path deleteFile = new Path(location + "/delete+" + compactName + "+" +
path.getName());
+ if (acuTableConf.getBoolean(Property.GC_TRASH_IGNORE) ||
!fs.moveToTrash(deleteFile))
+ fs.deleteRecursively(deleteFile);
+ }
+ }
+
TestWrapper(VolumeManager vm, AccumuloConfiguration conf, String
compactName,
String... inputFiles) throws IOException {
this.vm = vm;
@@ -81,21 +116,20 @@ public class RootFilesTest {
}
void prepareReplacement() throws IOException {
- RootFiles.prepareReplacement(vm, new Path(rootTabletDir.toURI()),
oldDatafiles, compactName);
+ prepareReplacement(vm, new Path(rootTabletDir.toURI()), oldDatafiles,
compactName);
}
void renameReplacement() throws IOException {
- RootFiles.renameReplacement(vm, tmpDatafile, newDatafile);
+ renameReplacement(vm, tmpDatafile, newDatafile);
}
public void finishReplacement() throws IOException {
- RootFiles.finishReplacement(conf, vm, new Path(rootTabletDir.toURI()),
oldDatafiles,
- compactName);
+ finishReplacement(conf, vm, new Path(rootTabletDir.toURI()),
oldDatafiles, compactName);
}
public Collection<String> cleanupReplacement(String... expectedFiles)
throws IOException {
Collection<String> ret =
- RootFiles.cleanupReplacement(vm, vm.listStatus(new
Path(rootTabletDir.toURI())), true);
+ Upgrader9to10.cleanupRootTabletFiles(vm,
rootTabletDir.toString()).keySet();
HashSet<String> expected = new HashSet<>();
for (String efile : expectedFiles)
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 c76ca45..0d49dd4 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
@@ -2486,12 +2486,7 @@ public class TabletServer extends AbstractServer {
TabletResourceManager trm =
resourceManager.createTabletResourceManager(extent,
getTableConfiguration(extent));
- TabletData data;
- if (extent.isRootTablet()) {
- data = new TabletData(getContext(), fs,
getTableConfiguration(extent), tabletMetadata);
- } else {
- data = new TabletData(extent, fs, tabletMetadata);
- }
+ TabletData data = new TabletData(extent, fs, tabletMetadata);
tablet = new Tablet(TabletServer.this, extent, trm, data);
// If a minor compaction starts after a tablet opens, this indicates a
log recovery
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/DatafileManager.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/DatafileManager.java
index 058817d..1ab4c20 100644
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/DatafileManager.java
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/DatafileManager.java
@@ -38,7 +38,6 @@ import org.apache.accumulo.core.metadata.schema.DataFileValue;
import org.apache.accumulo.core.replication.ReplicationConfigurationUtil;
import org.apache.accumulo.core.util.MapCounter;
import org.apache.accumulo.core.util.Pair;
-import org.apache.accumulo.fate.zookeeper.IZooReaderWriter;
import org.apache.accumulo.server.ServerConstants;
import org.apache.accumulo.server.fs.FileRef;
import org.apache.accumulo.server.fs.VolumeManager;
@@ -342,17 +341,6 @@ class DatafileManager {
void bringMinorCompactionOnline(FileRef tmpDatafile, FileRef newDatafile,
FileRef absMergeFile,
DataFileValue dfv, CommitSession commitSession, long flushId) {
- IZooReaderWriter zoo = tablet.getContext().getZooReaderWriter();
- if (tablet.getExtent().isRootTablet()) {
- try {
- if (!zoo.isLockHeld(tablet.getTabletServer().getLock().getLockID())) {
- throw new IllegalStateException();
- }
- } catch (Exception e) {
- throw new IllegalStateException("Can not bring major compaction
online, lock not held", e);
- }
- }
-
// rename before putting in metadata table, so files in metadata table
should
// always exist
do {
@@ -521,20 +509,17 @@ class DatafileManager {
final KeyExtent extent = tablet.getExtent();
long t1, t2;
- if (!extent.isRootTablet()) {
-
- if (tablet.getTabletServer().getFileSystem().exists(newDatafile.path()))
{
- log.error("Target map file already exist " + newDatafile, new
Exception());
- throw new IllegalStateException("Target map file already exist " +
newDatafile);
- }
+ if (tablet.getTabletServer().getFileSystem().exists(newDatafile.path())) {
+ log.error("Target map file already exist " + newDatafile, new
Exception());
+ throw new IllegalStateException("Target map file already exist " +
newDatafile);
+ }
- // rename before putting in metadata table, so files in metadata table
should
- // always exist
- rename(tablet.getTabletServer().getFileSystem(), tmpDatafile.path(),
newDatafile.path());
+ // rename before putting in metadata table, so files in metadata table
should
+ // always exist
+ rename(tablet.getTabletServer().getFileSystem(), tmpDatafile.path(),
newDatafile.path());
- if (dfv.getNumEntries() == 0) {
-
tablet.getTabletServer().getFileSystem().deleteRecursively(newDatafile.path());
- }
+ if (dfv.getNumEntries() == 0) {
+
tablet.getTabletServer().getFileSystem().deleteRecursively(newDatafile.path());
}
TServerInstance lastLocation = null;
@@ -542,33 +527,8 @@ class DatafileManager {
t1 = System.currentTimeMillis();
- IZooReaderWriter zoo = tablet.getContext().getZooReaderWriter();
-
tablet.incrementDataSourceDeletions();
- if (extent.isRootTablet()) {
-
- waitForScansToFinish(oldDatafiles, true, Long.MAX_VALUE);
-
- try {
- if (!zoo.isLockHeld(tablet.getTabletServer().getLock().getLockID()))
{
- throw new IllegalStateException();
- }
- } catch (Exception e) {
- throw new IllegalStateException("Can not bring major compaction
online, lock not held",
- e);
- }
-
- // mark files as ready for deletion, but
- // do not delete them until we successfully
- // rename the compacted map file, in case
- // the system goes down
-
- RootFiles.replaceFiles(tablet.getTableConfiguration(),
- tablet.getTabletServer().getFileSystem(), tablet.getLocation(),
oldDatafiles,
- tmpDatafile, newDatafile);
- }
-
// atomically remove old files and add new file
for (FileRef oldDatafile : oldDatafiles) {
if (!datafileSizes.containsKey(oldDatafile)) {
@@ -597,16 +557,14 @@ class DatafileManager {
t2 = System.currentTimeMillis();
}
- if (!extent.isRootTablet()) {
- Set<FileRef> filesInUseByScans = waitForScansToFinish(oldDatafiles,
false, 10000);
- if (filesInUseByScans.size() > 0)
- log.debug("Adding scan refs to metadata {} {}", extent,
filesInUseByScans);
- MasterMetadataUtil.replaceDatafiles(tablet.getContext(), extent,
oldDatafiles,
- filesInUseByScans, newDatafile, compactionId, dfv,
- tablet.getTabletServer().getClientAddressString(), lastLocation,
- tablet.getTabletServer().getLock());
- removeFilesAfterScan(filesInUseByScans);
- }
+ Set<FileRef> filesInUseByScans = waitForScansToFinish(oldDatafiles, false,
10000);
+ if (filesInUseByScans.size() > 0)
+ log.debug("Adding scan refs to metadata {} {}", extent,
filesInUseByScans);
+ MasterMetadataUtil.replaceDatafiles(tablet.getContext(), extent,
oldDatafiles,
+ filesInUseByScans, newDatafile, compactionId, dfv,
+ tablet.getTabletServer().getClientAddressString(), lastLocation,
+ tablet.getTabletServer().getLock());
+ removeFilesAfterScan(filesInUseByScans);
log.debug(String.format("MajC finish lock %.2f secs", (t2 - t1) / 1000.0));
log.debug("TABLET_HIST {} MajC --> {}", oldDatafiles, newDatafile);
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/RootFiles.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/RootFiles.java
deleted file mode 100644
index 367ce21..0000000
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/RootFiles.java
+++ /dev/null
@@ -1,139 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.accumulo.tserver.tablet;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Set;
-
-import org.apache.accumulo.core.Constants;
-import org.apache.accumulo.core.conf.AccumuloConfiguration;
-import org.apache.accumulo.core.conf.Property;
-import org.apache.accumulo.core.file.FileOperations;
-import org.apache.accumulo.server.fs.FileRef;
-import org.apache.accumulo.server.fs.VolumeManager;
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.Path;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class RootFiles {
-
- private static final Logger log = LoggerFactory.getLogger(RootFiles.class);
-
- public static void prepareReplacement(VolumeManager fs, Path location,
Set<FileRef> oldDatafiles,
- String compactName) throws IOException {
- for (FileRef ref : oldDatafiles) {
- Path path = ref.path();
- DatafileManager.rename(fs, path,
- new Path(location + "/delete+" + compactName + "+" +
path.getName()));
- }
- }
-
- public static void renameReplacement(VolumeManager fs, FileRef tmpDatafile,
FileRef newDatafile)
- throws IOException {
- if (fs.exists(newDatafile.path())) {
- log.error("Target map file already exist " + newDatafile, new
Exception());
- throw new IllegalStateException("Target map file already exist " +
newDatafile);
- }
-
- DatafileManager.rename(fs, tmpDatafile.path(), newDatafile.path());
- }
-
- public static void finishReplacement(AccumuloConfiguration acuTableConf,
VolumeManager fs,
- Path location, Set<FileRef> oldDatafiles, String compactName) throws
IOException {
- // start deleting files, if we do not finish they will be cleaned
- // up later
- for (FileRef ref : oldDatafiles) {
- Path path = ref.path();
- Path deleteFile = new Path(location + "/delete+" + compactName + "+" +
path.getName());
- if (acuTableConf.getBoolean(Property.GC_TRASH_IGNORE) ||
!fs.moveToTrash(deleteFile))
- fs.deleteRecursively(deleteFile);
- }
- }
-
- public static void replaceFiles(AccumuloConfiguration acuTableConf,
VolumeManager fs,
- Path location, Set<FileRef> oldDatafiles, FileRef tmpDatafile, FileRef
newDatafile)
- throws IOException {
- String compactName = newDatafile.path().getName();
-
- prepareReplacement(fs, location, oldDatafiles, compactName);
- renameReplacement(fs, tmpDatafile, newDatafile);
- finishReplacement(acuTableConf, fs, location, oldDatafiles, compactName);
- }
-
- public static Collection<String> cleanupReplacement(VolumeManager fs,
FileStatus[] files,
- boolean deleteTmp) throws IOException {
- /*
- * called in constructor and before major compactions
- */
- Collection<String> goodFiles = new ArrayList<>(files.length);
-
- for (FileStatus file : files) {
-
- String path = file.getPath().toString();
- if (file.getPath().toUri().getScheme() == null) {
- // depending on the behavior of HDFS, if list status does not return
fully qualified volumes
- // then could switch to the default volume
- throw new IllegalArgumentException("Require fully qualified paths " +
file.getPath());
- }
-
- String filename = file.getPath().getName();
-
- // check for incomplete major compaction, this should only occur
- // for root tablet
- if (filename.startsWith("delete+")) {
- String expectedCompactedFile =
- path.substring(0, path.lastIndexOf("/delete+")) + "/" +
filename.split("\\+")[1];
- if (fs.exists(new Path(expectedCompactedFile))) {
- // compaction finished, but did not finish deleting compacted
files.. so delete it
- if (!fs.deleteRecursively(file.getPath()))
- log.warn("Delete of file: {} return false", file.getPath());
- continue;
- }
- // compaction did not finish, so put files back
-
- // reset path and filename for rest of loop
- filename = filename.split("\\+", 3)[2];
- path = path.substring(0, path.lastIndexOf("/delete+")) + "/" +
filename;
-
- DatafileManager.rename(fs, file.getPath(), new Path(path));
- }
-
- if (filename.endsWith("_tmp")) {
- if (deleteTmp) {
- log.warn("cleaning up old tmp file: {}", path);
- if (!fs.deleteRecursively(file.getPath()))
- log.warn("Delete of tmp file: {} return false", file.getPath());
-
- }
- continue;
- }
-
- if (!filename.startsWith(Constants.MAPFILE_EXTENSION + "_")
- &&
!FileOperations.getValidExtensions().contains(filename.split("\\.")[1])) {
- log.error("unknown file in tablet: {}", path);
- continue;
- }
-
- goodFiles.add(path);
- }
-
- return goodFiles;
- }
-}
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
index 24c2780..bbc70a4 100644
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
@@ -77,7 +77,6 @@ import
org.apache.accumulo.core.iterators.system.SourceSwitchingIterator;
import org.apache.accumulo.core.master.thrift.BulkImportState;
import org.apache.accumulo.core.master.thrift.TabletLoadState;
import org.apache.accumulo.core.metadata.MetadataTable;
-import org.apache.accumulo.core.metadata.RootTable;
import org.apache.accumulo.core.metadata.schema.DataFileValue;
import org.apache.accumulo.core.metadata.schema.MetadataTime;
import org.apache.accumulo.core.protobuf.ProtobufUtil;
@@ -1427,25 +1426,12 @@ public class Tablet {
throw new RuntimeException(msg);
}
- if (extent.isRootTablet()) {
- if (!fileLog.getSecond().keySet()
- .equals(getDatafileManager().getDatafileSizes().keySet())) {
- String msg = "Data file in " + RootTable.NAME + " differ from in
memory data " + extent
- + " " + fileLog.getSecond().keySet() + " "
- + getDatafileManager().getDatafileSizes().keySet();
- log.error(msg);
- throw new RuntimeException(msg);
- }
- } else {
- if
(!fileLog.getSecond().equals(getDatafileManager().getDatafileSizes())) {
- String msg =
- "Data file in " + MetadataTable.NAME + " differ from in memory
data " + extent + " "
- + fileLog.getSecond() + " " +
getDatafileManager().getDatafileSizes();
- log.error(msg);
- throw new RuntimeException(msg);
- }
+ if
(!fileLog.getSecond().equals(getDatafileManager().getDatafileSizes())) {
+ String msg = "Data files in differ from in memory data " + extent + "
"
+ + fileLog.getSecond() + " " +
getDatafileManager().getDatafileSizes();
+ log.error(msg);
+ throw new RuntimeException(msg);
}
-
} catch (Exception e) {
String msg = "Failed to do close consistency check for tablet " + extent;
log.error(msg, e);
@@ -1797,14 +1783,6 @@ public class Tablet {
majorCompactionState = CompactionState.IN_PROGRESS;
notifyAll();
- VolumeManager fs = getTabletServer().getFileSystem();
- if (extent.isRootTablet()) {
- // very important that we call this before doing major compaction,
- // otherwise deleted compacted files could possible be brought back
- // at some point if the file they were compacted to was legitimately
- // removed by a major compaction
- RootFiles.cleanupReplacement(fs, fs.listStatus(this.location), false);
- }
SortedMap<FileRef,DataFileValue> allFiles =
getDatafileManager().getDatafileSizes();
List<FileRef> inputFiles = new ArrayList<>();
if (reason == MajorCompactionReason.CHOP) {
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/TabletData.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/TabletData.java
index 53f5ebc..4f501a7 100644
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/TabletData.java
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/TabletData.java
@@ -16,9 +16,7 @@
*/
package org.apache.accumulo.tserver.tablet;
-import java.io.IOException;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -26,26 +24,14 @@ import java.util.Map;
import java.util.SortedMap;
import java.util.TreeMap;
-import org.apache.accumulo.core.client.admin.TimeType;
-import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.dataImpl.KeyExtent;
-import org.apache.accumulo.core.file.FileOperations;
-import org.apache.accumulo.core.file.FileSKVIterator;
-import org.apache.accumulo.core.metadata.RootTable;
import org.apache.accumulo.core.metadata.schema.DataFileValue;
import org.apache.accumulo.core.metadata.schema.MetadataTime;
import org.apache.accumulo.core.metadata.schema.TabletMetadata;
import org.apache.accumulo.core.tabletserver.log.LogEntry;
-import org.apache.accumulo.server.ServerContext;
import org.apache.accumulo.server.fs.FileRef;
import org.apache.accumulo.server.fs.VolumeManager;
-import org.apache.accumulo.server.fs.VolumeUtil;
import org.apache.accumulo.server.master.state.TServerInstance;
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-
-import com.google.common.base.Preconditions;
/*
* Basic information needed to create a tablet.
@@ -85,45 +71,6 @@ public class TabletData {
});
}
- // Read basic root table metadata from zookeeper
- public TabletData(ServerContext context, VolumeManager fs,
AccumuloConfiguration conf,
- TabletMetadata rootMeta) throws IOException {
- Preconditions.checkArgument(rootMeta.getExtent().equals(RootTable.EXTENT));
-
- directory = VolumeUtil.switchRootTableVolume(context, rootMeta.getDir());
-
- Path location = new Path(directory);
-
- // cleanReplacement() has special handling for deleting files
- FileStatus[] files = fs.listStatus(location);
- Collection<String> goodPaths = RootFiles.cleanupReplacement(fs, files,
true);
- long rtime = Long.MIN_VALUE;
- for (String good : goodPaths) {
- Path path = new Path(good);
- String filename = path.getName();
- FileRef ref = new FileRef(location + "/" + filename, path);
- DataFileValue dfv = new DataFileValue(0, 0);
- dataFiles.put(ref, dfv);
-
- FileSystem ns = fs.getVolumeByPath(path).getFileSystem();
- long maxTime = -1;
- try (FileSKVIterator reader =
FileOperations.getInstance().newReaderBuilder()
- .forFile(path.toString(), ns, ns.getConf(),
context.getCryptoService())
- .withTableConfiguration(conf).seekToBeginning().build()) {
- while (reader.hasTop()) {
- maxTime = Math.max(maxTime, reader.getTopKey().getTimestamp());
- reader.next();
- }
- }
- if (maxTime > rtime) {
- time = new MetadataTime(maxTime, TimeType.LOGICAL);
- rtime = maxTime;
- }
- }
-
- this.logEntries.addAll(rootMeta.getLogs());
- }
-
// Data pulled from an existing tablet to make a split
public TabletData(String tabletDirectory, SortedMap<FileRef,DataFileValue>
highDatafileSizes,
MetadataTime time, long lastFlushID, long lastCompactID, TServerInstance
lastLocation,
