@Test public void ignoreReferenceLogOnDeadServer() throws Exception { ServerContext context = EasyMock.createMock(ServerContext.class); VolumeManager fs = EasyMock.createMock(VolumeManager.class); WalStateManager marker = EasyMock.createMock(WalStateManager.class); LiveTServerSet tserverSet = EasyMock.createMock(LiveTServerSet.class); Scanner mscanner = EasyMock.createMock(Scanner.class); Scanner rscanner = EasyMock.createMock(Scanner.class); GCStatus status = new GCStatus(null, null, null, new GcCycleStats()); tserverSet.scanServers(); EasyMock.expectLastCall(); EasyMock.expect(tserverSet.getCurrentServers()).andReturn(Collections.singleton(server1)); EasyMock.expect(marker.getAllMarkers()).andReturn(markers2).once(); EasyMock.expect(marker.state(server2, id)).andReturn(new Pair<>(WalState.OPEN, path)); EasyMock.expect(context.createScanner(ReplicationTable.NAME, Authorizations.EMPTY)) .andReturn(rscanner); rscanner.fetchColumnFamily(ReplicationSchema.StatusSection.NAME); EasyMock.expectLastCall().once(); EasyMock.expect(rscanner.iterator()).andReturn(emptyKV); EasyMock.expect(context.createScanner(MetadataTable.NAME, Authorizations.EMPTY)) .andReturn(mscanner); mscanner.fetchColumnFamily(ReplicationSection.COLF); EasyMock.expectLastCall().once(); mscanner.setRange(ReplicationSection.getRange()); EasyMock.expectLastCall().once(); EasyMock.expect(mscanner.iterator()).andReturn(emptyKV); EasyMock.replay(context, fs, marker, tserverSet, rscanner, mscanner); GarbageCollectWriteAheadLogs gc = new GarbageCollectWriteAheadLogs(context, fs, false, tserverSet, marker, tabletOnServer2List) { @Override protected Map<UUID,Path> getSortedWALogs() { return Collections.emptyMap(); } }; gc.collect(status); EasyMock.verify(context, fs, marker, tserverSet, rscanner, mscanner); }
public void collect(GCStatus status) { try { long count; long fileScanStop; Map<TServerInstance,Set<UUID>> logsByServer; Map<UUID,Pair<WalState,Path>> logsState; Map<UUID,Path> recoveryLogs; try (TraceScope span = Trace.startSpan("getCandidates")) { status.currentLog.started = System.currentTimeMillis(); recoveryLogs = getSortedWALogs(); logsByServer = new HashMap<>(); logsState = new HashMap<>(); count = getCurrent(logsByServer, logsState); fileScanStop = System.currentTimeMillis(); log.info(String.format("Fetched %d files for %d servers in %.2f seconds", count, logsByServer.size(), (fileScanStop - status.currentLog.started) / 1000.)); status.currentLog.candidates = count; } liveServers.scanServers(); Set<TServerInstance> currentServers = liveServers.getCurrentServers(); Map<UUID,TServerInstance> uuidToTServer; try (TraceScope span = Trace.startSpan("removeEntriesInUse")) { uuidToTServer = removeEntriesInUse(logsByServer, currentServers, logsState, recoveryLogs); count = uuidToTServer.size(); } catch (Exception ex) { log.error("Unable to scan metadata table", ex); return; } long logEntryScanStop = System.currentTimeMillis(); log.info(String.format("%d log entries scanned in %.2f seconds", count, (logEntryScanStop - fileScanStop) / 1000.)); try (TraceScope span = Trace.startSpan("removeReplicationEntries")) { count = removeReplicationEntries(uuidToTServer); } catch (Exception ex) { log.error("Unable to scan replication table", ex); return; } long replicationEntryScanStop = System.currentTimeMillis(); log.info(String.format("%d replication entries scanned in %.2f seconds", count, (replicationEntryScanStop - logEntryScanStop) / 1000.)); long removeStop; try (TraceScope span = Trace.startSpan("removeFiles")) { logsState.keySet().retainAll(uuidToTServer.keySet()); count = removeFiles(logsState.values(), status); removeStop = System.currentTimeMillis(); log.info(String.format("%d total logs removed from %d servers in %.2f seconds", count, logsByServer.size(), (removeStop - logEntryScanStop) / 1000.)); count = removeFiles(recoveryLogs.values()); log.info("{} recovery logs removed", count); } try (TraceScope span = Trace.startSpan("removeMarkers")) { count = removeTabletServerMarkers(uuidToTServer, logsByServer, currentServers); long removeMarkersStop = System.currentTimeMillis(); log.info(String.format("%d markers removed in %.2f seconds", count, (removeMarkersStop - removeStop) / 1000.)); } status.currentLog.finished = removeStop; status.lastLog = status.currentLog; status.currentLog = new GcCycleStats(); } catch (Exception e) { log.error("exception occured while garbage collecting write ahead logs", e); } }
GarbageCollectWriteAheadLogs { public void collect(GCStatus status) { try { long count; long fileScanStop; Map<TServerInstance,Set<UUID>> logsByServer; Map<UUID,Pair<WalState,Path>> logsState; Map<UUID,Path> recoveryLogs; try (TraceScope span = Trace.startSpan("getCandidates")) { status.currentLog.started = System.currentTimeMillis(); recoveryLogs = getSortedWALogs(); logsByServer = new HashMap<>(); logsState = new HashMap<>(); count = getCurrent(logsByServer, logsState); fileScanStop = System.currentTimeMillis(); log.info(String.format("Fetched %d files for %d servers in %.2f seconds", count, logsByServer.size(), (fileScanStop - status.currentLog.started) / 1000.)); status.currentLog.candidates = count; } liveServers.scanServers(); Set<TServerInstance> currentServers = liveServers.getCurrentServers(); Map<UUID,TServerInstance> uuidToTServer; try (TraceScope span = Trace.startSpan("removeEntriesInUse")) { uuidToTServer = removeEntriesInUse(logsByServer, currentServers, logsState, recoveryLogs); count = uuidToTServer.size(); } catch (Exception ex) { log.error("Unable to scan metadata table", ex); return; } long logEntryScanStop = System.currentTimeMillis(); log.info(String.format("%d log entries scanned in %.2f seconds", count, (logEntryScanStop - fileScanStop) / 1000.)); try (TraceScope span = Trace.startSpan("removeReplicationEntries")) { count = removeReplicationEntries(uuidToTServer); } catch (Exception ex) { log.error("Unable to scan replication table", ex); return; } long replicationEntryScanStop = System.currentTimeMillis(); log.info(String.format("%d replication entries scanned in %.2f seconds", count, (replicationEntryScanStop - logEntryScanStop) / 1000.)); long removeStop; try (TraceScope span = Trace.startSpan("removeFiles")) { logsState.keySet().retainAll(uuidToTServer.keySet()); count = removeFiles(logsState.values(), status); removeStop = System.currentTimeMillis(); log.info(String.format("%d total logs removed from %d servers in %.2f seconds", count, logsByServer.size(), (removeStop - logEntryScanStop) / 1000.)); count = removeFiles(recoveryLogs.values()); log.info("{} recovery logs removed", count); } try (TraceScope span = Trace.startSpan("removeMarkers")) { count = removeTabletServerMarkers(uuidToTServer, logsByServer, currentServers); long removeMarkersStop = System.currentTimeMillis(); log.info(String.format("%d markers removed in %.2f seconds", count, (removeMarkersStop - removeStop) / 1000.)); } status.currentLog.finished = removeStop; status.lastLog = status.currentLog; status.currentLog = new GcCycleStats(); } catch (Exception e) { log.error("exception occured while garbage collecting write ahead logs", e); } } }
GarbageCollectWriteAheadLogs { public void collect(GCStatus status) { try { long count; long fileScanStop; Map<TServerInstance,Set<UUID>> logsByServer; Map<UUID,Pair<WalState,Path>> logsState; Map<UUID,Path> recoveryLogs; try (TraceScope span = Trace.startSpan("getCandidates")) { status.currentLog.started = System.currentTimeMillis(); recoveryLogs = getSortedWALogs(); logsByServer = new HashMap<>(); logsState = new HashMap<>(); count = getCurrent(logsByServer, logsState); fileScanStop = System.currentTimeMillis(); log.info(String.format("Fetched %d files for %d servers in %.2f seconds", count, logsByServer.size(), (fileScanStop - status.currentLog.started) / 1000.)); status.currentLog.candidates = count; } liveServers.scanServers(); Set<TServerInstance> currentServers = liveServers.getCurrentServers(); Map<UUID,TServerInstance> uuidToTServer; try (TraceScope span = Trace.startSpan("removeEntriesInUse")) { uuidToTServer = removeEntriesInUse(logsByServer, currentServers, logsState, recoveryLogs); count = uuidToTServer.size(); } catch (Exception ex) { log.error("Unable to scan metadata table", ex); return; } long logEntryScanStop = System.currentTimeMillis(); log.info(String.format("%d log entries scanned in %.2f seconds", count, (logEntryScanStop - fileScanStop) / 1000.)); try (TraceScope span = Trace.startSpan("removeReplicationEntries")) { count = removeReplicationEntries(uuidToTServer); } catch (Exception ex) { log.error("Unable to scan replication table", ex); return; } long replicationEntryScanStop = System.currentTimeMillis(); log.info(String.format("%d replication entries scanned in %.2f seconds", count, (replicationEntryScanStop - logEntryScanStop) / 1000.)); long removeStop; try (TraceScope span = Trace.startSpan("removeFiles")) { logsState.keySet().retainAll(uuidToTServer.keySet()); count = removeFiles(logsState.values(), status); removeStop = System.currentTimeMillis(); log.info(String.format("%d total logs removed from %d servers in %.2f seconds", count, logsByServer.size(), (removeStop - logEntryScanStop) / 1000.)); count = removeFiles(recoveryLogs.values()); log.info("{} recovery logs removed", count); } try (TraceScope span = Trace.startSpan("removeMarkers")) { count = removeTabletServerMarkers(uuidToTServer, logsByServer, currentServers); long removeMarkersStop = System.currentTimeMillis(); log.info(String.format("%d markers removed in %.2f seconds", count, (removeMarkersStop - removeStop) / 1000.)); } status.currentLog.finished = removeStop; status.lastLog = status.currentLog; status.currentLog = new GcCycleStats(); } catch (Exception e) { log.error("exception occured while garbage collecting write ahead logs", e); } } GarbageCollectWriteAheadLogs(final ServerContext context, final VolumeManager fs,
      final LiveTServerSet liveServers, boolean useTrash); @VisibleForTesting  GarbageCollectWriteAheadLogs(ServerContext context, VolumeManager fs, boolean useTrash,
      LiveTServerSet liveTServerSet, WalStateManager walMarker,
      Iterable<TabletLocationState> store); }
GarbageCollectWriteAheadLogs { public void collect(GCStatus status) { try { long count; long fileScanStop; Map<TServerInstance,Set<UUID>> logsByServer; Map<UUID,Pair<WalState,Path>> logsState; Map<UUID,Path> recoveryLogs; try (TraceScope span = Trace.startSpan("getCandidates")) { status.currentLog.started = System.currentTimeMillis(); recoveryLogs = getSortedWALogs(); logsByServer = new HashMap<>(); logsState = new HashMap<>(); count = getCurrent(logsByServer, logsState); fileScanStop = System.currentTimeMillis(); log.info(String.format("Fetched %d files for %d servers in %.2f seconds", count, logsByServer.size(), (fileScanStop - status.currentLog.started) / 1000.)); status.currentLog.candidates = count; } liveServers.scanServers(); Set<TServerInstance> currentServers = liveServers.getCurrentServers(); Map<UUID,TServerInstance> uuidToTServer; try (TraceScope span = Trace.startSpan("removeEntriesInUse")) { uuidToTServer = removeEntriesInUse(logsByServer, currentServers, logsState, recoveryLogs); count = uuidToTServer.size(); } catch (Exception ex) { log.error("Unable to scan metadata table", ex); return; } long logEntryScanStop = System.currentTimeMillis(); log.info(String.format("%d log entries scanned in %.2f seconds", count, (logEntryScanStop - fileScanStop) / 1000.)); try (TraceScope span = Trace.startSpan("removeReplicationEntries")) { count = removeReplicationEntries(uuidToTServer); } catch (Exception ex) { log.error("Unable to scan replication table", ex); return; } long replicationEntryScanStop = System.currentTimeMillis(); log.info(String.format("%d replication entries scanned in %.2f seconds", count, (replicationEntryScanStop - logEntryScanStop) / 1000.)); long removeStop; try (TraceScope span = Trace.startSpan("removeFiles")) { logsState.keySet().retainAll(uuidToTServer.keySet()); count = removeFiles(logsState.values(), status); removeStop = System.currentTimeMillis(); log.info(String.format("%d total logs removed from %d servers in %.2f seconds", count, logsByServer.size(), (removeStop - logEntryScanStop) / 1000.)); count = removeFiles(recoveryLogs.values()); log.info("{} recovery logs removed", count); } try (TraceScope span = Trace.startSpan("removeMarkers")) { count = removeTabletServerMarkers(uuidToTServer, logsByServer, currentServers); long removeMarkersStop = System.currentTimeMillis(); log.info(String.format("%d markers removed in %.2f seconds", count, (removeMarkersStop - removeStop) / 1000.)); } status.currentLog.finished = removeStop; status.lastLog = status.currentLog; status.currentLog = new GcCycleStats(); } catch (Exception e) { log.error("exception occured while garbage collecting write ahead logs", e); } } GarbageCollectWriteAheadLogs(final ServerContext context, final VolumeManager fs,
      final LiveTServerSet liveServers, boolean useTrash); @VisibleForTesting  GarbageCollectWriteAheadLogs(ServerContext context, VolumeManager fs, boolean useTrash,
      LiveTServerSet liveTServerSet, WalStateManager walMarker,
      Iterable<TabletLocationState> store); void collect(GCStatus status); }
GarbageCollectWriteAheadLogs { public void collect(GCStatus status) { try { long count; long fileScanStop; Map<TServerInstance,Set<UUID>> logsByServer; Map<UUID,Pair<WalState,Path>> logsState; Map<UUID,Path> recoveryLogs; try (TraceScope span = Trace.startSpan("getCandidates")) { status.currentLog.started = System.currentTimeMillis(); recoveryLogs = getSortedWALogs(); logsByServer = new HashMap<>(); logsState = new HashMap<>(); count = getCurrent(logsByServer, logsState); fileScanStop = System.currentTimeMillis(); log.info(String.format("Fetched %d files for %d servers in %.2f seconds", count, logsByServer.size(), (fileScanStop - status.currentLog.started) / 1000.)); status.currentLog.candidates = count; } liveServers.scanServers(); Set<TServerInstance> currentServers = liveServers.getCurrentServers(); Map<UUID,TServerInstance> uuidToTServer; try (TraceScope span = Trace.startSpan("removeEntriesInUse")) { uuidToTServer = removeEntriesInUse(logsByServer, currentServers, logsState, recoveryLogs); count = uuidToTServer.size(); } catch (Exception ex) { log.error("Unable to scan metadata table", ex); return; } long logEntryScanStop = System.currentTimeMillis(); log.info(String.format("%d log entries scanned in %.2f seconds", count, (logEntryScanStop - fileScanStop) / 1000.)); try (TraceScope span = Trace.startSpan("removeReplicationEntries")) { count = removeReplicationEntries(uuidToTServer); } catch (Exception ex) { log.error("Unable to scan replication table", ex); return; } long replicationEntryScanStop = System.currentTimeMillis(); log.info(String.format("%d replication entries scanned in %.2f seconds", count, (replicationEntryScanStop - logEntryScanStop) / 1000.)); long removeStop; try (TraceScope span = Trace.startSpan("removeFiles")) { logsState.keySet().retainAll(uuidToTServer.keySet()); count = removeFiles(logsState.values(), status); removeStop = System.currentTimeMillis(); log.info(String.format("%d total logs removed from %d servers in %.2f seconds", count, logsByServer.size(), (removeStop - logEntryScanStop) / 1000.)); count = removeFiles(recoveryLogs.values()); log.info("{} recovery logs removed", count); } try (TraceScope span = Trace.startSpan("removeMarkers")) { count = removeTabletServerMarkers(uuidToTServer, logsByServer, currentServers); long removeMarkersStop = System.currentTimeMillis(); log.info(String.format("%d markers removed in %.2f seconds", count, (removeMarkersStop - removeStop) / 1000.)); } status.currentLog.finished = removeStop; status.lastLog = status.currentLog; status.currentLog = new GcCycleStats(); } catch (Exception e) { log.error("exception occured while garbage collecting write ahead logs", e); } } GarbageCollectWriteAheadLogs(final ServerContext context, final VolumeManager fs,
      final LiveTServerSet liveServers, boolean useTrash); @VisibleForTesting  GarbageCollectWriteAheadLogs(ServerContext context, VolumeManager fs, boolean useTrash,
      LiveTServerSet liveTServerSet, WalStateManager walMarker,
      Iterable<TabletLocationState> store); void collect(GCStatus status);  }