Make GCWAL autoclosable, revert back to iterator

Author: DomGarguilo

server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java

@@ -26,6 +26,7 @@
 
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -34,7 +35,7 @@
 import java.util.Map.Entry;
 import java.util.Set;
 import java.util.UUID;
-import java.util.stream.Stream;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.accumulo.core.gc.thrift.GCStatus;
 import org.apache.accumulo.core.gc.thrift.GcCycleStats;
@@ -59,19 +60,22 @@
 import org.slf4j.LoggerFactory;
 
 import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.Streams;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Iterators;
 
 import io.opentelemetry.api.trace.Span;
 import io.opentelemetry.context.Scope;
 
-public class GarbageCollectWriteAheadLogs {
+public class GarbageCollectWriteAheadLogs implements AutoCloseable {
   private static final Logger log = LoggerFactory.getLogger(GarbageCollectWriteAheadLogs.class);
 
   private final ServerContext context;
   private final VolumeManager fs;
   private final LiveTServerSet liveServers;
   private final WalStateManager walMarker;
-  private final Stream<TabletMetadata> store;
+  private final Iterable<TabletMetadata> store;
+  private final List<AutoCloseable> closeables = new ArrayList<>();
+  private final AtomicBoolean hasCollected;
 
   /**
    * Creates a new GC WAL object.
@@ -85,17 +89,20 @@ public class GarbageCollectWriteAheadLogs {
     this.fs = fs;
     this.liveServers = liveServers;
     this.walMarker = new WalStateManager(context);
+
     TabletsMetadata root = context.getAmple().readTablets().forLevel(DataLevel.ROOT)
         .filter(new HasWalsFilter()).fetch(LOCATION, LAST, LOGS, PREV_ROW, SUSPEND).build();
     TabletsMetadata metadata = context.getAmple().readTablets().forLevel(DataLevel.METADATA)
         .filter(new HasWalsFilter()).fetch(LOCATION, LAST, LOGS, PREV_ROW, SUSPEND).build();
     TabletsMetadata user = context.getAmple().readTablets().forLevel(DataLevel.USER)
         .filter(new HasWalsFilter()).fetch(LOCATION, LAST, LOGS, PREV_ROW, SUSPEND).build();
-    this.store = Streams.concat(root.stream(), metadata.stream(), user.stream()).onClose(() -> {
-      root.close();
-      metadata.close();
-      user.close();
-    });
+
+    closeables.add(root);
+    closeables.add(metadata);
+    closeables.add(user);
+
+    this.store = () -> Iterators.concat(root.iterator(), metadata.iterator(), user.iterator());
+    this.hasCollected = new AtomicBoolean(false);
   }
 
   /**
@@ -107,15 +114,17 @@ public class GarbageCollectWriteAheadLogs {
    */
   @VisibleForTesting
   GarbageCollectWriteAheadLogs(ServerContext context, VolumeManager fs,
-      LiveTServerSet liveTServerSet, WalStateManager walMarker, Stream<TabletMetadata> store) {
+      LiveTServerSet liveTServerSet, WalStateManager walMarker, Iterable<TabletMetadata> store) {
     this.context = context;
     this.fs = fs;
     this.liveServers = liveTServerSet;
     this.walMarker = walMarker;
     this.store = store;
+    this.hasCollected = new AtomicBoolean(false);
   }
 
   public void collect(GCStatus status) {
+    Preconditions.checkState(!hasCollected.get(), "Can only call collect once per object");
     try {
       long count;
       long fileScanStop;
@@ -203,7 +212,7 @@ public void collect(GCStatus status) {
       } finally {
         span5.end();
       }
-
+      hasCollected.set(true);
     } catch (Exception e) {
       log.error("exception occurred while garbage collecting write ahead logs", e);
     } finally {
@@ -285,7 +294,7 @@ private Map<UUID,TServerInstance> removeEntriesInUse(Map<TServerInstance,Set<UUI
     }
 
     // remove any entries if there's a log reference (recovery hasn't finished)
-    store.forEach(tabletMetadata -> {
+    for (TabletMetadata tabletMetadata : store) {
       // Tablet is still assigned to a dead server. Manager has moved markers and reassigned it
       // Easiest to just ignore all the WALs for the dead server.
       if (TabletState.compute(tabletMetadata, liveServers) == TabletState.ASSIGNED_TO_DEAD_SERVER) {
@@ -307,8 +316,7 @@ private Map<UUID,TServerInstance> removeEntriesInUse(Map<TServerInstance,Set<UUI
           recoveryLogs.keySet().removeAll(idsToIgnore);
         }
       }
-    });
-    store.close();
+    }
 
     // Remove OPEN and CLOSED logs for live servers: they are still in use
     for (TServerInstance liveServer : liveServers) {
@@ -376,4 +384,15 @@ protected Map<UUID,Path> getSortedWALogs() throws IOException {
     }
     return result;
   }
+
+  @Override
+  public void close() throws Exception {
+    for (AutoCloseable closable : closeables) {
+      try {
+        closable.close();
+      } catch (Exception e) {
+        throw new RuntimeException(e);
+      }
+    }
+  }
 }

server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java

@@ -232,10 +232,11 @@ public void run() {
           // Clean up any unused write-ahead logs
           Span walSpan = TraceUtil.startSpan(this.getClass(), "walogs");
           try (Scope walScope = walSpan.makeCurrent()) {
-            GarbageCollectWriteAheadLogs walogCollector =
-                new GarbageCollectWriteAheadLogs(getContext(), fs, liveTServerSet);
-            log.info("Beginning garbage collection of write-ahead logs");
-            walogCollector.collect(status);
+            try (GarbageCollectWriteAheadLogs walogCollector =
+                new GarbageCollectWriteAheadLogs(getContext(), fs, liveTServerSet)) {
+              log.info("Beginning garbage collection of write-ahead logs");
+              walogCollector.collect(status);
+            }
             gcCycleMetrics.setLastWalCollect(status.lastLog);
           } catch (Exception e) {
             TraceUtil.setException(walSpan, e, false);

server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java

@@ -26,7 +26,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
-import java.util.stream.Stream;
 
 import org.apache.accumulo.core.client.admin.TabletAvailability;
 import org.apache.accumulo.core.dataImpl.KeyExtent;
@@ -75,8 +74,10 @@ public class GarbageCollectWriteAheadLogsTest {
     }
   }
 
-  private final Stream<TabletMetadata> tabletOnServer1List = Stream.of(tabletAssignedToServer1);
-  private final Stream<TabletMetadata> tabletOnServer2List = Stream.of(tabletAssignedToServer2);
+  private final Iterable<TabletMetadata> tabletOnServer1List =
+      Collections.singletonList(tabletAssignedToServer1);
+  private final Iterable<TabletMetadata> tabletOnServer2List =
+      Collections.singletonList(tabletAssignedToServer2);
 
   @Test
   public void testRemoveUnusedLog() throws Exception {
@@ -98,14 +99,15 @@ public void testRemoveUnusedLog() throws Exception {
     marker.removeWalMarker(server1, id);
     EasyMock.expectLastCall().once();
     EasyMock.replay(context, fs, marker, tserverSet);
-    GarbageCollectWriteAheadLogs gc =
+    try (GarbageCollectWriteAheadLogs gc =
         new GarbageCollectWriteAheadLogs(context, fs, tserverSet, marker, tabletOnServer1List) {
           @Override
           protected Map<UUID,Path> getSortedWALogs() {
             return Collections.emptyMap();
           }
-        };
-    gc.collect(status);
+        }) {
+      gc.collect(status);
+    }
     EasyMock.verify(context, fs, marker, tserverSet);
   }
 
@@ -125,14 +127,15 @@ public void testKeepClosedLog() throws Exception {
     EasyMock.expect(marker.getAllMarkers()).andReturn(markers).once();
     EasyMock.expect(marker.state(server1, id)).andReturn(new Pair<>(WalState.CLOSED, path));
     EasyMock.replay(context, marker, tserverSet, fs);
-    GarbageCollectWriteAheadLogs gc =
+    try (GarbageCollectWriteAheadLogs gc =
         new GarbageCollectWriteAheadLogs(context, fs, tserverSet, marker, tabletOnServer1List) {
           @Override
           protected Map<UUID,Path> getSortedWALogs() {
             return Collections.emptyMap();
           }
-        };
-    gc.collect(status);
+        }) {
+      gc.collect(status);
+    }
     EasyMock.verify(context, marker, tserverSet, fs);
   }
 
@@ -159,14 +162,15 @@ public void deleteUnreferenceLogOnDeadServer() throws Exception {
     marker.forget(server2);
     EasyMock.expectLastCall().once();
     EasyMock.replay(context, fs, marker, tserverSet);
-    GarbageCollectWriteAheadLogs gc =
+    try (GarbageCollectWriteAheadLogs gc =
         new GarbageCollectWriteAheadLogs(context, fs, tserverSet, marker, tabletOnServer1List) {
           @Override
           protected Map<UUID,Path> getSortedWALogs() {
             return Collections.emptyMap();
           }
-        };
-    gc.collect(status);
+        }) {
+      gc.collect(status);
+    }
     EasyMock.verify(context, fs, marker, tserverSet);
   }
 
@@ -187,14 +191,15 @@ public void ignoreReferenceLogOnDeadServer() throws Exception {
     EasyMock.expect(marker.state(server2, id)).andReturn(new Pair<>(WalState.OPEN, path));
 
     EasyMock.replay(context, fs, marker, tserverSet);
-    GarbageCollectWriteAheadLogs gc =
+    try (GarbageCollectWriteAheadLogs gc =
         new GarbageCollectWriteAheadLogs(context, fs, tserverSet, marker, tabletOnServer2List) {
           @Override
           protected Map<UUID,Path> getSortedWALogs() {
             return Collections.emptyMap();
           }
-        };
-    gc.collect(status);
+        }) {
+      gc.collect(status);
+    }
     EasyMock.verify(context, fs, marker, tserverSet);
   }