Added new scan executor metric

core/src/main/java/org/apache/accumulo/core/metrics/MetricsProducer.java

@@ -725,6 +725,7 @@ public interface MetricsProducer {
   String METRICS_SCAN_SCANNED_ENTRIES = METRICS_SCAN_PREFIX + "query.scanned.entries";
   String METRICS_SCAN_ZOMBIE_THREADS = METRICS_SCAN_PREFIX + "zombie.threads";
   String METRICS_SCAN_TABLET_METADATA_CACHE = METRICS_SCAN_PREFIX + "tablet.metadata.cache";
+  String METRICS_SCAN_EXECUTOR_EXCEPTIONS = METRICS_SCAN_PREFIX + "executor.exceptions";
 
   String METRICS_TSERVER_PREFIX = "accumulo.tserver.";
   String METRICS_TSERVER_ENTRIES = METRICS_TSERVER_PREFIX + "entries";

server/tserver/src/main/java/org/apache/accumulo/tserver/metrics/TabletServerScanMetrics.java

@@ -19,6 +19,7 @@
 package org.apache.accumulo.tserver.metrics;
 
 import java.time.Duration;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.LongAdder;
 import java.util.function.IntSupplier;
@@ -28,6 +29,7 @@
 
 import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.DistributionSummary;
+import io.micrometer.core.instrument.FunctionCounter;
 import io.micrometer.core.instrument.Gauge;
 import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.Timer;
@@ -47,6 +49,9 @@ public class TabletServerScanMetrics implements MetricsProducer {
   private final LongAdder queryResultCount = new LongAdder();
   private final LongAdder queryResultBytes = new LongAdder();
   private final LongAdder scannedCount = new LongAdder();
+  private final ConcurrentHashMap<String,AtomicLong> executorExceptionCounts =
+      new ConcurrentHashMap<>();
+  private volatile MeterRegistry registry = null;
 
   public void incrementLookupCount(long amount) {
     this.lookupCount.add(amount);
@@ -124,8 +129,28 @@ public TabletServerScanMetrics(IntSupplier openFileSupplier) {
     openFiles = openFileSupplier;
   }
 
+  public void incrementExecutorExceptions(String executorName) {
+    executorExceptionCounts.computeIfAbsent(executorName, k -> {
+      AtomicLong counter = new AtomicLong(0);
+      // Register the counter if the registry is already available
+      if (registry != null) {
+        registerExecutorExceptionCounter(executorName, counter);
+      }
+      return counter;
+    }).incrementAndGet();
+  }
+
+  private void registerExecutorExceptionCounter(String executorName, AtomicLong counter) {
+    FunctionCounter.builder(METRICS_SCAN_EXECUTOR_EXCEPTIONS, counter, AtomicLong::get)
+        .tags("executor", executorName)
+        .description(
+            "Number of exceptions thrown from the iterator stack during scan execution, tagged by executor name")
+        .register(registry);
+  }
+
   @Override
   public void registerMetrics(MeterRegistry registry) {
+    this.registry = registry;
     Gauge.builder(METRICS_SCAN_OPEN_FILES, openFiles::getAsInt)
         .description("Number of files open for scans").register(registry);
     scans = Timer.builder(METRICS_SCAN_TIMES).description("Scans").register(registry);
@@ -156,6 +181,7 @@ public void registerMetrics(MeterRegistry registry) {
     Gauge.builder(METRICS_SCAN_ZOMBIE_THREADS, this, TabletServerScanMetrics::getZombieThreadsCount)
         .description("Number of scan threads that have no associated client session")
         .register(registry);
+    executorExceptionCounts.forEach(this::registerExecutorExceptionCounter);
   }
 
 }

server/tserver/src/main/java/org/apache/accumulo/tserver/scan/NextBatchTask.java

@@ -23,6 +23,7 @@
 
 import org.apache.accumulo.core.client.SampleNotPresentException;
 import org.apache.accumulo.core.iteratorsImpl.system.IterationInterruptedException;
+import org.apache.accumulo.core.spi.scan.SimpleScanDispatcher;
 import org.apache.accumulo.server.fs.TooManyFilesException;
 import org.apache.accumulo.tserver.TabletHostingServer;
 import org.apache.accumulo.tserver.session.SingleScanSession;
@@ -48,6 +49,18 @@ public NextBatchTask(TabletHostingServer server, long scanID, AtomicBoolean inte
     }
   }
 
+  private void recordException(SingleScanSession scanSession) {
+    if (scanSession != null && server.getScanMetrics() != null) {
+      String executorName = getExecutorName(scanSession);
+      server.getScanMetrics().incrementExecutorExceptions(executorName);
+    }
+  }
+
+  private String getExecutorName(SingleScanSession scanSession) {
+    String executorName = scanSession.getExecutionHints().get("scan_type");
+    return executorName != null ? executorName : SimpleScanDispatcher.DEFAULT_SCAN_EXECUTOR_NAME;
+  }
+
   @Override
   public void run() {
 
@@ -93,10 +106,12 @@ public void run() {
         addResult(iie);
       }
     } catch (TooManyFilesException | SampleNotPresentException e) {
+      recordException(scanSession);
       addResult(e);
     } catch (IOException | RuntimeException e) {
       log.warn("exception while scanning tablet {} for {}", scanSession.extent, scanSession.client,
           e);
+      recordException(scanSession);
       addResult(e);
     } finally {
       transitionFromRunning();