Merge branch 'main' into 4973-new-monitor-metrics

Author: dlmarion

core/src/main/java/org/apache/accumulo/core/Constants.java

@@ -74,8 +74,6 @@ public class Constants {
   public static final String ZDEAD = "/dead";
   public static final String ZDEADTSERVERS = ZDEAD + "/tservers";
 
-  public static final String ZPROBLEMS = "/problems";
-
   public static final String ZFATE = "/fate";
 
   public static final String ZNEXT_FILE = "/next_file";

core/src/main/java/org/apache/accumulo/core/clientImpl/InstanceOperationsImpl.java

@@ -278,18 +278,23 @@ public List<ActiveScan> getActiveScans(String tserver)
   @Override
   public List<ActiveScan> getActiveScans(Collection<ServerId> servers)
       throws AccumuloException, AccumuloSecurityException {
+    servers.forEach(InstanceOperationsImpl::checkActiveScanServer);
     return queryServers(servers, this::getActiveScans, INSTANCE_OPS_SCANS_FINDER_POOL);
   }
 
-  private List<ActiveScan> getActiveScans(ServerId server)
-      throws AccumuloException, AccumuloSecurityException {
-
+  private static void checkActiveScanServer(ServerId server) {
     Objects.requireNonNull(server);
     Preconditions.checkArgument(
         server.getType() == ServerId.Type.SCAN_SERVER
             || server.getType() == ServerId.Type.TABLET_SERVER,
         "Server type %s is not %s or %s.", server.getType(), ServerId.Type.SCAN_SERVER,
         ServerId.Type.TABLET_SERVER);
+  }
+
+  private List<ActiveScan> getActiveScans(ServerId server)
+      throws AccumuloException, AccumuloSecurityException {
+
+    checkActiveScanServer(server);
 
     final var parsedTserver = HostAndPort.fromParts(server.getHost(), server.getPort());
     TabletScanClientService.Client rpcClient = null;
@@ -335,12 +340,7 @@ public List<ActiveCompaction> getActiveCompactions(String server)
   private List<ActiveCompaction> getActiveCompactions(ServerId server)
       throws AccumuloException, AccumuloSecurityException {
 
-    Objects.requireNonNull(server);
-    Preconditions.checkArgument(
-        server.getType() == ServerId.Type.COMPACTOR
-            || server.getType() == ServerId.Type.TABLET_SERVER,
-        "Server type %s is not %s or %s.", server.getType(), ServerId.Type.COMPACTOR,
-        ServerId.Type.TABLET_SERVER);
+    checkActiveCompactionServer(server);
 
     final HostAndPort serverHostAndPort = HostAndPort.fromParts(server.getHost(), server.getPort());
     final List<ActiveCompaction> as = new ArrayList<>();
@@ -371,6 +371,13 @@ private List<ActiveCompaction> getActiveCompactions(ServerId server)
     }
   }
 
+  private static void checkActiveCompactionServer(ServerId server) {
+    Objects.requireNonNull(server);
+    Preconditions.checkArgument(
+        server.getType() == Type.COMPACTOR || server.getType() == Type.TABLET_SERVER,
+        "Server type %s is not %s or %s.", server.getType(), Type.COMPACTOR, Type.TABLET_SERVER);
+  }
+
   @Override
   @Deprecated
   public List<ActiveCompaction> getActiveCompactions()
@@ -386,6 +393,7 @@ public List<ActiveCompaction> getActiveCompactions()
   @Override
   public List<ActiveCompaction> getActiveCompactions(Collection<ServerId> compactionServers)
       throws AccumuloException, AccumuloSecurityException {
+    compactionServers.forEach(InstanceOperationsImpl::checkActiveCompactionServer);
     return queryServers(compactionServers, this::getActiveCompactions,
         INSTANCE_OPS_COMPACTIONS_FINDER_POOL);
   }

core/src/main/java/org/apache/accumulo/core/conf/Property.java

@@ -362,6 +362,11 @@ public enum Property {
   GENERAL_PROCESS_BIND_ADDRESS("general.process.bind.addr", "0.0.0.0", PropertyType.STRING,
       "The local IP address to which this server should bind for sending and receiving network traffic.",
       "3.0.0"),
+  GENERAL_SERVER_LOCK_VERIFICATION_INTERVAL("general.server.lock.verification.interval", "2m",
+      PropertyType.TIMEDURATION,
+      "Interval at which the Manager and TabletServer should verify their server locks. A value of zero"
+          + " disables this check. The default value change from 0 to 2m in 3.1.0.",
+      "2.1.4"),
   // properties that are specific to manager server behavior
   MANAGER_PREFIX("manager.", null, PropertyType.PREFIX,
       "Properties in this category affect the behavior of the manager server.", "2.1.0"),
@@ -425,6 +430,12 @@ public enum Property {
       "The number of threads used to run fault-tolerant executions (FATE)."
           + " These are primarily table operations like merge.",
       "1.4.3"),
+  MANAGER_FATE_IDLE_CHECK_INTERVAL("manager.fate.idle.check.interval", "60m",
+      PropertyType.TIMEDURATION,
+      "The interval at which to check if the number of idle Fate threads has consistently been zero."
+          + " The way this is checked is an approximation. Logs a warning in the Manager log to increase"
+          + " MANAGER_FATE_THREADPOOL_SIZE. A value of zero disables this check and has a maximum value of 60m.",
+      "4.0.0"),
   MANAGER_STATUS_THREAD_POOL_SIZE("manager.status.threadpool.size", "0", PropertyType.COUNT,
       "The number of threads to use when fetching the tablet server status for balancing.  Zero "
           + "indicates an unlimited number of threads will be used.",
@@ -921,6 +932,14 @@ public enum Property {
       "The maximum amount of memory that will be used to cache results of a client query/scan. "
           + "Once this limit is reached, the buffered data is sent to the client.",
       "1.3.5"),
+  TABLE_BULK_MAX_TABLETS("table.bulk.max.tablets", "100", PropertyType.COUNT,
+      "The maximum number of tablets allowed for one bulk import file. Value of 0 is Unlimited.",
+      "2.1.0"),
+  TABLE_BULK_MAX_TABLET_FILES("table.bulk.max.tablet.files", "100", PropertyType.COUNT,
+      "The maximum number of files a bulk import can add to a single tablet.  When this property "
+          + "is exceeded for any tablet the entire bulk import operation will fail before making any "
+          + "changes. Value of 0 is unlimited.",
+      "4.0.0"),
   TABLE_FILE_TYPE("table.file.type", RFile.EXTENSION, PropertyType.FILENAME_EXT,
       "Change the type of file a table writes.", "1.3.5"),
   TABLE_LOAD_BALANCER("table.balancer", "org.apache.accumulo.core.spi.balancer.SimpleLoadBalancer",
@@ -947,17 +966,28 @@ public enum Property {
           + " defaults are used.",
       "1.3.5"),
   TABLE_FILE_MAX("table.file.max", "15", PropertyType.COUNT,
-      "The maximum number of RFiles each tablet in a table can have. When"
-          + " adjusting this property you may want to consider adjusting"
-          + " table.compaction.major.ratio also. Setting this property to 0 will make"
-          + " it default to tserver.scan.files.open.max-1, this will prevent a tablet"
-          + " from having more RFiles than can be opened. Prior to 2.1.0 this property"
-          + " was used to trigger merging minor compactions, but merging minor compactions"
-          + " were removed in 2.1.0. Now this property is only used by the"
-          + " DefaultCompactionStrategy and the RatioBasedCompactionPlanner."
-          + " The RatioBasedCompactionPlanner started using this property in 2.1.3, before"
-          + " that it did not use the property.",
+      "This property is used to signal to the compaction planner that it should be more "
+          + "aggressive for compacting tablets that exceed this limit. The "
+          + "RatioBasedCompactionPlanner will lower the compaction ratio and increase the "
+          + "priority for tablets that exceed this limit. When  adjusting this property you may "
+          + "want to consider adjusting table.compaction.major.ratio also. Setting this property "
+          + "to 0 will make it default to tserver.scan.files.open.max-1, this will prevent a tablet"
+          + " from having more RFiles than can be opened by a scan.",
       "1.4.0"),
+  TABLE_FILE_PAUSE("table.file.pause", "100", PropertyType.COUNT,
+      "When a tablet has more than this number of files, bulk imports and minor compactions "
+          + "will wait until the tablet has less files before proceeding.  This will cause back "
+          + "pressure on bulk imports and writes to tables when compactions are not keeping up. "
+          + "Only the number of files a tablet currently has is considered for pausing, the "
+          + "number of files a bulk import will add is not considered. This means a bulk import "
+          + "can surge above this limit once causing future bulk imports or minor compactions to "
+          + "pause until compactions can catch up.  This property plus "
+          + TABLE_BULK_MAX_TABLET_FILES.getKey()
+          + " determines the total number of files a tablet could temporarily surge to based on bulk "
+          + "imports.  Ideally this property would be set higher than " + TABLE_FILE_MAX.getKey()
+          + " so that compactions are more aggressive prior to reaching the pause point. Value of 0 is "
+          + "unlimited.",
+      "4.0.0"),
   TABLE_MERGE_FILE_MAX("table.merge.file.max", "10000", PropertyType.COUNT,
       "The maximum number of files that a merge operation will process.  Before "
           + "merging a sum of the number of files in the merge range is computed and if it "
@@ -994,10 +1024,6 @@ public enum Property {
       "1.3.5"),
   TABLE_BLOOM_HASHTYPE("table.bloom.hash.type", "murmur", PropertyType.STRING,
       "The bloom filter hash type.", "1.3.5"),
-  TABLE_BULK_MAX_TABLETS("table.bulk.max.tablets", "0", PropertyType.COUNT,
-      "The maximum number of tablets allowed for one bulk import file. Value of 0 is Unlimited. "
-          + "This property is only enforced in the new bulk import API.",
-      "2.1.0"),
   TABLE_DURABILITY("table.durability", "sync", PropertyType.DURABILITY,
       "The durability used to write to the write-ahead log. Legal values are:"
           + " none, which skips the write-ahead log; log, which sends the data to the"

core/src/main/java/org/apache/accumulo/core/fate/Fate.java

@@ -34,6 +34,7 @@
 import java.time.Duration;
 import java.util.EnumSet;
 import java.util.Optional;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedTransferQueue;
 import java.util.concurrent.RejectedExecutionException;
@@ -83,6 +84,7 @@ public class Fate<T> {
   private final AtomicBoolean keepRunning = new AtomicBoolean(true);
   private final TransferQueue<FateId> workQueue;
   private final Thread workFinder;
+  private final ConcurrentLinkedQueue<Integer> idleCountHistory = new ConcurrentLinkedQueue<>();
 
   public enum TxInfo {
     TX_NAME, AUTO_CLEAN, EXCEPTION, TX_AGEOFF, RETURN_VALUE
@@ -355,7 +357,8 @@ public Fate(T environment, FateStore<T> store, boolean runDeadResCleaner,
       // resize the pool if the property changed
       ThreadPools.resizePool(pool, conf, Property.MANAGER_FATE_THREADPOOL_SIZE);
       // If the pool grew, then ensure that there is a TransactionRunner for each thread
-      int needed = conf.getCount(Property.MANAGER_FATE_THREADPOOL_SIZE) - pool.getQueue().size();
+      final int configured = conf.getCount(Property.MANAGER_FATE_THREADPOOL_SIZE);
+      final int needed = configured - pool.getQueue().size();
       if (needed > 0) {
         for (int i = 0; i < needed; i++) {
           try {
@@ -372,6 +375,41 @@ public Fate(T environment, FateStore<T> store, boolean runDeadResCleaner,
             break;
           }
         }
+        idleCountHistory.clear();
+      } else {
+        // The property did not change, but should it based on idle Fate threads? Maintain
+        // count of the last X minutes of idle Fate threads. If zero 95% of the time, then suggest
+        // that the
+        // MANAGER_FATE_THREADPOOL_SIZE be increased.
+        final long interval = Math.min(60, TimeUnit.MILLISECONDS
+            .toMinutes(conf.getTimeInMillis(Property.MANAGER_FATE_IDLE_CHECK_INTERVAL)));
+        if (interval == 0) {
+          idleCountHistory.clear();
+        } else {
+          if (idleCountHistory.size() >= interval * 2) { // this task runs every 30s
+            int zeroFateThreadsIdleCount = 0;
+            for (Integer idleConsumerCount : idleCountHistory) {
+              if (idleConsumerCount == 0) {
+                zeroFateThreadsIdleCount++;
+              }
+            }
+            boolean needMoreThreads =
+                (zeroFateThreadsIdleCount / (double) idleCountHistory.size()) >= 0.95;
+            if (needMoreThreads) {
+              log.warn(
+                  "All Fate threads appear to be busy for the last {} minutes,"
+                      + " consider increasing property: {}",
+                  interval, Property.MANAGER_FATE_THREADPOOL_SIZE.getKey());
+              // Clear the history so that we don't log for interval minutes.
+              idleCountHistory.clear();
+            } else {
+              while (idleCountHistory.size() >= interval * 2) {
+                idleCountHistory.remove();
+              }
+            }
+          }
+          idleCountHistory.add(workQueue.getWaitingConsumerCount());
+        }
       }
     }, 3, 30, SECONDS));
     this.transactionExecutor = pool;
@@ -611,5 +649,6 @@ public void shutdown(long timeout, TimeUnit timeUnit) {
     if (deadResCleanerExecutor != null) {
       deadResCleanerExecutor.shutdownNow();
     }
+    idleCountHistory.clear();
   }
 }

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

@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.core.fate.zookeeper;
+
+import java.util.List;
+import java.util.Objects;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * Immutable data class used by zoo cache to hold what it is caching for single zookeeper node. Data
+ * and children are obtained from zookeeper at different times. This class is structured so that
+ * data can be obtained first and then children added later or visa veras.
+ *
+ * <p>
+ * Four distinct states can be cached for a zookeeper node.
+ * <ul>
+ * <li>Can cache that a node does not exist in zookeeper. This state is represented by data, state,
+ * and children all being null.</li>
+ * <li>Can cache only the data for a zookeeper node. For this state data and stat are non-null while
+ * children is null. Calling getChildren on node in this state will throw an exception.</li>
+ * <li>Can cache only the children for a zookeeper node. For this state children is non-null while
+ * data and stat are null. Calling getData or getStat on node in this state will throw an
+ * exception.</li>
+ * <li>Can cache the children and data for a zookeeper node. For this state data,stat, and children
+ * are non-null.</li>
+ * </ul>
+ * <p>
+ *
+ */
+class ZcNode {
+
+  private final byte[] data;
+  private final ZooCache.ZcStat stat;
+  private final List<String> children;
+
+  static final ZcNode NON_EXISTENT = new ZcNode();
+
+  private ZcNode() {
+    this.data = null;
+    this.stat = null;
+    this.children = null;
+  }
+
+  /**
+   * Creates a new ZcNode that combines the data and stat from an existing ZcNode and sets the
+   * children.
+   */
+  ZcNode(List<String> children, ZcNode existing) {
+    Objects.requireNonNull(children);
+    if (existing == null) {
+      this.data = null;
+      this.stat = null;
+    } else {
+      this.data = existing.data;
+      this.stat = existing.stat;
+    }
+
+    this.children = List.copyOf(children);
+  }
+
+  /**
+   * Creates a new ZcNode that combines the children from an existing ZcNode and sets the data and
+   * stat.
+   */
+  ZcNode(byte[] data, ZooCache.ZcStat zstat, ZcNode existing) {
+    this.data = Objects.requireNonNull(data);
+    this.stat = Objects.requireNonNull(zstat);
+    if (existing == null) {
+      this.children = null;
+    } else {
+      this.children = existing.children;
+    }
+  }
+
+  /**
+   * @return the data if the node exists and the data was set OR return null when the node does not
+   *         exist
+   * @throws IllegalStateException in the case where the node exists and the data was never set
+   */
+  byte[] getData() {
+    Preconditions.checkState(cachedData());
+    return data;
+  }
+
+  /**
+   * @return the stat if the node exists and the stat was set OR return null when the node does not
+   *         exist
+   * @throws IllegalStateException in the case where the node exists and the data was never set
+   */
+  ZooCache.ZcStat getStat() {
+    Preconditions.checkState(cachedData());
+    return stat;
+  }
+
+  /**
+   * @return the children if the node exists and the children were set OR return null when the node
+   *         does not exist exists
+   * @throws IllegalStateException in the case where the node exists and the children were never set
+   */
+  List<String> getChildren() {
+    Preconditions.checkState(cachedChildren());
+    return children;
+  }
+
+  /**
+   * @return true if the node does not exists or it exists and children are cached.
+   */
+  boolean cachedChildren() {
+    return children != null || notExists();
+  }
+
+  /**
+   * @return true if the node does not exists or it exists and data and stat cached.
+   */
+  boolean cachedData() {
+    return data != null || notExists();
+  }
+
+  /**
+   * @return true if the node does not exists in zookeeper
+   */
+  boolean notExists() {
+    return stat == null && data == null && children == null;
+  }
+}