diff --git a/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/master/BackupLogCleaner.java b/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/master/BackupLogCleaner.java index 0eac3d617b00..229ac7f6de53 100644 --- a/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/master/BackupLogCleaner.java +++ b/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/master/BackupLogCleaner.java @@ -23,6 +23,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.stream.Collectors; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.Path; @@ -80,9 +81,16 @@ public void init(Map params) { } } - private Map getServersToOldestBackupMapping(List backups) + private Map getServerToNewestBackupTs(List backups) throws IOException { - Map serverAddressToLastBackupMap = new HashMap<>(); + if (LOG.isDebugEnabled()) { + LOG.debug( + "Cleaning WALs if they are older than the newest backups. " + + "Checking WALs against {} backups: {}", + backups.size(), + backups.stream().map(BackupInfo::getBackupId).sorted().collect(Collectors.joining(", "))); + } + Map serverAddressToNewestBackupMap = new HashMap<>(); Map tableNameBackupInfoMap = new HashMap<>(); for (BackupInfo backupInfo : backups) { @@ -92,13 +100,20 @@ private Map getServersToOldestBackupMapping(List back tableNameBackupInfoMap.put(table, backupInfo.getStartTs()); for (Map.Entry entry : backupInfo.getTableSetTimestampMap().get(table) .entrySet()) { - serverAddressToLastBackupMap.put(Address.fromString(entry.getKey()), entry.getValue()); + serverAddressToNewestBackupMap.put(Address.fromString(entry.getKey()), + entry.getValue()); } } } } - return serverAddressToLastBackupMap; + if (LOG.isDebugEnabled()) { + for (Map.Entry entry : serverAddressToNewestBackupMap.entrySet()) { + LOG.debug("Server: {}, Newest Backup: {}", entry.getKey().getHostName(), entry.getValue()); + } + } + + return serverAddressToNewestBackupMap; } @Override @@ -113,11 +128,10 @@ public Iterable getDeletableFiles(Iterable files) { return files; } - Map addressToLastBackupMap; + Map addressToNewestBackupMap; try { try (BackupManager backupManager = new BackupManager(conn, getConf())) { - addressToLastBackupMap = - getServersToOldestBackupMapping(backupManager.getBackupHistory(true)); + addressToNewestBackupMap = getServerToNewestBackupTs(backupManager.getBackupHistory(true)); } } catch (IOException ex) { LOG.error("Failed to analyse backup history with exception: {}. Retaining all logs", @@ -125,7 +139,7 @@ public Iterable getDeletableFiles(Iterable files) { return Collections.emptyList(); } for (FileStatus file : files) { - if (canDeleteFile(addressToLastBackupMap, file.getPath())) { + if (canDeleteFile(addressToNewestBackupMap, file.getPath())) { filteredFiles.add(file); } } @@ -160,7 +174,7 @@ public boolean isStopped() { return this.stopped; } - protected static boolean canDeleteFile(Map addressToLastBackupMap, Path path) { + protected static boolean canDeleteFile(Map addressToNewestBackupMap, Path path) { if (isHMasterWAL(path)) { return true; } @@ -176,12 +190,29 @@ protected static boolean canDeleteFile(Map addressToLastBackupMap Address walServerAddress = Address.fromString(hostname); long walTimestamp = WAL.getTimestamp(path.getName()); - if ( - !addressToLastBackupMap.containsKey(walServerAddress) - || addressToLastBackupMap.get(walServerAddress) >= walTimestamp - ) { + if (!addressToNewestBackupMap.containsKey(walServerAddress)) { + if (LOG.isDebugEnabled()) { + LOG.debug("No backup found for server: {}. Deleting file: {}", + walServerAddress.getHostName(), path); + } return true; } + + Long lastBackupTs = addressToNewestBackupMap.get(walServerAddress); + if (lastBackupTs >= walTimestamp) { + if (LOG.isDebugEnabled()) { + LOG.debug( + "Backup found for server: {}. Backup from {} is newer than file, so deleting: {}", + walServerAddress.getHostName(), lastBackupTs, path); + } + return true; + } + + if (LOG.isDebugEnabled()) { + LOG.debug( + "Backup found for server: {}. Backup from {} is older than the file, so keeping: {}", + walServerAddress.getHostName(), lastBackupTs, path); + } } catch (Exception ex) { LOG.warn("Error occurred while filtering file: {}. Ignoring cleanup of this log", path, ex); return false;