Skip to content

Cleanup recovery code #6261

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ddanielr wants to merge 2 commits into
base: main
Choose a base branch
from
+33 −37
Open

Cleanup recovery code #6261

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d52d9e6
Fix log message formats
ddanielr Mar 27, 2026
934968c
Removes ServerContext from method signature
ddanielr Mar 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 13 additions & 14 deletions server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -222,8 +222,8 @@ protected TabletServer(ServerOpts opts,
super(ServerId.Type.TABLET_SERVER, opts, serverContextFactory, args);
context = super.getContext();
final AccumuloConfiguration aconf = getConfiguration();
log.info("Version " + Constants.VERSION);
log.info("Instance " + getInstanceID());
log.info("Version {}", Constants.VERSION);
log.info("Instance {}", getInstanceID());
this.sessionManager = new SessionManager(context);
this.logSorter = new LogSorter(this);
this.statsKeeper = new TabletStatsKeeper();
Expand Down Expand Up @@ -423,7 +423,7 @@ private HostAndPort getManagerAddress() {
}
return HostAndPort.fromString(managers.iterator().next().toHostPortString());
} catch (Exception e) {
log.warn("Failed to obtain manager host " + e);
log.warn("Failed to obtain manager host", e);
Copy link
Copy Markdown
Contributor

@dlmarion dlmarion Apr 8, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If this prints the entire stack trace, then I think this is a behavior change. Previously this would print Exception.toString, not the entire stack trace. Suggest adding a placeholder to the message {}.

}

return null;
Expand All @@ -438,7 +438,7 @@ private ManagerClientService.Client managerConnection(HostAndPort address) {
// log.info("Listener API to manager has been opened");
return ThriftUtil.getClient(ThriftClientTypes.MANAGER, address, getContext());
} catch (Exception e) {
log.warn("Issue with managerConnection (" + address + ") " + e, e);
log.warn("Issue with managerConnection ({}) {}", address, e, e);
}
return null;
}
Expand Down Expand Up @@ -735,7 +735,7 @@ public TServerInstance getTabletSession() {
try {
return new TServerInstance(getAdvertiseAddress().toString(), lockSessionId);
} catch (Exception ex) {
log.warn("Unable to read session from tablet server lock" + ex);
log.warn("Unable to read session from tablet server lock", ex);
Copy link
Copy Markdown
Contributor

@dlmarion dlmarion Apr 8, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same comment here about behavior change.

return null;
}
}
Expand Down Expand Up @@ -890,15 +890,15 @@ public boolean needsRecovery(TabletMetadata tabletMetadata) {
}

try {
return logger.needsRecovery(getContext(), tabletMetadata.getExtent(), logEntries);
return logger.needsRecovery(tabletMetadata.getExtent(), logEntries);
} catch (IOException e) {
throw new UncheckedIOException(e);
}
}

public void recover(VolumeManager fs, KeyExtent extent, Collection<LogEntry> logEntries,
Set<String> tabletFiles, MutationReceiver mutationReceiver) throws IOException {
logger.recover(getContext(), extent, logEntries, tabletFiles, mutationReceiver);
public void recover(KeyExtent extent, Collection<LogEntry> logEntries, Set<String> tabletFiles,
MutationReceiver mutationReceiver) throws IOException {
logger.recover(extent, logEntries, tabletFiles, mutationReceiver);
}

public int createLogId() {
Expand Down Expand Up @@ -1016,7 +1016,7 @@ private void markUnusedWALs() {
try {
TServerInstance session = this.getTabletSession();
for (DfsLogger candidate : eligible) {
log.info("Marking " + candidate.getPath() + " as unreferenced");
log.info("Marking {} as unreferenced", candidate.getPath());
walMarker.walUnreferenced(session, candidate.getPath());
}
synchronized (closedLogs) {
Expand All @@ -1028,7 +1028,7 @@ private void markUnusedWALs() {
}

public void addNewLogMarker(DfsLogger copy) throws WalMarkerException {
log.info("Writing log marker for " + copy.getPath());
log.info("Writing log marker for {}", copy.getPath());
walMarker.addNewWalMarker(getTabletSession(), copy.getPath());
}

Expand All @@ -1040,7 +1040,7 @@ public void walogClosed(DfsLogger currentLog) throws WalMarkerException {
closedLogs.add(currentLog);
clSize = closedLogs.size();
}
log.info("Marking " + currentLog.getPath() + " as closed. Total closed logs " + clSize);
log.info("Marking {} as closed. Total closed logs {}", currentLog.getPath(), clSize);
walMarker.closeWal(getTabletSession(), currentLog.getPath());

// whenever a new log is added to the set of closed logs, go through all of the tablets and
Expand All @@ -1058,8 +1058,7 @@ public void walogClosed(DfsLogger currentLog) throws WalMarkerException {
}
}
} else {
log.info(
"Marking " + currentLog.getPath() + " as unreferenced (skipping closed writes == 0)");
log.info("Marking {} as unreferenced (skipping closed writes == 0)", currentLog.getPath());
walMarker.walUnreferenced(getTabletSession(), currentLog.getPath());
}
}
Expand Down
18 changes: 8 additions & 10 deletions server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,6 @@
import org.apache.accumulo.core.util.Retry;
import org.apache.accumulo.core.util.Retry.RetryFactory;
import org.apache.accumulo.core.util.threads.Threads;
import org.apache.accumulo.server.ServerContext;
import org.apache.accumulo.server.log.WalStateManager.WalMarkerException;
import org.apache.accumulo.tserver.TabletMutations;
import org.apache.accumulo.tserver.TabletServer;
Expand Down Expand Up @@ -293,7 +292,7 @@ private synchronized void startLogMaker() {
try {
tserver.addNewLogMarker(alog);
} catch (Exception e) {
log.error("Failed to add new WAL marker for " + alog.getLogEntry(), e);
log.error("Failed to add new WAL marker for {}", alog.getLogEntry(), e);

try {
// Intentionally not deleting walog because it may have been advertised in ZK. See
Expand All @@ -311,7 +310,7 @@ private synchronized void startLogMaker() {
try {
tserver.walogClosed(alog);
} catch (WalMarkerException | RuntimeException e2) {
log.error("Failed to close WAL that failed to open: " + alog.getLogEntry(), e2);
log.error("Failed to close WAL that failed to open: {}", alog.getLogEntry(), e2);
}

try {
Expand Down Expand Up @@ -348,7 +347,7 @@ private synchronized void close() throws IOException {
} catch (DfsLogger.LogClosedException ex) {
// ignore
} catch (IOException | RuntimeException ex) {
log.error("Unable to cleanly close log " + currentLog.getLogEntry() + ": " + ex, ex);
log.error("Unable to cleanly close log {}: {}", currentLog.getLogEntry(), ex, ex);
Copy link
Copy Markdown
Contributor

@dlmarion dlmarion Apr 8, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the first ex can just be removed. This is going to log the entire stack trace.

} finally {
try {
this.tserver.walogClosed(currentLog);
Expand Down Expand Up @@ -523,26 +522,25 @@ private CacheProvider createCacheProvider(TabletServerResourceManager resourceMg
LoggingBlockCache.wrap(CacheType.DATA, resourceMgr.getDataCache()));
}

public boolean needsRecovery(ServerContext context, KeyExtent extent, Collection<LogEntry> walogs)
throws IOException {
public boolean needsRecovery(KeyExtent extent, Collection<LogEntry> walogs) throws IOException {
try {
var resourceMgr = tserver.getResourceManager();
var cacheProvider = createCacheProvider(resourceMgr);
SortedLogRecovery recovery =
new SortedLogRecovery(context, resourceMgr.getFileLenCache(), cacheProvider);
new SortedLogRecovery(tserver.getContext(), resourceMgr.getFileLenCache(), cacheProvider);
return recovery.needsRecovery(extent, resolve(walogs));
} catch (Exception e) {
throw new IOException(e);
}
}

public void recover(ServerContext context, KeyExtent extent, Collection<LogEntry> walogs,
Set<String> tabletFiles, MutationReceiver mr) throws IOException {
public void recover(KeyExtent extent, Collection<LogEntry> walogs, Set<String> tabletFiles,
MutationReceiver mr) throws IOException {
try {
var resourceMgr = tserver.getResourceManager();
var cacheProvider = createCacheProvider(resourceMgr);
SortedLogRecovery recovery =
new SortedLogRecovery(context, resourceMgr.getFileLenCache(), cacheProvider);
new SortedLogRecovery(tserver.getContext(), resourceMgr.getFileLenCache(), cacheProvider);
recovery.recover(extent, resolve(walogs), tabletFiles, mr);
} catch (Exception e) {
throw new IOException(e);
Expand Down
25 changes: 12 additions & 13 deletions server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -279,19 +279,18 @@ public Tablet(final TabletServer tabletServer, final KeyExtent extent,
absPaths.add(ref.getNormalizedPathStr());
}

tabletServer.recover(this.getTabletServer().getVolumeManager(), extent, logEntries,
absPaths, m -> {
Collection<ColumnUpdate> muts = m.getUpdates();
for (ColumnUpdate columnUpdate : muts) {
if (!columnUpdate.hasTimestamp()) {
// if it is not a user set timestamp, it must have been set
// by the system
maxTime.set(Math.max(maxTime.get(), columnUpdate.getTimestamp()));
}
}
getTabletMemory().mutate(commitSession, Collections.singletonList(m), 1);
entriesUsedOnTablet.incrementAndGet();
});
tabletServer.recover(extent, logEntries, absPaths, m -> {
Collection<ColumnUpdate> muts = m.getUpdates();
for (ColumnUpdate columnUpdate : muts) {
if (!columnUpdate.hasTimestamp()) {
// if it is not a user set timestamp, it must have been set
// by the system
maxTime.set(Math.max(maxTime.get(), columnUpdate.getTimestamp()));
}
}
getTabletMemory().mutate(commitSession, Collections.singletonList(m), 1);
entriesUsedOnTablet.incrementAndGet();
});

if (maxTime.get() != Long.MIN_VALUE) {
tabletTime.updateTimeIfGreater(maxTime.get());
Expand Down