This is an automated email from the ASF dual-hosted git repository.
dlmarion pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/main by this push:
new 1f2b0dc54d Update existing code that expects relative file references
(#6616)
1f2b0dc54d is described below
commit 1f2b0dc54d8b96b10c2c34ccaab3bf001b1ead7c
Author: Dave Marion <[email protected]>
AuthorDate: Tue Nov 10 11:09:19 2026 -0500
Update existing code that expects relative file references (#6616)
This change modifies the remaining code that allows for relative
RFile and WAL references. The RFiles and WALs that are created
today are done with absolute paths, but some of the existing
code does not enforce or check that.
---
.../accumulo/core/conf/AccumuloConfiguration.java | 9 +-
.../accumulo/core/tabletserver/log/LogEntry.java | 44 ++++++--
.../core/metadata/schema/TabletMetadataTest.java | 18 ++--
.../core/tabletserver/log/LogEntryTest.java | 48 +++++++--
.../org/apache/accumulo/server/fs/VolumeUtil.java | 4 +-
.../server/util/adminCommand/ListVolumesUsed.java | 6 +-
.../apache/accumulo/server/fs/VolumeUtilTest.java | 9 +-
.../server/manager/state/TabletManagementTest.java | 4 +-
.../accumulo/gc/GarbageCollectionAlgorithm.java | 53 +++++----
.../apache/accumulo/gc/GarbageCollectionTest.java | 106 +++++++++---------
.../accumulo/manager/recovery/RecoveryManager.java | 4 +-
.../accumulo/manager/upgrade/Upgrader11to12.java | 85 ++++++++++++++-
.../manager/tableOps/merge/MergeTabletsTest.java | 2 +-
.../manager/tableOps/split/UpdateTabletsTest.java | 2 +-
.../manager/upgrade/Upgrader11to12Test.java | 120 +++++++++++++++++++--
.../org/apache/accumulo/tserver/InMemoryMap.java | 2 +-
.../accumulo/tserver/TabletClientHandler.java | 3 +-
.../org/apache/accumulo/tserver/log/DfsLogger.java | 37 +++----
.../accumulo/tserver/log/ResolvedSortedLog.java | 2 +-
.../accumulo/tserver/log/TabletServerLogger.java | 2 +-
.../apache/accumulo/tserver/logger/LogReader.java | 3 +-
.../accumulo/tserver/WalRemovalOrderTest.java | 3 +-
.../shell/commands/ListTabletsCommandTest.java | 4 +-
.../test/MissingWalHeaderCompletesRecoveryIT.java | 40 +++----
.../java/org/apache/accumulo/test/RecoveryIT.java | 4 +-
.../test/fate/ManagerRepoIT_SimpleSuite.java | 9 +-
.../test/functional/AmpleConditionalWriterIT.java | 37 +++----
.../functional/TabletManagementIteratorIT.java | 6 +-
28 files changed, 456 insertions(+), 210 deletions(-)
diff --git
a/core/src/main/java/org/apache/accumulo/core/conf/AccumuloConfiguration.java
b/core/src/main/java/org/apache/accumulo/core/conf/AccumuloConfiguration.java
index b8bb4f0fe3..b105c3ebc6 100644
---
a/core/src/main/java/org/apache/accumulo/core/conf/AccumuloConfiguration.java
+++
b/core/src/main/java/org/apache/accumulo/core/conf/AccumuloConfiguration.java
@@ -385,14 +385,19 @@ public abstract class AccumuloConfiguration implements
Iterable<Entry<String,Str
}
/**
- * Gets a property of type {@link PropertyType#PATH}.
+ * Gets a property of type {@link PropertyType#PATH} or {@link
PropertyType#ABSOLUTEPATH}.
*
* @param property property to get
* @return property value
* @throws IllegalArgumentException if the property is of the wrong type
*/
public String getPath(Property property) {
- checkType(property, PropertyType.PATH);
+ if (property.getType() != PropertyType.PATH
+ && property.getType() != PropertyType.ABSOLUTEPATH) {
+ throw new IllegalArgumentException("Configuration method intended for
type "
+ + PropertyType.PATH + " or " + PropertyType.ABSOLUTEPATH + " called
with a "
+ + property.getType() + " argument (" + property.getKey() + ")");
+ }
String pathString = get(property);
if (pathString == null) {
diff --git
a/core/src/main/java/org/apache/accumulo/core/tabletserver/log/LogEntry.java
b/core/src/main/java/org/apache/accumulo/core/tabletserver/log/LogEntry.java
index 57820065b0..63cdcb0530 100644
--- a/core/src/main/java/org/apache/accumulo/core/tabletserver/log/LogEntry.java
+++ b/core/src/main/java/org/apache/accumulo/core/tabletserver/log/LogEntry.java
@@ -22,10 +22,12 @@ import java.util.Map.Entry;
import java.util.Objects;
import java.util.UUID;
+import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Value;
import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.LogColumnFamily;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.checkerframework.checker.nullness.qual.NonNull;
@@ -35,22 +37,37 @@ import com.google.common.net.HostAndPort;
public final class LogEntry {
- private final String path;
+ private static final String LOG_PATH =
+ "%s" + Path.SEPARATOR + Constants.WAL_DIR + Path.SEPARATOR + "%s" +
Path.SEPARATOR + "%s";
+
+ private final Path path;
private final HostAndPort tserver;
private final UUID uniqueId;
private final Text columnQualifier;
- private LogEntry(String path, HostAndPort tserver, UUID uniqueId, Text
columnQualifier) {
+ private LogEntry(Path path, HostAndPort tserver, UUID uniqueId, Text
columnQualifier) {
this.path = path;
this.tserver = tserver;
this.uniqueId = uniqueId;
this.columnQualifier = columnQualifier;
}
+ public static LogEntry withSwitchedVolume(LogEntry entry, Path newPath) {
+ return new LogEntry(newPath, entry.getTServer(), entry.getUniqueID(),
+ entry.getColumnQualifier());
+ }
+
+ public static LogEntry forServer(String volume, HostAndPort address) {
+ UUID filename = UUID.randomUUID();
+ String addressForFilename = address.toString().replace(':', '+');
+ String filePath = LOG_PATH.formatted(volume, addressForFilename, filename);
+ return new LogEntry(validatePath(new Path(filePath)), address, filename,
null);
+ }
+
/**
- * Creates a new LogEntry object after validating the expected format of the
path. We expect the
- * path to contain a tserver (host+port) followed by a UUID as the file name
as the last two
- * components.<br>
+ * Creates a new LogEntry object after validating the expected format of the
path. The path must
+ * be an absolute filesystem URI and contain a tserver (host+port) followed
by a UUID as the file
+ * name as the last two components.<br>
* For example,
file:///some/dir/path/localhost+1234/927ba659-d109-4bce-b0a5-bcbbcb9942a2 is a
* valid path.
*
@@ -96,7 +113,16 @@ public final class LogEntry {
throw new IllegalArgumentException(badUuidMsg);
}
- return new LogEntry(path, tserver, uuid, columnQualifier);
+ Path logPath = validatePath(new Path(path));
+ return new LogEntry(logPath, tserver, uuid, columnQualifier);
+ }
+
+ private static Path validatePath(Path path) {
+ if (path.toUri().getScheme() == null || !path.isAbsolute()) {
+ throw new IllegalArgumentException(
+ "Invalid path format. The path must be an absolute URI: " + path);
+ }
+ return path;
}
/**
@@ -120,12 +146,12 @@ public final class LogEntry {
@NonNull
@VisibleForTesting
- HostAndPort getTServer() {
+ public HostAndPort getTServer() {
return tserver;
}
@NonNull
- public String getPath() {
+ public Path getPath() {
return path;
}
@@ -136,7 +162,7 @@ public final class LogEntry {
@Override
public String toString() {
- return path;
+ return path.toString();
}
@Override
diff --git
a/core/src/test/java/org/apache/accumulo/core/metadata/schema/TabletMetadataTest.java
b/core/src/test/java/org/apache/accumulo/core/metadata/schema/TabletMetadataTest.java
index b56315d8d7..7e8c7252ce 100644
---
a/core/src/test/java/org/apache/accumulo/core/metadata/schema/TabletMetadataTest.java
+++
b/core/src/test/java/org/apache/accumulo/core/metadata/schema/TabletMetadataTest.java
@@ -171,9 +171,9 @@ public class TabletMetadataTest {
mutation.at().family(LastLocationColumnFamily.NAME).qualifier("s000").put("server2:8555");
- LogEntry le1 = LogEntry.fromPath("localhost+8020/" + UUID.randomUUID());
+ LogEntry le1 = LogEntry.fromPath("file:/accumulo/wal/localhost+8020/" +
UUID.randomUUID());
le1.addToMutation(mutation);
- LogEntry le2 = LogEntry.fromPath("localhost+8020/" + UUID.randomUUID());
+ LogEntry le2 = LogEntry.fromPath("file:/accumulo/wal/localhost+8020/" +
UUID.randomUUID());
le2.addToMutation(mutation);
StoredTabletFile sf1 = StoredTabletFile.of(new
Path("hdfs://nn1/acc/tables/1/t-0001/sf1.rf"));
@@ -515,8 +515,8 @@ public class TabletMetadataTest {
assertThrows(UnsupportedOperationException.class,
() -> tm.getFilesMap().put(stf, new DataFileValue(0, 0, 0)));
assertTrue(tm.getLogs().isEmpty());
- assertThrows(UnsupportedOperationException.class,
- () -> tm.getLogs().add(LogEntry.fromPath("localhost+8020/" +
UUID.randomUUID())));
+ assertThrows(UnsupportedOperationException.class, () -> tm.getLogs()
+ .add(LogEntry.fromPath("file:/accumulo/wal/localhost+8020/" +
UUID.randomUUID())));
assertTrue(tm.getScans().isEmpty());
assertThrows(UnsupportedOperationException.class, () ->
tm.getScans().add(stf));
assertTrue(tm.getLoaded().isEmpty());
@@ -534,7 +534,7 @@ public class TabletMetadataTest {
b.table(TableId.of("4"));
b.extCompaction(ecid, ecMeta);
b.file(stf, new DataFileValue(0, 0, 0));
- b.log(LogEntry.fromPath("localhost+8020/" + UUID.randomUUID()));
+ b.log(LogEntry.fromPath("file:/accumulo/wal/localhost+8020/" +
UUID.randomUUID()));
b.scan(stf);
b.loadedFile(stf, FateId.from(FateInstanceType.USER, UUID.randomUUID()));
b.compacted(FateId.from(FateInstanceType.USER, UUID.randomUUID()));
@@ -552,8 +552,8 @@ public class TabletMetadataTest {
assertThrows(UnsupportedOperationException.class,
() -> tm2.getFilesMap().put(stf, new DataFileValue(0, 0, 0)));
assertEquals(1, tm2.getLogs().size());
- assertThrows(UnsupportedOperationException.class,
- () -> tm2.getLogs().add(LogEntry.fromPath("localhost+8020/" +
UUID.randomUUID())));
+ assertThrows(UnsupportedOperationException.class, () -> tm2.getLogs()
+ .add(LogEntry.fromPath("file:/accumulo/wal/localhost+8020/" +
UUID.randomUUID())));
assertEquals(1, tm2.getScans().size());
assertThrows(UnsupportedOperationException.class, () ->
tm2.getScans().add(stf));
assertEquals(1, tm2.getLoaded().size());
@@ -829,8 +829,8 @@ public class TabletMetadataTest {
new CompactionMetadata(Set.of(sf1, sf2), rf1, "cid1",
CompactionKind.USER, (short) 3,
ResourceGroupId.of("Q1"), true, FateId.from(type,
UUID.randomUUID()));
- LogEntry le1 = LogEntry.fromPath("localhost+8020/" + UUID.randomUUID());
- LogEntry le2 = LogEntry.fromPath("localhost+8020/" + UUID.randomUUID());
+ LogEntry le1 = LogEntry.fromPath("file:/accumulo/wal/localhost+8020/" +
UUID.randomUUID());
+ LogEntry le2 = LogEntry.fromPath("file:/accumulo/wal/localhost+8020/" +
UUID.randomUUID());
FateId selFilesFateId = FateId.from(type, UUID.randomUUID());
SelectedFiles selFiles = new SelectedFiles(Set.of(sf1, sf4), false,
selFilesFateId,
diff --git
a/core/src/test/java/org/apache/accumulo/core/tabletserver/log/LogEntryTest.java
b/core/src/test/java/org/apache/accumulo/core/tabletserver/log/LogEntryTest.java
index 579571d81d..f9c994464f 100644
---
a/core/src/test/java/org/apache/accumulo/core/tabletserver/log/LogEntryTest.java
+++
b/core/src/test/java/org/apache/accumulo/core/tabletserver/log/LogEntryTest.java
@@ -29,6 +29,7 @@ import java.util.UUID;
import java.util.stream.Stream;
import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Value;
import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.LogColumnFamily;
import org.apache.hadoop.io.Text;
import org.junit.jupiter.api.Test;
@@ -60,10 +61,33 @@ public class LogEntryTest {
verifyLogEntry(logEntry, new Text(columnQualifier));
}
+ @Test
+ public void testForServer() {
+ var logEntry = LogEntry.forServer("viewfs:/a/accumulo",
+ HostAndPort.fromString(validHost.replace('+', ':')));
+ assertTrue(logEntry.getPath().toUri().getScheme() != null);
+
+ var roundTripped = LogEntry.fromMetaWalEntry(new SimpleImmutableEntry<>(
+ new Key("1<", LogColumnFamily.STR_NAME,
logEntry.getColumnQualifier().toString()), null));
+ assertEquals(logEntry, roundTripped);
+ assertEquals(logEntry.getPath(), roundTripped.getPath());
+ assertEquals(logEntry.getUniqueID(), roundTripped.getUniqueID());
+ assertEquals(logEntry.getTServer(), roundTripped.getTServer());
+ assertEquals(HostAndPort.fromString(validHost.replace('+', ':')),
logEntry.getTServer());
+ assertEquals("viewfs:/a/accumulo/wal/" + validHost + "/" +
roundTripped.getUniqueID(),
+ logEntry.getPath().toString());
+ }
+
+ @Test
+ public void testForServerRequiresQualifiedVolume() {
+ assertThrows(IllegalArgumentException.class,
+ () -> LogEntry.forServer("/accumulo",
HostAndPort.fromString("localhost:9997")));
+ }
+
// helper for testing build from constructor or from metadata
private void verifyLogEntry(LogEntry logEntry, Text expectedColumnQualifier)
{
assertEquals(validPath, logEntry.toString());
- assertEquals(validPath, logEntry.getPath());
+ assertEquals(validPath, logEntry.getPath().toString());
assertEquals(HostAndPort.fromString(validHost.replace('+', ':')),
logEntry.getTServer());
assertEquals(expectedColumnQualifier, logEntry.getColumnQualifier());
assertEquals(validUUID, logEntry.getUniqueID());
@@ -88,12 +112,24 @@ public class LogEntryTest {
@Test
public void testValidPaths() {
- var validPath1 = validHost + "/" + validUUID;
- var validPath2 = "dir1/" + validPath1;
- var validPath3 = "dir2/" + validPath2;
+ var validPath1 = "file:///accumulo/wal/" + validHost + "/" + validUUID;
+ var validPath2 = "hdfs://nn1/accumulo/wal/" + validHost + "/" + validUUID;
- Stream.of(validPath1, validPath2, validPath3)
- .forEach(s -> assertDoesNotThrow(() -> LogEntry.fromPath(s)));
+ Stream.of(validPath1, validPath2).forEach(s -> assertDoesNotThrow(() ->
LogEntry.fromPath(s)));
+ }
+
+ @Test
+ public void testRelativePathsAreInvalid() {
+ Stream.of(validHost + "/" + validUUID, "file:accumulo/wal/" + validHost +
"/" + validUUID)
+ .forEach(s -> assertThrows(IllegalArgumentException.class, () ->
LogEntry.fromPath(s)));
+ }
+
+ @Test
+ public void testRelativeMetadataPathIsInvalid() {
+ var qualifier = "prefix/" + validHost + "/" + validUUID;
+ var entry = new SimpleImmutableEntry<Key,Value>(
+ new Key("1<", LogColumnFamily.STR_NAME, qualifier), null);
+ assertThrows(IllegalArgumentException.class, () ->
LogEntry.fromMetaWalEntry(entry));
}
@Test
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 e41237b087..d56a434a9c 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
@@ -80,8 +80,8 @@ public class VolumeUtil {
}
public static LogEntry switchVolume(LogEntry le, Map<Path,Path>
replacements) {
- Path switchedPath = switchVolume(new Path(le.getPath()), FileType.WAL,
replacements);
- return switchedPath == null ? null :
LogEntry.fromPath(switchedPath.toString());
+ Path switchedPath = switchVolume(le.getPath(), FileType.WAL, replacements);
+ return switchedPath == null ? null : LogEntry.withSwitchedVolume(le,
switchedPath);
}
public static boolean needsVolumeReplacement(final Map<Path,Path>
replacements,
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/util/adminCommand/ListVolumesUsed.java
b/server/base/src/main/java/org/apache/accumulo/server/util/adminCommand/ListVolumesUsed.java
index 90d585b674..7a13db610a 100644
---
a/server/base/src/main/java/org/apache/accumulo/server/util/adminCommand/ListVolumesUsed.java
+++
b/server/base/src/main/java/org/apache/accumulo/server/util/adminCommand/ListVolumesUsed.java
@@ -79,8 +79,8 @@ public class ListVolumesUsed extends
ServerKeywordExecutable<ServerOpts> {
return ret.toString();
}
- private String getLogURI(String logEntry) {
- Path ret = FileType.WAL.getVolume(new Path(logEntry));
+ private String getLogURI(Path logEntry) {
+ Path ret = FileType.WAL.getVolume(logEntry);
if (ret == null) {
return "RELATIVE";
}
@@ -126,7 +126,7 @@ public class ListVolumesUsed extends
ServerKeywordExecutable<ServerOpts> {
WalStateManager wals = new WalStateManager(context);
for (var wal : wals.getAllState()) {
- volumes.add(getLogURI(wal.path().toString()));
+ volumes.add(getLogURI(wal.path()));
}
for (String volume : volumes) {
System.out.println("\tVolume : " + volume);
diff --git
a/server/base/src/test/java/org/apache/accumulo/server/fs/VolumeUtilTest.java
b/server/base/src/test/java/org/apache/accumulo/server/fs/VolumeUtilTest.java
index 033cf17212..51c2b94e0f 100644
---
a/server/base/src/test/java/org/apache/accumulo/server/fs/VolumeUtilTest.java
+++
b/server/base/src/test/java/org/apache/accumulo/server/fs/VolumeUtilTest.java
@@ -172,16 +172,19 @@ public class VolumeUtilTest {
String fileName = "hdfs://nn1/accumulo/wal/localhost+9997/" + walUUID;
LogEntry le = LogEntry.fromPath(fileName);
LogEntry fixedVolume = VolumeUtil.switchVolume(le, replacements);
- assertEquals("viewfs:/a/accumulo/wal/localhost+9997/" + walUUID,
fixedVolume.getPath());
+ assertEquals("viewfs:/a/accumulo/wal/localhost+9997/" + walUUID,
+ fixedVolume.getPath().toString());
fileName = "hdfs://nn1:9000/accumulo/wal/localhost+9997/" + walUUID;
le = LogEntry.fromPath(fileName);
fixedVolume = VolumeUtil.switchVolume(le, replacements);
- assertEquals("viewfs:/a/accumulo/wal/localhost+9997/" + walUUID,
fixedVolume.getPath());
+ assertEquals("viewfs:/a/accumulo/wal/localhost+9997/" + walUUID,
+ fixedVolume.getPath().toString());
fileName = "hdfs://nn2/accumulo/wal/localhost+9997/" + walUUID;
le = LogEntry.fromPath(fileName);
fixedVolume = VolumeUtil.switchVolume(le, replacements);
- assertEquals("viewfs:/b/accumulo/wal/localhost+9997/" + walUUID,
fixedVolume.getPath());
+ assertEquals("viewfs:/b/accumulo/wal/localhost+9997/" + walUUID,
+ fixedVolume.getPath().toString());
}
}
diff --git
a/server/base/src/test/java/org/apache/accumulo/server/manager/state/TabletManagementTest.java
b/server/base/src/test/java/org/apache/accumulo/server/manager/state/TabletManagementTest.java
index 0bab62dfbc..5665548a14 100644
---
a/server/base/src/test/java/org/apache/accumulo/server/manager/state/TabletManagementTest.java
+++
b/server/base/src/test/java/org/apache/accumulo/server/manager/state/TabletManagementTest.java
@@ -108,9 +108,9 @@ public class TabletManagementTest {
mutation.at().family(LastLocationColumnFamily.NAME).qualifier("s000").put("server2:8555");
- LogEntry le1 = LogEntry.fromPath("localhost+8020/" + UUID.randomUUID());
+ LogEntry le1 = LogEntry.fromPath("file:/accumulo/wal/localhost+8020/" +
UUID.randomUUID());
le1.addToMutation(mutation);
- LogEntry le2 = LogEntry.fromPath("localhost+8020/" + UUID.randomUUID());
+ LogEntry le2 = LogEntry.fromPath("file:/accumulo/wal/localhost+8020/" +
UUID.randomUUID());
le2.addToMutation(mutation);
StoredTabletFile sf1 =
diff --git
a/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectionAlgorithm.java
b/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectionAlgorithm.java
index 462d593cf4..f0e8994dd1 100644
---
a/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectionAlgorithm.java
+++
b/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectionAlgorithm.java
@@ -42,6 +42,7 @@ import org.apache.accumulo.core.gc.Reference;
import org.apache.accumulo.core.metadata.schema.Ample.GcCandidateType;
import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily;
import org.apache.accumulo.core.trace.TraceUtil;
+import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -55,18 +56,8 @@ public class GarbageCollectionAlgorithm {
private static final Logger log =
LoggerFactory.getLogger(GarbageCollectionAlgorithm.class);
/**
- * This method takes a file or directory path and returns a relative path in
1 of 2 forms:
- *
- * <pre>
- * 1- For files: table-id/tablet-directory/filename.rf
- * 2- For directories: table-id/tablet-directory
- * </pre>
- *
- * For example, for full file path like
hdfs://foo:6000/accumulo/tables/4/t0/F000.rf it will
- * return 4/t0/F000.rf. For a directory that already is relative, like 4/t0,
it will just return
- * the original path. This method will also remove prefixed relative paths
like ../4/t0/F000.rf
- * and return 4/t0/F000.rf. It also strips out empty tokens from paths like
- * hdfs://foo.com:6000/accumulo/tables/4//t0//F001.rf returning 4/t0/F001.rf.
+ * Normalizes a candidate, directory, or blip path to a table-relative GC
key. File references use
+ * {@link #makeRelativeFileReference(String)} because they must be absolute
URIs.
*/
private String makeRelative(String path, int expectedLen) {
String relPath = path;
@@ -119,6 +110,34 @@ public class GarbageCollectionAlgorithm {
return relPath;
}
+ /**
+ * Extracts the table-relative GC key from an absolute file reference URI,
such as
+ * hdfs://foo:6000/accumulo/tables/4/t0/F000.rf -> 4/t0/F000.rf.
+ */
+ private String makeRelativeFileReference(String path) {
+ Path referencePath = new Path(path);
+ if (!referencePath.isAbsolute() || referencePath.toUri().getScheme() ==
null) {
+ throw new IllegalArgumentException("File reference must be an absolute
URI: " + path);
+ }
+
+ String[] tokens = path.split("/");
+ String[] suffix = new String[4];
+ int suffixLen = 0;
+ for (int i = tokens.length - 1; i >= 0 && suffixLen < suffix.length; i--) {
+ if (!tokens[i].isEmpty()) {
+ suffix[suffixLen++] = tokens[i];
+ }
+ }
+
+ if (suffixLen != suffix.length || !Constants.TABLE_DIR.equals(suffix[3])) {
+ throw new IllegalArgumentException("Failed to make path relative. Bad
reference: " + path);
+ }
+
+ String relativePath = suffix[2] + "/" + suffix[1] + "/" + suffix[0];
+ log.trace("{} -> {}", path, relativePath);
+ return relativePath;
+ }
+
private SortedMap<String,GcCandidate> makeRelative(Collection<GcCandidate>
candidates) {
SortedMap<String,GcCandidate> ret = new TreeMap<>();
@@ -161,15 +180,7 @@ public class GarbageCollectionAlgorithm {
}
} else {
String reference = ref.getMetadataPath();
- if (reference.startsWith("/")) {
- log.debug("Candidate {} has a relative path, prepend tableId {}",
reference,
- ref.getTableId());
- reference = "/" + ref.getTableId() + ref.getMetadataPath();
- } else if (!reference.contains(":") && !reference.startsWith("../"))
{
- throw new RuntimeException("Bad file reference " + reference);
- }
-
- String relativePath = makeRelative(reference, 3);
+ String relativePath = makeRelativeFileReference(reference);
// WARNING: This line is EXTREMELY IMPORTANT.
// You MUST REMOVE candidates that are still in use
diff --git
a/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectionTest.java
b/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectionTest.java
index 72cc61e474..8058aab514 100644
--- a/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectionTest.java
+++ b/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectionTest.java
@@ -408,24 +408,24 @@ public class GarbageCollectionTest {
}
@Test
- public void testRelative() throws Exception {
+ public void testAbsoluteReferencesWithRelativeCandidates() throws Exception {
TestGCE gce = new TestGCE();
var candOne = gce.addCandidate("/4/t0/F000.rf");
var candTwo = gce.addCandidate("/4/t0/F002.rf");
var candThree =
gce.addCandidate("hdfs://foo.com:6000/accumulo/tables/4/t0/F001.rf");
- gce.addFileReference("4", null, "/t0/F000.rf");
- gce.addFileReference("4", null, "/t0/F001.rf");
- gce.addFileReference("4", null, "/t0/F002.rf");
- gce.addFileReference("5", null, "../4/t0/F000.rf");
+ gce.addFileReference("4", null,
"hdfs://foo.com:6000/accumulo/tables/4/t0/F000.rf");
+ gce.addFileReference("4", null,
"hdfs://foo.com:6000/accumulo/tables/4/t0/F001.rf");
+ gce.addFileReference("4", null,
"hdfs://foo.com:6000/accumulo/tables/4/t0/F002.rf");
+ gce.addFileReference("5", null,
"hdfs://foo.com:6000/accumulo/tables/4/t0/F000.rf");
gce.addFileReference("6", null,
"hdfs://foo.com:6000/accumulo/tables/4/t0/F000.rf");
GarbageCollectionAlgorithm gca = new GarbageCollectionAlgorithm();
List<String[]> refsToRemove = new ArrayList<>();
- refsToRemove.add(new String[] {"4", "/t0/F000.rf"});
- refsToRemove.add(new String[] {"5", "../4/t0/F000.rf"});
+ refsToRemove.add(new String[] {"4",
"hdfs://foo.com:6000/accumulo/tables/4/t0/F000.rf"});
+ refsToRemove.add(new String[] {"5",
"hdfs://foo.com:6000/accumulo/tables/4/t0/F000.rf"});
refsToRemove.add(new String[] {"6",
"hdfs://foo.com:6000/accumulo/tables/4/t0/F000.rf"});
Collections.shuffle(refsToRemove);
@@ -443,7 +443,7 @@ public class GarbageCollectionTest {
gca.collect(gce);
assertFileDeleted(gce, cand);
- gce.removeFileReference("4", null, "/t0/F001.rf");
+ gce.removeFileReference("4", null,
"hdfs://foo.com:6000/accumulo/tables/4/t0/F001.rf");
candThree = gce.addCandidate("/4/t0/F001.rf");
gca.collect(gce);
assertFileDeleted(gce, candThree);
@@ -457,7 +457,7 @@ public class GarbageCollectionTest {
// Re-add the absolute candidate
candFour =
gce.addCandidate("hdfs://foo.com:6000/accumulo/tables/4/t0/F002.rf");
- gce.removeFileReference("4", null, "/t0/F002.rf");
+ gce.removeFileReference("4", null,
"hdfs://foo.com:6000/accumulo/tables/4/t0/F002.rf");
gca.collect(gce);
assertFileDeleted(gce, candFour);
assertNoCandidatesRemoved(gce);
@@ -530,14 +530,14 @@ public class GarbageCollectionTest {
gce.addDirReference("6", null, "t-0");
gce.addDirReference("7", null, "t-0");
- gce.addFileReference("8", "m", "/t-0/F00.rf");
- gce.addFileReference("9", "m", "/t-0/F00.rf");
+ gce.addFileReference("8", "m",
"hdfs://foo:6000/accumulo/tables/8/t-0/F00.rf");
+ gce.addFileReference("9", "m",
"hdfs://foo:6000/accumulo/tables/9/t-0/F00.rf");
gce.addFileReference("a", "m",
"hdfs://foo.com:6000/accumulo/tables/a/t-0/F00.rf");
gce.addFileReference("b", "m",
"hdfs://foo.com:6000/accumulo/tables/b/t-0/F00.rf");
- gce.addFileReference("e", "m", "../c/t-0/F00.rf");
- gce.addFileReference("f", "m", "../d/t-0/F00.rf");
+ gce.addFileReference("e", "m",
"hdfs://foo:6000/accumulo/tables/c/t-0/F00.rf");
+ gce.addFileReference("f", "m",
"hdfs://foo:6000/accumulo/tables/d/t-0/F00.rf");
GarbageCollectionAlgorithm gca = new GarbageCollectionAlgorithm();
@@ -560,12 +560,12 @@ public class GarbageCollectionTest {
assertFileDeleted(gce, new GcCandidate("/6/t-0", 3L),
new GcCandidate("hdfs://foo:6000/accumulo/tables/7/t-0/", 4L));
- gce.removeFileReference("8", "m", "/t-0/F00.rf");
- gce.removeFileReference("9", "m", "/t-0/F00.rf");
+ gce.removeFileReference("8", "m",
"hdfs://foo:6000/accumulo/tables/8/t-0/F00.rf");
+ gce.removeFileReference("9", "m",
"hdfs://foo:6000/accumulo/tables/9/t-0/F00.rf");
gce.removeFileReference("a", "m",
"hdfs://foo.com:6000/accumulo/tables/a/t-0/F00.rf");
gce.removeFileReference("b", "m",
"hdfs://foo.com:6000/accumulo/tables/b/t-0/F00.rf");
- gce.removeFileReference("e", "m", "../c/t-0/F00.rf");
- gce.removeFileReference("f", "m", "../d/t-0/F00.rf");
+ gce.removeFileReference("e", "m",
"hdfs://foo:6000/accumulo/tables/c/t-0/F00.rf");
+ gce.removeFileReference("f", "m",
"hdfs://foo:6000/accumulo/tables/d/t-0/F00.rf");
gca.collect(gce);
assertFileDeleted(gce, new GcCandidate("/8/t-0", 5L),
new GcCandidate("hdfs://foo:6000/accumulo/tables/9/t-0", 6L), new
GcCandidate("/a/t-0", 7L),
@@ -598,14 +598,14 @@ public class GarbageCollectionTest {
gce.addDirReference("6", null, "t-0");
gce.addDirReference("7", null, "t-0");
- gce.addFileReference("8", "m", "/t-0/F00.rf");
- gce.addFileReference("9", "m", "/t-0/F00.rf");
+ gce.addFileReference("8", "m",
"hdfs://foo:6000/accumulo/tables/8/t-0/F00.rf");
+ gce.addFileReference("9", "m",
"hdfs://foo:6000/accumulo/tables/9/t-0/F00.rf");
gce.addFileReference("a", "m",
"hdfs://foo.com:6000/user/foo/tables/a/t-0/F00.rf");
gce.addFileReference("b", "m",
"hdfs://foo.com:6000/user/foo/tables/b/t-0/F00.rf");
- gce.addFileReference("e", "m", "../c/t-0/F00.rf");
- gce.addFileReference("f", "m", "../d/t-0/F00.rf");
+ gce.addFileReference("e", "m",
"hdfs://foo.com:6000/user/foo/tables/c/t-0/F00.rf");
+ gce.addFileReference("f", "m",
"hdfs://foo.com:6000/user/foo/tables/d/t-0/F00.rf");
GarbageCollectionAlgorithm gca = new GarbageCollectionAlgorithm();
@@ -616,7 +616,7 @@ public class GarbageCollectionTest {
// Removing the dir reference for a table will delete all tablet
directories
gce.removeDirReference("5", null);
// but we need to add a file ref
- gce.addFileReference("8", "m", "/t-0/F00.rf");
+ gce.addFileReference("8", "m",
"hdfs://foo:6000/accumulo/tables/8/t-0/F00.rf");
gca.collect(gce);
assertFileDeleted(gce, candidates.get(3));
@@ -629,12 +629,12 @@ public class GarbageCollectionTest {
gca.collect(gce);
assertFileDeleted(gce, candidates.get(4), candidates.get(5));
- gce.removeFileReference("8", "m", "/t-0/F00.rf");
- gce.removeFileReference("9", "m", "/t-0/F00.rf");
+ gce.removeFileReference("8", "m",
"hdfs://foo:6000/accumulo/tables/8/t-0/F00.rf");
+ gce.removeFileReference("9", "m",
"hdfs://foo:6000/accumulo/tables/9/t-0/F00.rf");
gce.removeFileReference("a", "m",
"hdfs://foo.com:6000/user/foo/tables/a/t-0/F00.rf");
gce.removeFileReference("b", "m",
"hdfs://foo.com:6000/user/foo/tables/b/t-0/F00.rf");
- gce.removeFileReference("e", "m", "../c/t-0/F00.rf");
- gce.removeFileReference("f", "m", "../d/t-0/F00.rf");
+ gce.removeFileReference("e", "m",
"hdfs://foo.com:6000/user/foo/tables/c/t-0/F00.rf");
+ gce.removeFileReference("f", "m",
"hdfs://foo.com:6000/user/foo/tables/d/t-0/F00.rf");
gca.collect(gce);
assertFileDeleted(gce, candidates.get(6), candidates.get(7),
candidates.get(8),
candidates.get(9), candidates.get(10), candidates.get(11));
@@ -665,6 +665,11 @@ public class GarbageCollectionTest {
assertThrows(IllegalArgumentException.class, () -> badRefTest("/F00.rf"));
}
+ @Test
+ public void testRelativeFileRefIsInvalid() {
+ assertThrows(IllegalArgumentException.class, () ->
badRefTest("/t-0/F00.rf"));
+ }
+
@Test
public void testBadFileRef2() {
assertThrows(IllegalArgumentException.class, () ->
badRefTest("../F00.rf"));
@@ -731,7 +736,8 @@ public class GarbageCollectionTest {
gca.collect(gce);
assertFileDeleted(gce, tempCandidate);
- gce.addFileReference("1636", null, "/default_tablet/someFile");
+ gce.addFileReference("1636", null,
+ "hdfs://foo:6000/accumulo/tables/1636/default_tablet/someFile");
gce.addCandidate("/1636/default_tablet/someFile");
gca.collect(gce);
assertFileDeleted(gce);
@@ -739,7 +745,7 @@ public class GarbageCollectionTest {
// have an indirect file reference
gce = new TestGCE();
- gce.addFileReference("1636", null, "../9/default_tablet/someFile");
+ gce.addFileReference("1636", null,
"hdfs://foo:6000/accumulo/tables/9/default_tablet/someFile");
gce.addDirReference("1636", null, "default_tablet");
gce.addCandidate("/9/default_tablet/someFile");
gca.collect(gce);
@@ -834,12 +840,12 @@ public class GarbageCollectionTest {
var candidate = gce.addCandidate("/4/t0/F000.rf");
- gce.addFileReference("4", null, "/t0/F000.rf");
- gce.addFileReference("4", null, "/t0/F001.rf");
- gce.addFileReference("4", null, "/t0/F002.rf");
- gce.addFileReference("4", null, "/t0/F003.rf");
- gce.addFileReference("5", null, "../4/t0/F000.rf");
- gce.addFileReference("9", null, "/t0/F003.rf");
+ gce.addFileReference("4", null,
"hdfs://foo:6000/accumulo/tables/4/t0/F000.rf");
+ gce.addFileReference("4", null,
"hdfs://foo:6000/accumulo/tables/4/t0/F001.rf");
+ gce.addFileReference("4", null,
"hdfs://foo:6000/accumulo/tables/4/t0/F002.rf");
+ gce.addFileReference("4", null,
"hdfs://foo:6000/accumulo/tables/4/t0/F003.rf");
+ gce.addFileReference("5", null,
"hdfs://foo:6000/accumulo/tables/4/t0/F000.rf");
+ gce.addFileReference("9", null,
"hdfs://foo:6000/accumulo/tables/9/t0/F003.rf");
gce.addFileReference("6", null,
"hdfs://foo.com:6000/accumulo/tables/4/t0/F000.rf");
GarbageCollectionAlgorithm gca = new GarbageCollectionAlgorithm();
@@ -850,8 +856,8 @@ public class GarbageCollectionTest {
var cand1 = gce.addCandidate("/9/t0/F003.rf");
var cand2 = gce.addCandidate("../144/t0/F003.rf");
- gce.removeFileReference("9", null, "/t0/F003.rf");
- gce.removeFileReference("4", null, "/t0/F003.rf");
+ gce.removeFileReference("9", null,
"hdfs://foo:6000/accumulo/tables/9/t0/F003.rf");
+ gce.removeFileReference("4", null,
"hdfs://foo:6000/accumulo/tables/4/t0/F003.rf");
gca.collect(gce);
assertNoCandidatesRemoved(gce);
@@ -870,12 +876,12 @@ public class GarbageCollectionTest {
toBeRemoved[2] = gce.addCandidate("/+r/t0/F002.rf");
toBeRemoved[3] = gce.addCandidate("/+r/t0/F003.rf");
- gce.addFileReference("+r", null, "/t0/F000.rf");
- gce.addFileReference("+r", null, "/t0/F001.rf");
- gce.addFileReference("+r", null, "/t0/F002.rf");
- gce.addFileReference("+r", null, "/t0/F003.rf");
- gce.addFileReference("+r", null, "../4/t0/F000.rf");
- gce.addFileReference("+r", null, "/t0/F003.rf");
+ gce.addFileReference("+r", null,
"hdfs://foo:6000/accumulo/tables/+r/t0/F000.rf");
+ gce.addFileReference("+r", null,
"hdfs://foo:6000/accumulo/tables/+r/t0/F001.rf");
+ gce.addFileReference("+r", null,
"hdfs://foo:6000/accumulo/tables/+r/t0/F002.rf");
+ gce.addFileReference("+r", null,
"hdfs://foo:6000/accumulo/tables/+r/t0/F003.rf");
+ gce.addFileReference("+r", null,
"hdfs://foo:6000/accumulo/tables/4/t0/F000.rf");
+ gce.addFileReference("+r", null,
"hdfs://foo:6000/accumulo/tables/+r/t0/F003.rf");
gce.addFileReference("+r", null,
"hdfs://foo.com:6000/accumulo/tables/4/t0/F000.rf");
GarbageCollectionAlgorithm gca = new GarbageCollectionAlgorithm();
@@ -884,10 +890,10 @@ public class GarbageCollectionTest {
// Due to the gce Datalevel of ROOT, InUse candidate deletion is not
supported
assertNoCandidatesRemoved(gce);
- gce.removeFileReference("+r", null, "/t0/F000.rf");
- gce.removeFileReference("+r", null, "/t0/F001.rf");
- gce.removeFileReference("+r", null, "/t0/F002.rf");
- gce.removeFileReference("+r", null, "/t0/F003.rf");
+ gce.removeFileReference("+r", null,
"hdfs://foo:6000/accumulo/tables/+r/t0/F000.rf");
+ gce.removeFileReference("+r", null,
"hdfs://foo:6000/accumulo/tables/+r/t0/F001.rf");
+ gce.removeFileReference("+r", null,
"hdfs://foo:6000/accumulo/tables/+r/t0/F002.rf");
+ gce.removeFileReference("+r", null,
"hdfs://foo:6000/accumulo/tables/+r/t0/F003.rf");
// With file references deleted, the GC should now process the candidates
gca.collect(gce);
@@ -927,7 +933,7 @@ public class GarbageCollectionTest {
gce.candidates.add(candidate);
// Then create an InUse candidate for a file in that directory.
- gce.addFileReference("6", null, "/t-0/F003.rf");
+ gce.addFileReference("6", null,
"hdfs://foo:6000/accumulo/tables/6/t-0/F003.rf");
var removedCandidate = gce.addCandidate("6/t-0/F003.rf");
gca.collect(gce);
@@ -946,8 +952,8 @@ public class GarbageCollectionTest {
var scanCandidate = gce.addCandidate("/4/t0/F010.rf");
var candOne = gce.addCandidate("/4/t0/F000.rf");
var candTwo = gce.addCandidate("/6/t0/F123.rf");
- gce.addScanReference("4", null, "/t0/F010.rf");
- gce.addFileReference("4", null, "/t0/F000.rf");
+ gce.addScanReference("4", null,
"hdfs://foo:6000/accumulo/tables/4/t0/F010.rf");
+ gce.addFileReference("4", null,
"hdfs://foo:6000/accumulo/tables/4/t0/F000.rf");
GarbageCollectionAlgorithm gca = new GarbageCollectionAlgorithm();
@@ -956,7 +962,7 @@ public class GarbageCollectionTest {
assertCandidateRemoved(gce, GcCandidateType.INUSE, candOne);
assertEquals(Set.of(scanCandidate), gce.candidates);
- gce.removeScanReference("4", null, "/t0/F010.rf");
+ gce.removeScanReference("4", null,
"hdfs://foo:6000/accumulo/tables/4/t0/F010.rf");
gca.collect(gce);
assertFileDeleted(gce, scanCandidate);
assertNoCandidatesRemoved(gce);
diff --git
a/server/manager/src/main/java/org/apache/accumulo/manager/recovery/RecoveryManager.java
b/server/manager/src/main/java/org/apache/accumulo/manager/recovery/RecoveryManager.java
index 8e1e00988f..95b738629d 100644
---
a/server/manager/src/main/java/org/apache/accumulo/manager/recovery/RecoveryManager.java
+++
b/server/manager/src/main/java/org/apache/accumulo/manager/recovery/RecoveryManager.java
@@ -191,8 +191,8 @@ public class RecoveryManager {
}
String sortId = walog.getUniqueID().toString();
- String filename = walog.getPath();
- String dest = RecoveryPath.getRecoveryPath(new Path(filename)).toString();
+ String filename = walog.getPath().toString();
+ String dest = RecoveryPath.getRecoveryPath(walog.getPath()).toString();
boolean sortQueued;
synchronized (this) {
diff --git
a/server/manager/src/main/java/org/apache/accumulo/manager/upgrade/Upgrader11to12.java
b/server/manager/src/main/java/org/apache/accumulo/manager/upgrade/Upgrader11to12.java
index d4ea54cc55..cee839ee8a 100644
---
a/server/manager/src/main/java/org/apache/accumulo/manager/upgrade/Upgrader11to12.java
+++
b/server/manager/src/main/java/org/apache/accumulo/manager/upgrade/Upgrader11to12.java
@@ -80,6 +80,7 @@ import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ChoppedColumnFamily;
import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.DataFileColumnFamily;
import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ExternalCompactionColumnFamily;
+import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.LogColumnFamily;
import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ScanFileColumnFamily;
import org.apache.accumulo.core.metadata.schema.RootTabletMetadata;
import org.apache.accumulo.core.metadata.schema.TabletMergeabilityMetadata;
@@ -87,6 +88,7 @@ import
org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType;
import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
import org.apache.accumulo.core.schema.Section;
import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.tabletserver.log.LogEntry;
import org.apache.accumulo.core.util.Encoding;
import org.apache.accumulo.core.util.TextUtil;
import org.apache.accumulo.core.util.UtilWaitThread;
@@ -226,7 +228,7 @@ public class Upgrader11to12 implements Upgrader {
@VisibleForTesting
static final Set<Text> UPGRADE_FAMILIES = Set.of(DataFileColumnFamily.NAME,
CHOPPED,
- ExternalCompactionColumnFamily.NAME, ScanFileColumnFamily.NAME);
+ ExternalCompactionColumnFamily.NAME, LogColumnFamily.NAME,
ScanFileColumnFamily.NAME);
@VisibleForTesting
static final String ZNAMESPACE_NAME = "/name";
@@ -821,8 +823,74 @@ public class Upgrader11to12 implements Upgrader {
}
}
+ private static void upgradeWalEntry(Entry<Key,Value> entry, Value value,
Mutation mutation,
+ Collection<Volume> volumes) throws IOException {
+ try {
+ LogEntry.fromMetaWalEntry(entry);
+ return;
+ } catch (IllegalArgumentException e) {
+ if (e.getMessage().startsWith("Invalid path format. The path must be an
absolute URI: ")) {
+ // The wal entry passed all of the other checks, except that it's not
an absolute
+ // URI.
+ Text qualifier = entry.getKey().getColumnQualifier();
+ String qualifierString = qualifier.toString();
+ int separator = qualifierString.indexOf('/');
+ if (separator < 0) {
+ throw new IllegalArgumentException("Malformed write-ahead log " +
qualifierString);
+ }
+
+ String oldPath = qualifierString.substring(separator + 1);
+ Path qualifiedPath = qualifyWalPath(new Path(oldPath), volumes);
+ // Validate the expected WAL path structure before writing an upgraded
reference.
+ LogEntry.fromPath(qualifiedPath.toString());
+
+ Text newQualifier = new Text(qualifierString.substring(0, separator +
1) + qualifiedPath);
+
mutation.at().family(LogColumnFamily.NAME).qualifier(newQualifier).put(value);
+
mutation.at().family(LogColumnFamily.NAME).qualifier(qualifier).delete();
+ } else {
+ throw e;
+ }
+ }
+ }
+
+ private static Path qualifyWalPath(Path walPath, Collection<Volume> volumes)
throws IOException {
+ String path = walPath.toUri().getPath();
+ String[] parts = path.split("/");
+ if (parts.length < 2) {
+ throw new IllegalArgumentException(
+ "WAL path must end with tablet server and UUID: " + walPath);
+ }
+
+ String tserver = parts[parts.length - 2];
+ String uuid = parts[parts.length - 1];
+ Set<Path> matchingPaths = new HashSet<>();
+
+ for (Volume volume : volumes) {
+ Path candidate;
+ if (walPath.isAbsolute()) {
+ // An absolute path without a scheme can be qualified by each
configured volume's FS.
+ candidate = volume.getFileSystem().makeQualified(new Path(path));
+ } else {
+ // Relative references may contain an obsolete prefix. WALs are stored
under
+ // <volume>/wal/<tserver>/<uuid>, so use the identifying suffix to
construct the candidate.
+ candidate = volume
+ .prefixChild(Constants.WAL_DIR + Path.SEPARATOR + tserver +
Path.SEPARATOR + uuid);
+ }
+
+ if (volume.containsPath(candidate) &&
volume.getFileSystem().exists(candidate)) {
+ matchingPaths.add(candidate);
+ }
+ }
+
+ if (matchingPaths.size() != 1) {
+ throw new IllegalArgumentException("Expected WAL " + walPath
+ + " to exist on exactly one configured volume, but found " +
matchingPaths);
+ }
+ return matchingPaths.iterator().next();
+ }
+
void processReferences(MutationWriter batchWriter,
Iterable<Map.Entry<Key,Value>> scanner,
- String tableName) {
+ String tableName, Collection<Volume> volumes) {
try {
Mutation update = null;
for (Map.Entry<Key,Value> entry : scanner) {
@@ -847,6 +915,13 @@ public class Upgrader11to12 implements Upgrader {
var family = key.getColumnFamily();
if (family.equals(DataFileColumnFamily.NAME)) {
upgradeDataFileCF(key, value, update);
+ } else if (family.equals(LogColumnFamily.NAME)) {
+ try {
+ upgradeWalEntry(entry, value, update, volumes);
+ } catch (IllegalArgumentException | IOException e) {
+ throw new IllegalStateException("Invalid WAL reference in " +
tableName + " tablet "
+ + key.getRow() + ": " + key.getColumnQualifier(), e);
+ }
} else if (family.equals(ScanFileColumnFamily.NAME)) {
LOG.debug("Deleting scan reference from:{}. Ref: {}", tableName,
key.getRow());
update.at().family(ScanFileColumnFamily.NAME).qualifier(key.getColumnQualifier())
@@ -895,7 +970,8 @@ public class Upgrader11to12 implements Upgrader {
.forEach(entry -> entries.put(entry.getKey(), entry.getValue()));
ArrayList<Mutation> mutations = new ArrayList<>();
- processReferences(mutations::add, entries.entrySet(),
"root_table_metadata");
+ processReferences(mutations::add, entries.entrySet(),
"root_table_metadata",
+ context.getVolumeManager().getVolumes());
Preconditions.checkState(mutations.size() <= 1);
@@ -949,7 +1025,8 @@ public class Upgrader11to12 implements Upgrader {
new IsolatedScanner(context.createScanner(metaName,
Authorizations.EMPTY))) {
UPGRADE_FAMILIES.forEach(scanner::fetchColumnFamily);
scanner.setRange(MetadataSchema.TabletsSection.getRange());
- processReferences(batchWriter::addMutation, scanner, metaName);
+ processReferences(batchWriter::addMutation, scanner, metaName,
+ context.getVolumeManager().getVolumes());
} catch (TableNotFoundException ex) {
throw new IllegalStateException("Failed to find table " + metaName, ex);
} catch (MutationsRejectedException mex) {
diff --git
a/server/manager/src/test/java/org/apache/accumulo/manager/tableOps/merge/MergeTabletsTest.java
b/server/manager/src/test/java/org/apache/accumulo/manager/tableOps/merge/MergeTabletsTest.java
index 002ffbeea8..88f1a0d6bb 100644
---
a/server/manager/src/test/java/org/apache/accumulo/manager/tableOps/merge/MergeTabletsTest.java
+++
b/server/manager/src/test/java/org/apache/accumulo/manager/tableOps/merge/MergeTabletsTest.java
@@ -330,7 +330,7 @@ public class MergeTabletsTest {
testUnexpectedColumn(tmb -> tmb.putOperation(otherOpid), "had unexpected
opid",
otherOpid.toString());
- var walog = LogEntry.fromPath("localhost+8020/" + UUID.randomUUID());
+ var walog = LogEntry.fromPath("file:/accumulo/wal/localhost+8020/" +
UUID.randomUUID());
testUnexpectedColumn(tmb -> tmb.putWal(walog), "has unexpected walogs 1");
FateId ucfid1 = otherFateId;
diff --git
a/server/manager/src/test/java/org/apache/accumulo/manager/tableOps/split/UpdateTabletsTest.java
b/server/manager/src/test/java/org/apache/accumulo/manager/tableOps/split/UpdateTabletsTest.java
index 13ad3bde70..abc7b3d336 100644
---
a/server/manager/src/test/java/org/apache/accumulo/manager/tableOps/split/UpdateTabletsTest.java
+++
b/server/manager/src/test/java/org/apache/accumulo/manager/tableOps/split/UpdateTabletsTest.java
@@ -466,7 +466,7 @@ public class UpdateTabletsTest {
assertTrue(e.getMessage().contains(opid2.toString()));
// Test splitting a tablet with walogs
- var walog = LogEntry.fromPath("localhost+8020/" + UUID.randomUUID());
+ var walog = LogEntry.fromPath("file:/accumulo/wal/localhost+8020/" +
UUID.randomUUID());
var tablet4 =
TabletMetadata.builder(origExtent).putOperation(opid).putWal(walog)
.build(ColumnType.LOCATION);
e = assertThrows(IllegalStateException.class, () -> testError(origExtent,
tablet4, fateId));
diff --git
a/server/manager/src/test/java/org/apache/accumulo/manager/upgrade/Upgrader11to12Test.java
b/server/manager/src/test/java/org/apache/accumulo/manager/upgrade/Upgrader11to12Test.java
index f1f1541eb8..806ff54ab7 100644
---
a/server/manager/src/test/java/org/apache/accumulo/manager/upgrade/Upgrader11to12Test.java
+++
b/server/manager/src/test/java/org/apache/accumulo/manager/upgrade/Upgrader11to12Test.java
@@ -72,8 +72,11 @@ import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.Ch
import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.DataFileColumnFamily;
import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ExternalCompactionColumnFamily;
import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.LastLocationColumnFamily;
+import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.LogColumnFamily;
import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ScanFileColumnFamily;
import org.apache.accumulo.core.metadata.schema.RootTabletMetadata;
+import org.apache.accumulo.core.volume.Volume;
+import org.apache.accumulo.core.volume.VolumeImpl;
import org.apache.accumulo.core.zookeeper.ZooSession;
import org.apache.accumulo.manager.WithTestNames;
import org.apache.accumulo.server.ServerContext;
@@ -82,12 +85,16 @@ import
org.apache.accumulo.server.conf.store.NamespacePropKey;
import org.apache.accumulo.server.conf.store.SystemPropKey;
import org.apache.accumulo.server.conf.store.TablePropKey;
import org.apache.accumulo.server.conf.store.impl.ZooPropStore;
+import org.apache.accumulo.server.fs.VolumeManager;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.data.Stat;
import org.easymock.Capture;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -169,10 +176,15 @@ public class Upgrader11to12Test extends WithTestNames {
Key.builder(false).row(row2).family(ScanFileColumnFamily.NAME).qualifier(fileName4).build();
scanData.put(key4, new Value());
+ String walPath = "hdfs://localhost:8020/accumulo/wal/localhost+9997/" +
UUID.randomUUID();
+ Key walKey =
+
Key.builder(false).row(row2).family(LogColumnFamily.NAME).qualifier("-/" +
walPath).build();
+ scanData.put(walKey, new Value());
+
ArrayList<Mutation> mutations = new ArrayList<>();
Upgrader11to12 upgrader = new Upgrader11to12();
- upgrader.processReferences(mutations::add, scanData.entrySet(),
"accumulo.metadata");
+ upgrader.processReferences(mutations::add, scanData.entrySet(),
"accumulo.metadata", List.of());
assertEquals(2, mutations.size());
@@ -186,7 +198,87 @@ public class Upgrader11to12Test extends WithTestNames {
// 1 add, 2 delete
assertEquals(3, u2.getUpdates().size());
assertEquals(2,
u2.getUpdates().stream().filter(ColumnUpdate::isDeleted).count());
+ assertTrue(UPGRADE_FAMILIES.contains(LogColumnFamily.NAME));
+
+ }
+ @Test
+ public void processReferencesQualifiesWalPathsUsingVolume(@TempDir
java.nio.file.Path tempDir)
+ throws Exception {
+ Configuration conf = new Configuration();
+ try (FileSystem fs = FileSystem.getLocal(conf)) {
+ Volume first = new VolumeImpl(fs, tempDir.resolve("first").toString());
+ Volume second = new VolumeImpl(fs, tempDir.resolve("second").toString());
+ Text row = new Text("123");
+ TreeMap<Key,Value> scanData = new TreeMap<>();
+ List<String> oldQualifiers = new ArrayList<>();
+ List<String> expectedQualifiers = new ArrayList<>();
+
+ for (int i = 0; i < 2; i++) {
+ String walName = "localhost+9997/" + UUID.randomUUID();
+ Path existingWal = second.prefixChild(Constants.WAL_DIR + "/" +
walName);
+ fs.mkdirs(existingWal.getParent());
+ fs.create(existingWal).close();
+
+ String oldPath = i == 0 ? walName : existingWal.toUri().getPath();
+ String oldQualifier = "-/" + oldPath;
+ Key walKey = Key.builder(false).row(row).family(LogColumnFamily.NAME)
+ .qualifier(oldQualifier).build();
+ scanData.put(walKey, new Value("wal-value"));
+ oldQualifiers.add(oldQualifier);
+ expectedQualifiers.add("-/" + existingWal);
+ }
+
+ ArrayList<Mutation> mutations = new ArrayList<>();
+ new Upgrader11to12().processReferences(mutations::add,
scanData.entrySet(),
+ "accumulo.metadata", List.of(first, second));
+
+ assertEquals(1, mutations.size());
+ List<String> deletedQualifiers = new ArrayList<>();
+ List<String> addedQualifiers = new ArrayList<>();
+ for (ColumnUpdate update : mutations.get(0).getUpdates()) {
+ String qualifier = new String(update.getColumnQualifier(), UTF_8);
+ if (update.isDeleted()) {
+ deletedQualifiers.add(qualifier);
+ } else {
+ addedQualifiers.add(qualifier);
+ assertEquals("wal-value", new String(update.getValue(), UTF_8));
+ }
+ }
+ oldQualifiers.sort(null);
+ expectedQualifiers.sort(null);
+ deletedQualifiers.sort(null);
+ addedQualifiers.sort(null);
+ assertEquals(oldQualifiers, deletedQualifiers);
+ assertEquals(expectedQualifiers, addedQualifiers);
+ }
+ }
+
+ @Test
+ public void processReferencesRejectsAmbiguousRelativeWalPath(@TempDir
java.nio.file.Path tempDir)
+ throws Exception {
+ Configuration conf = new Configuration();
+ try (FileSystem fs = FileSystem.getLocal(conf)) {
+ Volume first = new VolumeImpl(fs, tempDir.resolve("first").toString());
+ Volume second = new VolumeImpl(fs, tempDir.resolve("second").toString());
+ String walName = "localhost+9997/" + UUID.randomUUID();
+ for (Volume volume : List.of(first, second)) {
+ Path existingWal = volume.prefixChild(Constants.WAL_DIR + "/" +
walName);
+ fs.mkdirs(existingWal.getParent());
+ fs.create(existingWal).close();
+ }
+
+ Text row = new Text("123");
+ Key walKey = Key.builder(false).row(row).family(LogColumnFamily.NAME)
+ .qualifier("-/" + walName).build();
+ var scanData = Map.of(walKey, new Value());
+ var error = assertThrows(IllegalStateException.class,
+ () -> new Upgrader11to12().processReferences(mutation -> {},
scanData.entrySet(),
+ "accumulo.metadata", List.of(first, second)));
+
+ assertTrue(error.getMessage().contains("accumulo.metadata tablet 123"));
+ assertTrue(error.getCause().getMessage().contains("exactly one
configured volume"));
+ }
}
@Test
@@ -212,7 +304,7 @@ public class Upgrader11to12Test extends WithTestNames {
ArrayList<Mutation> mutations = new ArrayList<>();
Upgrader11to12 upgrader = new Upgrader11to12();
- upgrader.processReferences(mutations::add, scanData.entrySet(),
"accumulo.metadata");
+ upgrader.processReferences(mutations::add, scanData.entrySet(),
"accumulo.metadata", List.of());
assertEquals(1, mutations.size());
@@ -245,8 +337,9 @@ public class Upgrader11to12Test extends WithTestNames {
replay(batchWriter);
Upgrader11to12 upgrader = new Upgrader11to12();
- assertThrows(IllegalStateException.class, () -> upgrader
- .processReferences(batchWriter::addMutation, scanData.entrySet(),
"accumulo.metadata"));
+ assertThrows(IllegalStateException.class,
+ () -> upgrader.processReferences(batchWriter::addMutation,
scanData.entrySet(),
+ "accumulo.metadata", List.of()));
verify(batchWriter);
}
@@ -279,8 +372,9 @@ public class Upgrader11to12Test extends WithTestNames {
replay(batchWriter);
Upgrader11to12 upgrader = new Upgrader11to12();
- assertThrows(IllegalArgumentException.class, () -> upgrader
- .processReferences(batchWriter::addMutation, scanData.entrySet(),
"accumulo.metadata"));
+ assertThrows(IllegalArgumentException.class,
+ () -> upgrader.processReferences(batchWriter::addMutation,
scanData.entrySet(),
+ "accumulo.metadata", List.of()));
verify(batchWriter);
}
@@ -306,8 +400,9 @@ public class Upgrader11to12Test extends WithTestNames {
replay(batchWriter);
Upgrader11to12 upgrader = new Upgrader11to12();
- assertThrows(IllegalStateException.class, () -> upgrader
- .processReferences(batchWriter::addMutation, scanData.entrySet(),
"accumulo.metadata"));
+ assertThrows(IllegalStateException.class,
+ () -> upgrader.processReferences(batchWriter::addMutation,
scanData.entrySet(),
+ "accumulo.metadata", List.of()));
verify(batchWriter);
}
@@ -352,7 +447,7 @@ public class Upgrader11to12Test extends WithTestNames {
ArrayList<Mutation> mutations = new ArrayList<>();
Upgrader11to12 upgrader = new Upgrader11to12();
- upgrader.processReferences(mutations::add, scanData.entrySet(),
"accumulo.metadata");
+ upgrader.processReferences(mutations::add, scanData.entrySet(),
"accumulo.metadata", List.of());
assertEquals(1, mutations.size());
var u1 = mutations.get(0);
@@ -377,12 +472,15 @@ public class Upgrader11to12Test extends WithTestNames {
Upgrader11to12 upgrader = new Upgrader11to12();
ServerContext context = createMock(ServerContext.class);
+ VolumeManager volumeManager = createMock(VolumeManager.class);
ZooSession zk = createStrictMock(ZooSession.class);
ZooReaderWriter zrw = createStrictMock(ZooReaderWriter.class);
ZooPropStore store = createMock(ZooPropStore.class);
expect(context.getInstanceID()).andReturn(iid).anyTimes();
expect(context.getZooSession()).andReturn(zk).anyTimes();
+ expect(context.getVolumeManager()).andReturn(volumeManager).once();
+ expect(volumeManager.getVolumes()).andReturn(List.of()).once();
expect(zk.asReaderWriter()).andReturn(zrw).anyTimes();
zrw.recursiveDelete("/tracers", ZooUtil.NodeMissingPolicy.SKIP);
@@ -452,7 +550,7 @@ public class Upgrader11to12Test extends WithTestNames {
upgrader.processReferences(mutations::add,
rtm.getKeyValues().filter(e ->
UPGRADE_FAMILIES.contains(e.getKey().getColumnFamily()))
.collect(Collectors.toList()),
- "accumulo.metadata");
+ "accumulo.metadata", List.of());
assertEquals(1, mutations.size());
var mutation = mutations.get(0);
rtm.update(mutation);
@@ -478,7 +576,7 @@ public class Upgrader11to12Test extends WithTestNames {
upgrader.processReferences(mutations::add,
rtm.getKeyValues().filter(e ->
UPGRADE_FAMILIES.contains(e.getKey().getColumnFamily()))
.collect(Collectors.toList()),
- "accumulo.metadata");
+ "accumulo.metadata", List.of());
assertEquals(1, mutations.size());
var mutation = mutations.get(0);
rtm.update(mutation);
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/InMemoryMap.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/InMemoryMap.java
index 5f0296037e..e6885b1dbd 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/InMemoryMap.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/InMemoryMap.java
@@ -134,7 +134,7 @@ public class InMemoryMap {
boolean useNativeMap = config.getBoolean(Property.TSERV_NATIVEMAP_ENABLED);
- this.memDumpDir = config.get(Property.TSERV_MEMDUMP_DIR);
+ this.memDumpDir = config.getPath(Property.TSERV_MEMDUMP_DIR);
this.lggroups = LocalityGroupUtil.getLocalityGroupsIgnoringErrors(config,
tableId);
this.config = config;
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java
index 9907f9762a..8858068aa4 100644
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java
@@ -1297,7 +1297,8 @@ public class TabletClientHandler implements
TabletServerClientService.Iface,
public List<String> getActiveLogs(TInfo tinfo, TCredentials credentials) {
// Might be null if there is no active logger
LogEntry le = server.logger.getLogEntry();
- return le == null ? Collections.emptyList() :
Collections.singletonList(le.getPath());
+ return le == null ? Collections.emptyList()
+ : Collections.singletonList(le.getPath().toString());
}
@Override
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/DfsLogger.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/DfsLogger.java
index d021ff111b..044f90407b 100644
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/DfsLogger.java
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/DfsLogger.java
@@ -41,12 +41,10 @@ import java.util.Collection;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
-import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;
-import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.Durability;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.Property;
@@ -80,6 +78,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.base.Preconditions;
+import com.google.common.net.HostAndPort;
/**
* Wrap a connection to a logger.
@@ -314,22 +313,17 @@ public final class DfsLogger implements
Comparable<DfsLogger> {
* Create a new DfsLogger with the provided characteristics.
*/
public static DfsLogger createNew(ServerContext context, AtomicLong
syncCounter,
- AtomicLong flushCounter, String address) throws IOException {
-
- String filename = UUID.randomUUID().toString();
- String addressForFilename = address.replace(':', '+');
+ AtomicLong flushCounter, HostAndPort address) throws IOException {
var chooserEnv =
new
VolumeChooserEnvironmentImpl(VolumeChooserEnvironment.Scope.LOGGER, context);
- String logPath =
- context.getVolumeManager().choose(chooserEnv, context.getBaseUris()) +
Path.SEPARATOR
- + Constants.WAL_DIR + Path.SEPARATOR + addressForFilename +
Path.SEPARATOR + filename;
+ String logVolume = context.getVolumeManager().choose(chooserEnv,
context.getBaseUris());
- LogEntry log = LogEntry.fromPath(logPath);
+ LogEntry log = LogEntry.forServer(logVolume, address);
DfsLogger dfsLogger = new DfsLogger(log);
long slowFlushMillis =
context.getConfiguration().getTimeInMillis(Property.TSERV_SLOW_FLUSH_MILLIS);
- dfsLogger.open(context, logPath, filename, address, syncCounter,
flushCounter, slowFlushMillis);
+ dfsLogger.open(context, syncCounter, flushCounter, slowFlushMillis);
return dfsLogger;
}
@@ -398,13 +392,10 @@ public final class DfsLogger implements
Comparable<DfsLogger> {
* file. The file is ready to be used for ingest if this method returns
successfully. If an
* exception is thrown from this method, it is the callers responsibility to
ensure that
* {@link #close()} is called to prevent leaking the file handle and/or
syncing thread.
- *
- * @param address The address of the host using this WAL
*/
- private synchronized void open(ServerContext context, String logPath, String
filename,
- String address, AtomicLong syncCounter, AtomicLong flushCounter, long
slowFlushMillis)
- throws IOException {
- log.debug("Address is {}", address);
+ private synchronized void open(ServerContext context, AtomicLong syncCounter,
+ AtomicLong flushCounter, long slowFlushMillis) throws IOException {
+ log.debug("Address is {}", logEntry.getTServer());
log.debug("DfsLogger.open() begin");
@@ -413,7 +404,7 @@ public final class DfsLogger implements
Comparable<DfsLogger> {
LoggerOperation op;
var serverConf = context.getConfiguration();
try {
- Path logfilePath = new Path(logPath);
+ Path logfilePath = logEntry.getPath();
short replication = (short)
serverConf.getCount(Property.TSERV_WAL_REPLICATION);
if (replication == 0) {
replication = fs.getDefaultReplication(logfilePath);
@@ -436,7 +427,7 @@ public final class DfsLogger implements
Comparable<DfsLogger> {
// check again that logfile can be sync'd
if (!fs.canSyncAndFlush(logfilePath)) {
- log.warn("sync not supported for log file {}. Data loss may occur.",
logPath);
+ log.warn("sync not supported for log file {}. Data loss may occur.",
logEntry.getPath());
}
// Initialize the log file with a header and its encryption
@@ -446,7 +437,7 @@ public final class DfsLogger implements
Comparable<DfsLogger> {
logFile.write(LOG_FILE_HEADER_V4.getBytes(UTF_8));
log.debug("Using {} for encrypting WAL {}",
cryptoService.getClass().getSimpleName(),
- filename);
+ logEntry.getUniqueID());
FileEncrypter encrypter = cryptoService.getFileEncrypter(env);
byte[] cryptoParams = encrypter.getDecryptionParameters();
CryptoUtils.writeParams(cryptoParams, logFile);
@@ -465,8 +456,8 @@ public final class DfsLogger implements
Comparable<DfsLogger> {
LogFileKey key = new LogFileKey();
key.setEvent(OPEN);
- key.setTserverSession(filename);
- key.setFilename(filename);
+ key.setTserverSession(logEntry.getUniqueID().toString());
+ key.setFilename(logEntry.getUniqueID().toString());
op = logKeyData(key, Durability.SYNC);
} catch (Exception ex) {
if (logFile != null) {
@@ -502,7 +493,7 @@ public final class DfsLogger implements
Comparable<DfsLogger> {
}
public Path getPath() {
- return new Path(logEntry.getPath());
+ return logEntry.getPath();
}
public void close() throws IOException {
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/ResolvedSortedLog.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/ResolvedSortedLog.java
index 0de01e5ee5..1b87af6466 100644
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/ResolvedSortedLog.java
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/ResolvedSortedLog.java
@@ -87,7 +87,7 @@ public class ResolvedSortedLog {
// convert the path of an unsorted log to the expected path for the
corresponding sorted log
// dir
- Path sortedLogPath = RecoveryPath.getRecoveryPath(new
Path(logEntry.getPath()));
+ Path sortedLogPath = RecoveryPath.getRecoveryPath(logEntry.getPath());
boolean foundFinish = false;
// Path::getName compares the last component of each Path value. In this
case, the last
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java
index b07fa942cf..f21580a386 100644
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java
@@ -278,7 +278,7 @@ public class TabletServerLogger {
try {
alog = DfsLogger.createNew(tserver.getContext(), syncCounter,
flushCounter,
- tserver.getAdvertiseAddress().toString());
+ tserver.getAdvertiseAddress());
} catch (IOException | RuntimeException e) {
log.error("Failed to open WAL", e);
try {
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java
index c41455850f..1acc4fbaf2 100644
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java
@@ -136,7 +136,8 @@ public class LogReader extends
ServerKeywordExecutable<ReaderOpts> {
Set<Integer> tabletIds = new HashSet<>();
for (String file : opts.files) {
- Path path = new Path(file);
+ Path tmpPath = new Path(file);
+ Path path = fs.getFileSystemByPath(tmpPath).makeQualified(tmpPath);
LogFileKey key = new LogFileKey();
LogFileValue value = new LogFileValue();
diff --git
a/server/tserver/src/test/java/org/apache/accumulo/tserver/WalRemovalOrderTest.java
b/server/tserver/src/test/java/org/apache/accumulo/tserver/WalRemovalOrderTest.java
index 2e682b4231..0dbddd0a5d 100644
---
a/server/tserver/src/test/java/org/apache/accumulo/tserver/WalRemovalOrderTest.java
+++
b/server/tserver/src/test/java/org/apache/accumulo/tserver/WalRemovalOrderTest.java
@@ -34,7 +34,8 @@ import com.google.common.collect.Sets;
public class WalRemovalOrderTest {
private DfsLogger mockLogger(String filename) {
- var mockLogEntry = LogEntry.fromPath(filename +
"+1234/11111111-1111-1111-1111-111111111111");
+ var mockLogEntry = LogEntry
+ .fromPath("file:/accumulo/wal/" + filename +
"+1234/11111111-1111-1111-1111-111111111111");
return DfsLogger.fromLogEntry(mockLogEntry);
}
diff --git
a/shell/src/test/java/org/apache/accumulo/shell/commands/ListTabletsCommandTest.java
b/shell/src/test/java/org/apache/accumulo/shell/commands/ListTabletsCommandTest.java
index a12c23e3c5..44c019126e 100644
---
a/shell/src/test/java/org/apache/accumulo/shell/commands/ListTabletsCommandTest.java
+++
b/shell/src/test/java/org/apache/accumulo/shell/commands/ListTabletsCommandTest.java
@@ -191,8 +191,8 @@ public class ListTabletsCommandTest {
KeyExtent extent = new KeyExtent(tableId, new Text("d"), null);
- LogEntry le1 = LogEntry.fromPath("localhost+8020/" + UUID.randomUUID());
- LogEntry le2 = LogEntry.fromPath("localhost+8020/" + UUID.randomUUID());
+ LogEntry le1 = LogEntry.fromPath("file:/accumulo/wal/localhost+8020/" +
UUID.randomUUID());
+ LogEntry le2 = LogEntry.fromPath("file:/accumulo/wal/localhost+8020/" +
UUID.randomUUID());
TabletMetadata tm1 =
TabletMetadata.builder(extent).putTabletAvailability(TabletAvailability.ONDEMAND)
diff --git
a/test/src/main/java/org/apache/accumulo/test/MissingWalHeaderCompletesRecoveryIT.java
b/test/src/main/java/org/apache/accumulo/test/MissingWalHeaderCompletesRecoveryIT.java
index e33abf148b..28f4c9d8bd 100644
---
a/test/src/main/java/org/apache/accumulo/test/MissingWalHeaderCompletesRecoveryIT.java
+++
b/test/src/main/java/org/apache/accumulo/test/MissingWalHeaderCompletesRecoveryIT.java
@@ -24,9 +24,7 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.time.Duration;
-import java.util.UUID;
-import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.Accumulo;
import org.apache.accumulo.core.client.AccumuloClient;
import org.apache.accumulo.core.client.BatchWriter;
@@ -46,7 +44,6 @@ import org.apache.accumulo.tserver.log.DfsLogger;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RawLocalFileSystem;
import org.apache.hadoop.io.Text;
import org.junit.jupiter.api.AfterEach;
@@ -55,6 +52,8 @@ import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.net.HostAndPort;
+
public class MissingWalHeaderCompletesRecoveryIT extends ConfigurableMacBase {
private static final Logger log =
LoggerFactory.getLogger(MissingWalHeaderCompletesRecoveryIT.class);
@@ -114,17 +113,13 @@ public class MissingWalHeaderCompletesRecoveryIT extends
ConfigurableMacBase {
MiniAccumuloClusterImpl cluster = getCluster();
FileSystem fs = cluster.getFileSystem();
- // Fake out something that looks like host:port, it's irrelevant
- String fakeServer = "127.0.0.1:12345";
-
- java.nio.file.Path walogs =
-
cluster.getConfig().getAccumuloDir().toPath().resolve(Constants.WAL_DIR);
- java.nio.file.Path walogServerDir =
walogs.resolve(fakeServer.replace(':', '+'));
- java.nio.file.Path emptyWalog =
walogServerDir.resolve(UUID.randomUUID().toString());
+ LogEntry logEntry =
+
LogEntry.forServer(cluster.getConfig().getAccumuloDir().toPath().toUri().toString(),
+ HostAndPort.fromParts("127.0.0.1", 12345));
- log.info("Created empty WAL at {}", emptyWalog.toUri());
+ log.info("Created empty WAL at {}", logEntry.getPath().toUri());
- fs.create(new Path(emptyWalog.toUri())).close();
+ fs.create(logEntry.getPath()).close();
assertTrue(client.securityOperations().hasTablePermission("root",
SystemTables.METADATA.tableName(), TablePermission.WRITE),
@@ -136,8 +131,6 @@ public class MissingWalHeaderCompletesRecoveryIT extends
ConfigurableMacBase {
TableId tableId =
TableId.of(client.tableOperations().tableIdMap().get(tableName));
assertNotNull(tableId, "Table ID was null");
- LogEntry logEntry = LogEntry.fromPath(emptyWalog.toUri().toString());
-
log.info("Taking {} offline", tableName);
client.tableOperations().offline(tableName, true);
@@ -168,20 +161,15 @@ public class MissingWalHeaderCompletesRecoveryIT extends
ConfigurableMacBase {
public void testPartialHeaderWalRecoveryCompletes() throws Exception {
try (AccumuloClient client =
Accumulo.newClient().from(getClientProperties()).build()) {
MiniAccumuloClusterImpl cluster = getCluster();
- FileSystem fs = getCluster().getFileSystem();
-
- // Fake out something that looks like host:port, it's irrelevant
- String fakeServer = "127.0.0.1:12345";
-
- java.nio.file.Path walogs =
-
cluster.getConfig().getAccumuloDir().toPath().resolve(Constants.WAL_DIR);
- java.nio.file.Path walogServerDir =
walogs.resolve(fakeServer.replace(':', '+'));
- java.nio.file.Path partialHeaderWalog =
walogServerDir.resolve(UUID.randomUUID().toString());
+ FileSystem fs = cluster.getFileSystem();
- log.info("Created WAL with malformed header at {}",
partialHeaderWalog.toUri());
+ LogEntry logEntry =
+
LogEntry.forServer(cluster.getConfig().getAccumuloDir().toPath().toUri().toString(),
+ HostAndPort.fromParts("127.0.0.1", 12345));
+ log.info("Created WAL with malformed header at {}",
logEntry.getPath().toUri());
// Write half of the header
- FSDataOutputStream wal = fs.create(new Path(partialHeaderWalog.toUri()));
+ FSDataOutputStream wal = fs.create(logEntry.getPath());
wal.write(DfsLogger.LOG_FILE_HEADER_V4.getBytes(UTF_8), 0,
DfsLogger.LOG_FILE_HEADER_V4.length() / 2);
wal.close();
@@ -196,8 +184,6 @@ public class MissingWalHeaderCompletesRecoveryIT extends
ConfigurableMacBase {
TableId tableId =
TableId.of(client.tableOperations().tableIdMap().get(tableName));
assertNotNull(tableId, "Table ID was null");
- LogEntry logEntry =
LogEntry.fromPath(partialHeaderWalog.toUri().toString());
-
log.info("Taking {} offline", tableName);
client.tableOperations().offline(tableName, true);
diff --git a/test/src/main/java/org/apache/accumulo/test/RecoveryIT.java
b/test/src/main/java/org/apache/accumulo/test/RecoveryIT.java
index 2e9c8198e6..fd4642e3ae 100644
--- a/test/src/main/java/org/apache/accumulo/test/RecoveryIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/RecoveryIT.java
@@ -55,7 +55,6 @@ import org.apache.accumulo.test.functional.ReadWriteIT;
import org.apache.accumulo.test.harness.AccumuloClusterHarness;
import org.apache.accumulo.test.util.Wait;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RawLocalFileSystem;
import org.apache.hadoop.io.Text;
import org.junit.jupiter.api.Tag;
@@ -225,8 +224,7 @@ public class RecoveryIT extends AccumuloClusterHarness {
for (TabletMetadata tm : tabletsMetadata) {
for (LogEntry walog : tm.getLogs()) {
String sortId = walog.getUniqueID().toString();
- String filename = walog.getPath();
- String dest = RecoveryPath.getRecoveryPath(new
Path(filename)).toString();
+ String dest =
RecoveryPath.getRecoveryPath(walog.getPath()).toString();
if (ctx.getZooCache().get(Constants.ZRECOVERY + "/" + sortId) != null
||
!ctx.getVolumeManager().exists(SortedLogState.getFinishedMarkerPath(dest))) {
diff --git
a/test/src/main/java/org/apache/accumulo/test/fate/ManagerRepoIT_SimpleSuite.java
b/test/src/main/java/org/apache/accumulo/test/fate/ManagerRepoIT_SimpleSuite.java
index 6f39b63be1..ddee24d31e 100644
---
a/test/src/main/java/org/apache/accumulo/test/fate/ManagerRepoIT_SimpleSuite.java
+++
b/test/src/main/java/org/apache/accumulo/test/fate/ManagerRepoIT_SimpleSuite.java
@@ -31,7 +31,6 @@ import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import java.nio.file.Path;
import java.time.Duration;
import java.util.Arrays;
import java.util.HashSet;
@@ -103,6 +102,7 @@ import org.junit.jupiter.params.provider.EnumSource;
import org.junit.jupiter.params.provider.MethodSource;
import com.google.common.collect.Sets;
+import com.google.common.net.HostAndPort;
public class ManagerRepoIT_SimpleSuite extends SharedMiniClusterBase {
@@ -160,8 +160,11 @@ public class ManagerRepoIT_SimpleSuite extends
SharedMiniClusterBase {
// Write a WAL to the test metadata and then re-run the repo to check
for an error
try (TabletsMutator tm = testAmple.mutateTablets()) {
- var walFilePath = Path.of("tserver+8080",
UUID.randomUUID().toString()).toString();
-
tm.mutateTablet(extent).putWal(LogEntry.fromPath(walFilePath)).mutate();
+ tm.mutateTablet(extent)
+ .putWal(LogEntry.forServer(
+
getCluster().getConfig().getAccumuloDir().toPath().toUri().toString(),
+ HostAndPort.fromParts("tserver", 8080)))
+ .mutate();
}
// Should not be ready due to the presence of a WAL
diff --git
a/test/src/main/java/org/apache/accumulo/test/functional/AmpleConditionalWriterIT.java
b/test/src/main/java/org/apache/accumulo/test/functional/AmpleConditionalWriterIT.java
index 713eabdf82..85e609759d 100644
---
a/test/src/main/java/org/apache/accumulo/test/functional/AmpleConditionalWriterIT.java
+++
b/test/src/main/java/org/apache/accumulo/test/functional/AmpleConditionalWriterIT.java
@@ -117,6 +117,7 @@ import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
import com.google.common.collect.Sets;
+import com.google.common.net.HostAndPort;
public class AmpleConditionalWriterIT extends SharedMiniClusterBase {
@@ -471,9 +472,9 @@ public class AmpleConditionalWriterIT extends
SharedMiniClusterBase {
var context = getCluster().getServerContext();
// Test adding a WAL to a tablet and verifying its presence
- String walFilePath =
- java.nio.file.Path.of("tserver+8080",
UUID.randomUUID().toString()).toString();
- final LogEntry originalLogEntry = LogEntry.fromPath(walFilePath);
+ final LogEntry originalLogEntry =
+
LogEntry.forServer(getCluster().getConfig().getAccumuloDir().toPath().toUri().toString(),
+ HostAndPort.fromParts("tserver", 8080));
// create a tablet metadata with no write ahead logs
var tmEmptySet = TabletMetadata.builder(e1).build(LOGS);
// tablet should not have any logs to start with so requireSame with the
empty logs should pass
@@ -489,9 +490,9 @@ public class AmpleConditionalWriterIT extends
SharedMiniClusterBase {
"The original LogEntry should be present.");
// Test adding another WAL and verifying the update
- String walFilePath2 =
- java.nio.file.Path.of("tserver+8080",
UUID.randomUUID().toString()).toString();
- LogEntry newLogEntry = LogEntry.fromPath(walFilePath2);
+ LogEntry newLogEntry =
+
LogEntry.forServer(getCluster().getConfig().getAccumuloDir().toPath().toUri().toString(),
+ HostAndPort.fromParts("tserver", 8080));
try (var ctmi = context.getAmple().conditionallyMutateTablets()) {
ctmi.mutateTablet(e1).requireAbsentOperation().putWal(newLogEntry).submit(tm ->
false);
assertEquals(Status.ACCEPTED, ctmi.process().get(e1).getStatus());
@@ -502,10 +503,9 @@ public class AmpleConditionalWriterIT extends
SharedMiniClusterBase {
HashSet<LogEntry> actualLogs = new
HashSet<>(context.getAmple().readTablet(e1).getLogs());
assertEquals(expectedLogs, actualLogs, "Both original and new LogEntry
should be present.");
- String walFilePath3 =
- java.nio.file.Path.of("tserver+8080",
UUID.randomUUID().toString()).toString();
- LogEntry otherLogEntry = LogEntry.fromPath(walFilePath3);
-
+ LogEntry otherLogEntry =
+
LogEntry.forServer(getCluster().getConfig().getAccumuloDir().toPath().toUri().toString(),
+ HostAndPort.fromParts("tserver", 8080));
// create a powerset to ensure all possible subsets fail when using
requireSame except the
// expected current state
Set<LogEntry> allLogs = Set.of(originalLogEntry, newLogEntry,
otherLogEntry);
@@ -1234,9 +1234,9 @@ public class AmpleConditionalWriterIT extends
SharedMiniClusterBase {
"Initially, all tablets should be present");
String server = "server1+8555";
-
- String walFilePath = java.nio.file.Path.of(server,
UUID.randomUUID().toString()).toString();
- LogEntry wal = LogEntry.fromPath(walFilePath);
+ LogEntry wal =
+
LogEntry.forServer(getCluster().getConfig().getAccumuloDir().toPath().toUri().toString(),
+ HostAndPort.fromParts("server1", 8555));
// add wal compact and flush ID to these tablets
final Set<KeyExtent> tabletsWithWalCompactFlush = Set.of(e1, e2, e3);
@@ -1338,9 +1338,9 @@ public class AmpleConditionalWriterIT extends
SharedMiniClusterBase {
"Initially, all tablets should be present");
// add a wal to e2
- var walFilePath =
- java.nio.file.Path.of("tserver+8080",
UUID.randomUUID().toString()).toString();
- var wal = LogEntry.fromPath(walFilePath);
+ var wal =
+
LogEntry.forServer(getCluster().getConfig().getAccumuloDir().toPath().toUri().toString(),
+ HostAndPort.fromParts("tserver", 8080));
try (var ctmi = context.getAmple().conditionallyMutateTablets()) {
ctmi.mutateTablet(e2).requireAbsentOperation().putWal(wal).submit(tabletMetadata
-> false);
assertEquals(Status.ACCEPTED, ctmi.process().get(e2).getStatus());
@@ -1350,8 +1350,9 @@ public class AmpleConditionalWriterIT extends
SharedMiniClusterBase {
testFilterApplied(context, filter, Set.of(e2), "Only tablets with wals
should be returned");
// add wal to tablet e4
- walFilePath = java.nio.file.Path.of("tserver+8080",
UUID.randomUUID().toString()).toString();
- wal = LogEntry.fromPath(walFilePath);
+ wal =
+
LogEntry.forServer(getCluster().getConfig().getAccumuloDir().toPath().toUri().toString(),
+ HostAndPort.fromParts("tserver", 8080));
try (var ctmi = context.getAmple().conditionallyMutateTablets()) {
ctmi.mutateTablet(e4).requireAbsentOperation().putWal(wal).submit(tabletMetadata
-> false);
assertEquals(Status.ACCEPTED, ctmi.process().get(e4).getStatus());
diff --git
a/test/src/main/java/org/apache/accumulo/test/functional/TabletManagementIteratorIT.java
b/test/src/main/java/org/apache/accumulo/test/functional/TabletManagementIteratorIT.java
index 8936f6e80b..3ef429b89c 100644
---
a/test/src/main/java/org/apache/accumulo/test/functional/TabletManagementIteratorIT.java
+++
b/test/src/main/java/org/apache/accumulo/test/functional/TabletManagementIteratorIT.java
@@ -98,6 +98,7 @@ import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.tabletserver.log.LogEntry;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.core.util.time.SteadyTime;
+import org.apache.accumulo.miniclusterImpl.MiniAccumuloClusterImpl;
import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
import org.apache.accumulo.server.manager.LiveTServerSet;
import org.apache.accumulo.server.manager.state.TabletManagementIterator;
@@ -114,6 +115,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.collect.Sets;
+import com.google.common.net.HostAndPort;
/**
* Test to ensure that the {@link TabletManagementIterator} properly skips
over tablet information
@@ -671,8 +673,8 @@ public class TabletManagementIteratorIT extends
AccumuloClusterHarness {
TableId.of(client.tableOperations().tableIdMap().get(tableNameToModify));
KeyExtent extent = new KeyExtent(tableIdToModify, new Text("some split"),
null);
Mutation m = new Mutation(extent.toMetaRow());
- String fileName = "file:/accumulo/wal/localhost+9997/" +
UUID.randomUUID().toString();
- LogEntry logEntry = LogEntry.fromPath(fileName);
+ LogEntry logEntry = LogEntry.forServer(((MiniAccumuloClusterImpl)
getCluster()).getConfig()
+ .getAccumuloDir().toPath().toUri().toString(),
HostAndPort.fromParts("localhost", 9997));
logEntry.addToMutation(m);
try (BatchWriter bw = client.createBatchWriter(table)) {
bw.addMutation(m);
