Merge remote-tracking branch 'upstream/2.1' into testMaxWalReferenced

Author: DomGarguilo

core/src/main/java/org/apache/accumulo/core/fate/zookeeper/ServiceLock.java

@@ -18,7 +18,6 @@
  */
 package org.apache.accumulo.core.fate.zookeeper;
 
-import static java.nio.charset.StandardCharsets.UTF_8;
 import static java.util.Objects.requireNonNull;
 
 import java.util.ArrayList;
@@ -736,33 +735,6 @@ public static void deleteLock(ZooReaderWriter zk, ServiceLockPath path)
 
   }
 
-  public static boolean deleteLock(ZooReaderWriter zk, ServiceLockPath path, String lockData)
-      throws InterruptedException, KeeperException {
-
-    List<String> children = validateAndSort(path, zk.getChildren(path.toString()));
-
-    if (children.isEmpty()) {
-      throw new IllegalStateException("No lock is held at " + path);
-    }
-
-    String lockNode = children.get(0);
-
-    if (!lockNode.startsWith(ZLOCK_PREFIX)) {
-      throw new RuntimeException("Node " + lockNode + " at " + path + " is not a lock node");
-    }
-
-    byte[] data = zk.getData(path + "/" + lockNode);
-
-    if (lockData.equals(new String(data, UTF_8))) {
-      String pathToDelete = path + "/" + lockNode;
-      LOG.debug("Deleting all at path {} due to lock deletion", pathToDelete);
-      zk.recursiveDelete(pathToDelete, NodeMissingPolicy.FAIL);
-      return true;
-    }
-
-    return false;
-  }
-
   /**
    * Checks that the lock still exists in ZooKeeper. The typical mechanism for determining if a lock
    * is lost depends on a Watcher set on the lock node. There exists a case where the Watcher may

core/src/main/java/org/apache/accumulo/core/fate/zookeeper/ZooCache.java

@@ -30,6 +30,7 @@
 import java.util.concurrent.locks.LockSupport;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Predicate;
 
 import org.apache.accumulo.core.fate.zookeeper.ServiceLock.ServiceLockPath;
 import org.apache.zookeeper.KeeperException;
@@ -525,6 +526,33 @@ public void clear(String zPath) {
     }
   }
 
+  /**
+   * Removes all paths in the cache match the predicate.
+   */
+  public void clear(Predicate<String> pathPredicate) {
+    Preconditions.checkState(!closed);
+    Predicate<String> pathPredicateToUse;
+    if (log.isTraceEnabled()) {
+      pathPredicateToUse = pathPredicate.and(path -> {
+        log.trace("removing {} from cache", path);
+        return true;
+      });
+    } else {
+      pathPredicateToUse = pathPredicate;
+    }
+    cacheWriteLock.lock();
+    try {
+      cache.keySet().removeIf(pathPredicateToUse);
+      childrenCache.keySet().removeIf(pathPredicateToUse);
+      statCache.keySet().removeIf(pathPredicateToUse);
+
+      immutableCache = new ImmutableCacheCopies(++updateCount, cache, statCache, childrenCache);
+
+    } finally {
+      cacheWriteLock.unlock();
+    }
+  }
+
   public byte[] getLockData(ServiceLockPath path) {
     List<String> children = ServiceLock.validateAndSort(path, getChildren(path.toString()));
     if (children == null || children.isEmpty()) {

core/src/main/java/org/apache/accumulo/core/file/rfile/RFile.java

@@ -1288,24 +1288,32 @@ public Reader(CachableBlockFile.CachableBuilder b) throws IOException {
       this(new CachableBlockFile.Reader(b));
     }
 
-    private void closeLocalityGroupReaders() {
+    private void closeLocalityGroupReaders(boolean ignoreIOExceptions) throws IOException {
       for (LocalityGroupReader lgr : currentReaders) {
         try {
           lgr.close();
         } catch (IOException e) {
-          log.warn("Errored out attempting to close LocalityGroupReader.", e);
+          if (ignoreIOExceptions) {
+            log.warn("Errored out attempting to close LocalityGroupReader.", e);
+          } else {
+            throw e;
+          }
         }
       }
     }
 
     @Override
-    public void closeDeepCopies() {
+    public void closeDeepCopies() throws IOException {
+      closeDeepCopies(false);
+    }
+
+    private void closeDeepCopies(boolean ignoreIOExceptions) throws IOException {
       if (deepCopy) {
         throw new RuntimeException("Calling closeDeepCopies on a deep copy is not supported");
       }
 
       for (Reader deepCopy : deepCopies) {
-        deepCopy.closeLocalityGroupReaders();
+        deepCopy.closeLocalityGroupReaders(ignoreIOExceptions);
       }
 
       deepCopies.clear();
@@ -1317,8 +1325,9 @@ public void close() throws IOException {
         throw new RuntimeException("Calling close on a deep copy is not supported");
       }
 
-      closeDeepCopies();
-      closeLocalityGroupReaders();
+      // Closes as much as possible igoring and logging exceptions along the way
+      closeDeepCopies(true);
+      closeLocalityGroupReaders(true);
 
       if (sampleReaders != null) {
         for (LocalityGroupReader lgr : sampleReaders) {

core/src/main/java/org/apache/accumulo/core/util/tables/TableZooHelper.java

@@ -179,7 +179,7 @@ public TableState getTableState(TableId tableId, boolean clearCachedState) {
     String statePath = context.getZooKeeperRoot() + Constants.ZTABLES + "/" + tableId.canonical()
         + Constants.ZTABLE_STATE;
     if (clearCachedState) {
-      context.getZooCache().clear(context.getZooKeeperRoot() + statePath);
+      context.getZooCache().clear(statePath);
       instanceToMapCache.invalidateAll();
     }
     ZooCache zc = context.getZooCache();

core/src/test/java/org/apache/accumulo/core/cli/TestHelp.java

@@ -19,9 +19,12 @@
 package org.apache.accumulo.core.cli;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import org.junit.jupiter.api.Test;
 
+import com.beust.jcommander.Parameter;
+
 public class TestHelp {
   protected class HelpStub extends Help {
     @Override
@@ -46,4 +49,20 @@ public void testInvalidArgs() {
     }
   }
 
+  @Test
+  public void testHelpCommand() {
+    class TestHelpOpt extends HelpStub {
+      @Parameter(names = {"--test"})
+      boolean test = false;
+    }
+
+    String[] args = {"--help", "--test"};
+    TestHelpOpt opts = new TestHelpOpt();
+    try {
+      opts.parseArgs("program", args);
+      assertTrue(opts.test);
+    } catch (RuntimeException e) {
+      assertEquals("0", e.getMessage());
+    }
+  }
 }

minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterControl.java

@@ -47,6 +47,7 @@
 import org.apache.accumulo.minicluster.ServerType;
 import org.apache.accumulo.miniclusterImpl.MiniAccumuloClusterImpl.ProcessInfo;
 import org.apache.accumulo.server.util.Admin;
+import org.apache.accumulo.server.util.ZooZap;
 import org.apache.accumulo.tserver.ScanServer;
 import org.apache.thrift.TException;
 import org.apache.thrift.transport.TTransportException;
@@ -278,6 +279,11 @@ public synchronized void stop(ServerType server, String hostname) throws IOExcep
         if (managerProcess != null) {
           try {
             cluster.stopProcessWithTimeout(managerProcess, 30, TimeUnit.SECONDS);
+            try {
+              new ZooZap().zap(cluster.getServerContext().getSiteConfiguration(), "-manager");
+            } catch (RuntimeException e) {
+              log.error("Error zapping Manager zookeeper lock", e);
+            }
           } catch (ExecutionException | TimeoutException e) {
             log.warn("Manager did not fully stop after 30 seconds", e);
           } catch (InterruptedException e) {

minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterImpl.java

@@ -56,6 +56,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
+import java.util.function.Predicate;
 import java.util.function.Supplier;
 import java.util.stream.Stream;
 
@@ -90,6 +91,7 @@
 import org.apache.accumulo.server.init.Initialize;
 import org.apache.accumulo.server.util.AccumuloStatus;
 import org.apache.accumulo.server.util.PortUtils;
+import org.apache.accumulo.server.util.ZooZap;
 import org.apache.accumulo.start.Main;
 import org.apache.accumulo.start.classloader.vfs.MiniDFSUtil;
 import org.apache.accumulo.start.spi.KeywordExecutable;
@@ -824,9 +826,47 @@ public synchronized void stop() throws IOException, InterruptedException {
 
     control.stop(ServerType.GARBAGE_COLLECTOR, null);
     control.stop(ServerType.MANAGER, null);
+    control.stop(ServerType.COMPACTION_COORDINATOR);
     control.stop(ServerType.TABLET_SERVER, null);
+    control.stop(ServerType.COMPACTOR, null);
+    control.stop(ServerType.SCAN_SERVER, null);
+
+    // The method calls above kill the server
+    // Clean up the locks in ZooKeeper fo that if the cluster
+    // is restarted, then the processes will start right away
+    // and not wait for the old locks to be cleaned up.
+    try {
+      new ZooZap().zap(getServerContext().getSiteConfiguration(), "-manager",
+          "-compaction-coordinators", "-tservers", "-compactors", "-sservers");
+    } catch (RuntimeException e) {
+      log.error("Error zapping zookeeper locks", e);
+    }
     control.stop(ServerType.ZOOKEEPER, null);
 
+    // Clear the location of the servers in ZooCache.
+    // When ZooKeeper was stopped in the previous method call,
+    // the local ZooKeeper watcher did not fire. If MAC is
+    // restarted, then ZooKeeper will start on the same port with
+    // the same data, but no Watchers will fire.
+    boolean startCalled = true;
+    try {
+      getServerContext();
+    } catch (RuntimeException e) {
+      if (e.getMessage().startsWith("Accumulo not initialized")) {
+        startCalled = false;
+      }
+    }
+    if (startCalled) {
+      final ServerContext ctx = getServerContext();
+      final String zRoot = getServerContext().getZooKeeperRoot();
+      Predicate<String> pred = path -> false;
+      for (String lockPath : Set.of(Constants.ZMANAGER_LOCK, Constants.ZGC_LOCK,
+          Constants.ZCOMPACTORS, Constants.ZSSERVERS, Constants.ZTSERVERS)) {
+        pred = pred.or(path -> path.startsWith(zRoot + lockPath));
+      }
+      ctx.getZooCache().clear(pred);
+    }
+
     // ACCUMULO-2985 stop the ExecutorService after we finished using it to stop accumulo procs
     if (executor != null) {
       List<Runnable> tasksRemaining = executor.shutdownNow();

server/base/src/main/java/org/apache/accumulo/server/metadata/ServerAmpleImpl.java

@@ -320,7 +320,7 @@ public Stream<ExternalCompactionFinalState> getExternalCompactionFinalStates() {
 
     scanner.setRange(ExternalCompactionSection.getRange());
     int pLen = ExternalCompactionSection.getRowPrefix().length();
-    return scanner.stream()
+    return scanner.stream().onClose(scanner::close)
         .map(e -> ExternalCompactionFinalState.fromJson(
             ExternalCompactionId.of(e.getKey().getRowData().toString().substring(pLen)),
             e.getValue().toString()));