This is an automated email from the ASF dual-hosted git repository.
ctubbsii 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 37117b2cb5 Use LogEntry concrete type in more places (#4085)
37117b2cb5 is described below
commit 37117b2cb5e2d17d70adf832f72103432085feca
Author: Christopher Tubbs <ctubb...@apache.org>
AuthorDate: Thu Dec 21 00:15:48 2023 -0500
Use LogEntry concrete type in more places (#4085)
* Track the actual column qualifier read from the metadata table, so
when it is deleted, it can delete the actual entry, rather than try to
delete the existing entry with the new format (requires no longer
strictly requiring that log entries start with a `-/`).
This fixes #4061 for the main branch for 3.1; a similar issue may
still exist for 2.1, but will require a different fix.
* Remove use of `Collection<Collection<String>>` and use
`Collection<LogEntry>` where possible, since the former was for a
previous WAL storage format that we do not support, and the latter
preserves the strongly-typed LogEntry in more places
---
.../core/metadata/TabletLocationState.java | 6 +-
.../accumulo/core/tabletserver/log/LogEntry.java | 25 +++--
.../core/tabletserver/log/LogEntryTest.java | 13 +--
.../org/apache/accumulo/server/fs/VolumeUtil.java | 2 +-
.../server/manager/state/MetaDataTableScanner.java | 5 +-
.../server/manager/state/ZooTabletStateStore.java | 7 +-
.../manager/state/TabletLocationStateTest.java | 14 +--
.../accumulo/gc/GarbageCollectWriteAheadLogs.java | 25 ++---
.../gc/GarbageCollectWriteAheadLogsTest.java | 3 +-
.../accumulo/manager/recovery/RecoveryManager.java | 108 ++++++++++-----------
.../accumulo/manager/state/MergeStatsTest.java | 7 +-
.../apache/accumulo/test/manager/MergeStateIT.java | 2 +-
12 files changed, 107 insertions(+), 110 deletions(-)
diff --git
a/core/src/main/java/org/apache/accumulo/core/metadata/TabletLocationState.java
b/core/src/main/java/org/apache/accumulo/core/metadata/TabletLocationState.java
index 03c3dce59d..eeaa5cde95 100644
---
a/core/src/main/java/org/apache/accumulo/core/metadata/TabletLocationState.java
+++
b/core/src/main/java/org/apache/accumulo/core/metadata/TabletLocationState.java
@@ -27,6 +27,7 @@ import java.util.Set;
import org.apache.accumulo.core.dataImpl.KeyExtent;
import org.apache.accumulo.core.metadata.schema.TabletMetadata;
import org.apache.accumulo.core.metadata.schema.TabletMetadata.Location;
+import org.apache.accumulo.core.tabletserver.log.LogEntry;
import org.apache.accumulo.core.util.TextUtil;
import org.apache.hadoop.io.Text;
@@ -56,8 +57,7 @@ public class TabletLocationState {
}
public TabletLocationState(KeyExtent extent, Location future, Location
current, Location last,
- SuspendingTServer suspend, Collection<Collection<String>> walogs)
- throws BadLocationStateException {
+ SuspendingTServer suspend, Collection<LogEntry> walogs) throws
BadLocationStateException {
this.extent = extent;
this.future = validateLocation(future, TabletMetadata.LocationType.FUTURE);
this.current = validateLocation(current,
TabletMetadata.LocationType.CURRENT);
@@ -79,7 +79,7 @@ public class TabletLocationState {
public final Location current;
public final Location last;
public final SuspendingTServer suspend;
- public final Collection<Collection<String>> walogs;
+ public final Collection<LogEntry> walogs;
public TServerInstance getCurrentServer() {
return serverInstance(current);
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 dea6460eaf..6bd5ba57b4 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
@@ -38,11 +38,13 @@ public final class LogEntry {
private final String path;
private final HostAndPort tserver;
private final UUID uniqueId;
+ private final Text columnQualifier;
- private LogEntry(String path, HostAndPort tserver, UUID uniqueId) {
+ private LogEntry(String path, HostAndPort tserver, UUID uniqueId, Text
columnQualifier) {
this.path = path;
this.tserver = tserver;
this.uniqueId = uniqueId;
+ this.columnQualifier = columnQualifier;
}
/**
@@ -57,6 +59,10 @@ public final class LogEntry {
* @throws IllegalArgumentException if the path is invalid
*/
public static LogEntry fromPath(String path) {
+ return validatedLogEntry(path, null);
+ }
+
+ private static LogEntry validatedLogEntry(String path, Text columnQualifier)
{
String[] parts = path.split("/");
if (parts.length < 2) {
@@ -90,7 +96,7 @@ public final class LogEntry {
throw new IllegalArgumentException(badUuidMsg);
}
- return new LogEntry(path, tserver, uuid);
+ return new LogEntry(path, tserver, uuid, columnQualifier);
}
/**
@@ -106,11 +112,10 @@ public final class LogEntry {
Preconditions.checkArgument(LogColumnFamily.NAME.equals(fam),
"The provided metadata entry's column family is %s instead of %s", fam,
LogColumnFamily.NAME);
- String qualifier = entry.getKey().getColumnQualifier().toString();
- String[] parts = qualifier.split("/", 2);
- Preconditions.checkArgument(parts.length == 2 && parts[0].equals("-"),
- "Malformed write-ahead log %s", qualifier);
- return fromPath(parts[1]);
+ Text qualifier = entry.getKey().getColumnQualifier();
+ String[] parts = qualifier.toString().split("/", 2);
+ Preconditions.checkArgument(parts.length == 2, "Malformed write-ahead log
%s", qualifier);
+ return validatedLogEntry(parts[1], qualifier);
}
@NonNull
@@ -155,6 +160,10 @@ public final class LogEntry {
*/
@VisibleForTesting
Text getColumnQualifier() {
+ return columnQualifier == null ? newCQ() : new Text(columnQualifier);
+ }
+
+ private Text newCQ() {
return new Text("-/" + getPath());
}
@@ -173,7 +182,7 @@ public final class LogEntry {
* @param mutation the mutation to update
*/
public void addToMutation(Mutation mutation) {
- mutation.put(LogColumnFamily.NAME, getColumnQualifier(), new Value());
+ mutation.put(LogColumnFamily.NAME, newCQ(), new Value());
}
}
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 5273baae45..579571d81d 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
@@ -49,22 +49,23 @@ public class LogEntryTest {
@Test
public void testFromPath() {
var logEntry = LogEntry.fromPath(validPath);
- verifyLogEntry(logEntry);
+ verifyLogEntry(logEntry, new Text("-/" + validPath));
}
@Test
public void testFromMetadata() {
- var logEntry = LogEntry.fromMetaWalEntry(new SimpleImmutableEntry<>(
- new Key("1<", LogColumnFamily.STR_NAME, "-/" + validPath), null));
- verifyLogEntry(logEntry);
+ var columnQualifier = "prefix/" + validPath;
+ var logEntry = LogEntry.fromMetaWalEntry(
+ new SimpleImmutableEntry<>(new Key("1<", LogColumnFamily.STR_NAME,
columnQualifier), null));
+ verifyLogEntry(logEntry, new Text(columnQualifier));
}
// helper for testing build from constructor or from metadata
- private void verifyLogEntry(LogEntry logEntry) {
+ private void verifyLogEntry(LogEntry logEntry, Text expectedColumnQualifier)
{
assertEquals(validPath, logEntry.toString());
assertEquals(validPath, logEntry.getPath());
assertEquals(HostAndPort.fromString(validHost.replace('+', ':')),
logEntry.getTServer());
- assertEquals(new Text("-/" + validPath), logEntry.getColumnQualifier());
+ assertEquals(expectedColumnQualifier, logEntry.getColumnQualifier());
assertEquals(validUUID, logEntry.getUniqueID());
}
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 19841ac7a3..42af4aa47e 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
@@ -87,7 +87,7 @@ public class VolumeUtil {
return null;
}
- protected static LogEntry switchVolumes(LogEntry le, List<Pair<Path,Path>>
replacements) {
+ public static LogEntry switchVolumes(LogEntry le, List<Pair<Path,Path>>
replacements) {
Path switchedPath = switchVolume(new Path(le.getPath()), FileType.WAL,
replacements);
if (switchedPath == null) {
log.trace("Did not switch {}", le);
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/manager/state/MetaDataTableScanner.java
b/server/base/src/main/java/org/apache/accumulo/server/manager/state/MetaDataTableScanner.java
index 88a7af6fe8..b340c19679 100644
---
a/server/base/src/main/java/org/apache/accumulo/server/manager/state/MetaDataTableScanner.java
+++
b/server/base/src/main/java/org/apache/accumulo/server/manager/state/MetaDataTableScanner.java
@@ -22,7 +22,6 @@ import java.io.IOException;
import java.io.UncheckedIOException;
import java.lang.ref.Cleaner.Cleanable;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
@@ -152,7 +151,7 @@ public class MetaDataTableScanner implements
ClosableIterator<TabletLocationStat
Location last = null;
SuspendingTServer suspend = null;
long lastTimestamp = 0;
- List<Collection<String>> walogs = new ArrayList<>();
+ List<LogEntry> walogs = new ArrayList<>();
for (Entry<Key,Value> entry : decodedRow.entrySet()) {
@@ -176,7 +175,7 @@ public class MetaDataTableScanner implements
ClosableIterator<TabletLocationStat
}
current = location;
} else if (cf.compareTo(LogColumnFamily.NAME) == 0) {
-
walogs.add(Collections.singleton(LogEntry.fromMetaWalEntry(entry).getPath()));
+ walogs.add(LogEntry.fromMetaWalEntry(entry));
} else if (cf.compareTo(LastLocationColumnFamily.NAME) == 0) {
if (lastTimestamp < entry.getKey().getTimestamp()) {
last = Location.last(new TServerInstance(entry.getValue(), cq));
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/manager/state/ZooTabletStateStore.java
b/server/base/src/main/java/org/apache/accumulo/server/manager/state/ZooTabletStateStore.java
index 44ddbc4cb4..e18365bdf6 100644
---
a/server/base/src/main/java/org/apache/accumulo/server/manager/state/ZooTabletStateStore.java
+++
b/server/base/src/main/java/org/apache/accumulo/server/manager/state/ZooTabletStateStore.java
@@ -20,7 +20,6 @@ package org.apache.accumulo.server.manager.state;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -96,10 +95,10 @@ class ZooTabletStateStore implements TabletStateStore {
currentSession = loc;
}
- List<Collection<String>> logs = new ArrayList<>();
+ List<LogEntry> logs = new ArrayList<>();
rootMeta.getLogs().forEach(logEntry -> {
- logs.add(Collections.singleton(logEntry.getPath()));
- log.debug("root tablet log {}", logEntry.getPath());
+ logs.add(logEntry);
+ log.debug("root tablet log {}", logEntry);
});
return new TabletLocationState(RootTable.EXTENT, futureSession,
currentSession,
diff --git
a/server/base/src/test/java/org/apache/accumulo/server/manager/state/TabletLocationStateTest.java
b/server/base/src/test/java/org/apache/accumulo/server/manager/state/TabletLocationStateTest.java
index 35df34b291..093317316e 100644
---
a/server/base/src/test/java/org/apache/accumulo/server/manager/state/TabletLocationStateTest.java
+++
b/server/base/src/test/java/org/apache/accumulo/server/manager/state/TabletLocationStateTest.java
@@ -27,28 +27,22 @@ import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
-import java.util.Collection;
import java.util.Set;
+import java.util.UUID;
import org.apache.accumulo.core.dataImpl.KeyExtent;
import org.apache.accumulo.core.metadata.TServerInstance;
import org.apache.accumulo.core.metadata.TabletLocationState;
import org.apache.accumulo.core.metadata.TabletState;
import org.apache.accumulo.core.metadata.schema.TabletMetadata.Location;
+import org.apache.accumulo.core.tabletserver.log.LogEntry;
import org.apache.hadoop.io.Text;
-import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
public class TabletLocationStateTest {
- private static final Collection<String> innerWalogs = new
java.util.HashSet<>();
- private static final Collection<Collection<String>> walogs = new
java.util.HashSet<>();
-
- @BeforeAll
- public static void setUpClass() {
- walogs.add(innerWalogs);
- innerWalogs.add("somelog");
- }
+ private static final Set<LogEntry> walogs =
+ Set.of(LogEntry.fromPath("file:///dir/tserver+9997/" +
UUID.randomUUID()));
private KeyExtent keyExtent;
private Location future;
diff --git
a/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
b/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
index 807ba9f690..d6f3c63851 100644
---
a/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
+++
b/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
@@ -35,6 +35,7 @@ import org.apache.accumulo.core.metadata.TServerInstance;
import org.apache.accumulo.core.metadata.TabletLocationState;
import org.apache.accumulo.core.metadata.TabletState;
import org.apache.accumulo.core.metadata.schema.Ample.DataLevel;
+import org.apache.accumulo.core.tabletserver.log.LogEntry;
import org.apache.accumulo.core.trace.TraceUtil;
import org.apache.accumulo.core.util.Pair;
import org.apache.accumulo.server.ServerContext;
@@ -256,10 +257,6 @@ public class GarbageCollectWriteAheadLogs {
return count;
}
- private UUID path2uuid(Path path) {
- return UUID.fromString(path.getName());
- }
-
private Map<UUID,TServerInstance>
removeEntriesInUse(Map<TServerInstance,Set<UUID>> candidates,
Set<TServerInstance> liveServers, Map<UUID,Pair<WalState,Path>>
logsState,
Map<UUID,Path> recoveryLogs) {
@@ -286,16 +283,14 @@ public class GarbageCollectWriteAheadLogs {
}
// Tablet is being recovered and has WAL references, remove all the WALs
for the dead server
// that made the WALs.
- for (Collection<String> wals : state.walogs) {
- for (String wal : wals) {
- UUID walUUID = path2uuid(new Path(wal));
- TServerInstance dead = result.get(walUUID);
- // There's a reference to a log file, so skip that server's logs
- Set<UUID> idsToIgnore = candidates.remove(dead);
- if (idsToIgnore != null) {
- result.keySet().removeAll(idsToIgnore);
- recoveryLogs.keySet().removeAll(idsToIgnore);
- }
+ for (LogEntry wal : state.walogs) {
+ UUID walUUID = wal.getUniqueID();
+ TServerInstance dead = result.get(walUUID);
+ // There's a reference to a log file, so skip that server's logs
+ Set<UUID> idsToIgnore = candidates.remove(dead);
+ if (idsToIgnore != null) {
+ result.keySet().removeAll(idsToIgnore);
+ recoveryLogs.keySet().removeAll(idsToIgnore);
}
}
}
@@ -355,7 +350,7 @@ public class GarbageCollectWriteAheadLogs {
if (fs.exists(recoveryDir)) {
for (FileStatus status : fs.listStatus(recoveryDir)) {
try {
- UUID logId = path2uuid(status.getPath());
+ UUID logId = UUID.fromString(status.getPath().getName());
result.put(logId, status.getPath());
} catch (IllegalArgumentException iae) {
log.debug("Ignoring file " + status.getPath() + " because it
doesn't look like a uuid");
diff --git
a/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
b/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
index a8157f881f..8f09c4c623 100644
---
a/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
+++
b/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
@@ -30,6 +30,7 @@ import org.apache.accumulo.core.gc.thrift.GcCycleStats;
import org.apache.accumulo.core.metadata.TServerInstance;
import org.apache.accumulo.core.metadata.TabletLocationState;
import org.apache.accumulo.core.metadata.schema.TabletMetadata.Location;
+import org.apache.accumulo.core.tabletserver.log.LogEntry;
import org.apache.accumulo.core.util.Pair;
import org.apache.accumulo.server.ServerContext;
import org.apache.accumulo.server.fs.VolumeManager;
@@ -52,7 +53,7 @@ public class GarbageCollectWriteAheadLogsTest {
Collections.singletonMap(server2, Collections.singletonList(id));
private final Path path = new
Path("hdfs://localhost:9000/accumulo/wal/localhost+1234/" + id);
private final KeyExtent extent = KeyExtent.fromMetaRow(new Text("1<"));
- private final Collection<Collection<String>> walogs =
Collections.emptyList();
+ private final Collection<LogEntry> walogs = Collections.emptyList();
private final TabletLocationState tabletAssignedToServer1;
private final TabletLocationState tabletAssignedToServer2;
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 86bc06930c..530fd6e1c7 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
@@ -38,9 +38,9 @@ import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.dataImpl.KeyExtent;
import org.apache.accumulo.core.fate.zookeeper.ZooCache;
+import org.apache.accumulo.core.tabletserver.log.LogEntry;
import org.apache.accumulo.core.util.threads.ThreadPools;
import org.apache.accumulo.manager.Manager;
-import org.apache.accumulo.server.fs.VolumeManager.FileType;
import org.apache.accumulo.server.fs.VolumeUtil;
import org.apache.accumulo.server.log.SortedLogState;
import org.apache.accumulo.server.manager.recovery.HadoopLogCloser;
@@ -156,72 +156,68 @@ public class RecoveryManager {
}
}
- public boolean recoverLogs(KeyExtent extent, Collection<Collection<String>>
walogs)
- throws IOException {
+ public boolean recoverLogs(KeyExtent extent, Collection<LogEntry> walogs)
throws IOException {
boolean recoveryNeeded = false;
- for (Collection<String> logs : walogs) {
- for (String walog : logs) {
-
- Path switchedWalog = VolumeUtil.switchVolume(new Path(walog),
FileType.WAL,
- manager.getContext().getVolumeReplacements());
- if (switchedWalog != null) {
- // replaces the volume used for sorting, but do not change entry in
metadata table. When
- // the tablet loads it will change the metadata table entry. If
- // the tablet has the same replacement config, then it will find the
sorted log.
- log.info("Volume replaced {} -> {}", walog, switchedWalog);
- walog = switchedWalog.toString();
- }
+ for (LogEntry walog : walogs) {
+
+ LogEntry switchedWalog =
+ VolumeUtil.switchVolumes(walog,
manager.getContext().getVolumeReplacements());
+ if (switchedWalog != null) {
+ // replaces the volume used for sorting, but do not change entry in
metadata table. When
+ // the tablet loads it will change the metadata table entry. If
+ // the tablet has the same replacement config, then it will find the
sorted log.
+ log.info("Volume replaced {} -> {}", walog, switchedWalog);
+ walog = switchedWalog;
+ }
+
+ String sortId = walog.getUniqueID().toString();
+ String filename = walog.getPath();
+ String dest = RecoveryPath.getRecoveryPath(new
Path(filename)).toString();
- String[] parts = walog.split("/");
- String sortId = parts[parts.length - 1];
- String filename = new Path(walog).toString();
- String dest = RecoveryPath.getRecoveryPath(new
Path(filename)).toString();
+ boolean sortQueued;
+ synchronized (this) {
+ sortQueued = sortsQueued.contains(sortId);
+ }
- boolean sortQueued;
+ if (sortQueued
+ && zooCache.get(manager.getZooKeeperRoot() + Constants.ZRECOVERY +
"/" + sortId)
+ == null) {
synchronized (this) {
- sortQueued = sortsQueued.contains(sortId);
+ sortsQueued.remove(sortId);
}
+ }
- if (sortQueued
- && zooCache.get(manager.getZooKeeperRoot() + Constants.ZRECOVERY +
"/" + sortId)
- == null) {
- synchronized (this) {
- sortsQueued.remove(sortId);
- }
+ if (exists(SortedLogState.getFinishedMarkerPath(dest))) {
+ synchronized (this) {
+ closeTasksQueued.remove(sortId);
+ recoveryDelay.remove(sortId);
+ sortsQueued.remove(sortId);
}
+ continue;
+ }
- if (exists(SortedLogState.getFinishedMarkerPath(dest))) {
- synchronized (this) {
- closeTasksQueued.remove(sortId);
- recoveryDelay.remove(sortId);
- sortsQueued.remove(sortId);
+ recoveryNeeded = true;
+ synchronized (this) {
+ if (!closeTasksQueued.contains(sortId) &&
!sortsQueued.contains(sortId)) {
+ AccumuloConfiguration aconf = manager.getConfiguration();
+ LogCloser closer = Property.createInstanceFromPropertyName(aconf,
+ Property.MANAGER_WAL_CLOSER_IMPLEMENTATION, LogCloser.class, new
HadoopLogCloser());
+ Long delay = recoveryDelay.get(sortId);
+ if (delay == null) {
+ delay = aconf.getTimeInMillis(Property.MANAGER_RECOVERY_DELAY);
+ } else {
+ delay = Math.min(2 * delay, 1000 * 60 * 5L);
}
- continue;
- }
- recoveryNeeded = true;
- synchronized (this) {
- if (!closeTasksQueued.contains(sortId) &&
!sortsQueued.contains(sortId)) {
- AccumuloConfiguration aconf = manager.getConfiguration();
- LogCloser closer = Property.createInstanceFromPropertyName(aconf,
- Property.MANAGER_WAL_CLOSER_IMPLEMENTATION, LogCloser.class,
new HadoopLogCloser());
- Long delay = recoveryDelay.get(sortId);
- if (delay == null) {
- delay = aconf.getTimeInMillis(Property.MANAGER_RECOVERY_DELAY);
- } else {
- delay = Math.min(2 * delay, 1000 * 60 * 5L);
- }
-
- log.info("Starting recovery of {} (in : {}s), tablet {} holds a
reference", filename,
- (delay / 1000), extent);
-
- ScheduledFuture<?> future = executor.schedule(
- new LogSortTask(closer, filename, dest, sortId), delay,
TimeUnit.MILLISECONDS);
- ThreadPools.watchNonCriticalScheduledTask(future);
- closeTasksQueued.add(sortId);
- recoveryDelay.put(sortId, delay);
- }
+ log.info("Starting recovery of {} (in : {}s), tablet {} holds a
reference", filename,
+ (delay / 1000), extent);
+
+ ScheduledFuture<?> future = executor.schedule(
+ new LogSortTask(closer, filename, dest, sortId), delay,
TimeUnit.MILLISECONDS);
+ ThreadPools.watchNonCriticalScheduledTask(future);
+ closeTasksQueued.add(sortId);
+ recoveryDelay.put(sortId, delay);
}
}
}
diff --git
a/server/manager/src/test/java/org/apache/accumulo/manager/state/MergeStatsTest.java
b/server/manager/src/test/java/org/apache/accumulo/manager/state/MergeStatsTest.java
index 3e44f42981..b1663add64 100644
---
a/server/manager/src/test/java/org/apache/accumulo/manager/state/MergeStatsTest.java
+++
b/server/manager/src/test/java/org/apache/accumulo/manager/state/MergeStatsTest.java
@@ -24,11 +24,13 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import java.util.Collection;
import java.util.List;
+import java.util.UUID;
import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.dataImpl.KeyExtent;
import org.apache.accumulo.core.metadata.TabletLocationState;
import
org.apache.accumulo.core.metadata.TabletLocationState.BadLocationStateException;
+import org.apache.accumulo.core.tabletserver.log.LogEntry;
import org.apache.accumulo.server.manager.state.MergeInfo;
import org.apache.accumulo.server.manager.state.MergeInfo.Operation;
import org.apache.accumulo.server.manager.state.MergeState;
@@ -60,10 +62,11 @@ public class MergeStatsTest {
// Verify that if there are Walogs the return true, else false
assertTrue(stats.verifyWalogs(getState(keyExtent, List.of())));
- assertFalse(stats.verifyWalogs(getState(keyExtent,
List.of(List.of("log1")))));
+ LogEntry log1 = LogEntry.fromPath("file:///dir/tserver+9997/" +
UUID.randomUUID());
+ assertFalse(stats.verifyWalogs(getState(keyExtent, List.of(log1))));
}
- private TabletLocationState getState(KeyExtent keyExtent,
Collection<Collection<String>> walogs)
+ private TabletLocationState getState(KeyExtent keyExtent,
Collection<LogEntry> walogs)
throws BadLocationStateException {
return new TabletLocationState(keyExtent, null, null, null, null, walogs);
}
diff --git
a/test/src/main/java/org/apache/accumulo/test/manager/MergeStateIT.java
b/test/src/main/java/org/apache/accumulo/test/manager/MergeStateIT.java
index 981a2ff8c5..b6bf21103c 100644
--- a/test/src/main/java/org/apache/accumulo/test/manager/MergeStateIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/manager/MergeStateIT.java
@@ -195,7 +195,7 @@ public class MergeStateIT extends ConfigurableMacBase {
// take it offline
m = TabletColumnFamily.createPrevRowMutation(tablet);
- Collection<Collection<String>> walogs = Collections.emptyList();
+ Collection<LogEntry> walogs = Collections.emptyList();
metaDataStateStore.unassign(Collections.singletonList(new
TabletLocationState(tablet, null,
Location.current(state.someTServer), null, null, walogs)), null);
