Stop tracking last compactor check-in for non-existent groups

server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/CompactionCoordinator.java

@@ -156,7 +156,6 @@ public class CompactionCoordinator
       new ConcurrentHashMap<>();
 
   /* Map of group name to last time compactor called to get a compaction job */
-  // ELASTICITY_TODO need to clean out groups that are no longer configured..
   private final Map<CompactorGroupId,Long> TIME_COMPACTOR_LAST_CHECKED = new ConcurrentHashMap<>();
 
   private final ServerContext ctx;
@@ -251,6 +250,18 @@ protected void startIdleCompactionWatcher() {
     ThreadPools.watchNonCriticalScheduledTask(future);
   }
 
+  protected void startGroupCheckThread() {
+    ScheduledFuture<?> future = schedExecutor.scheduleWithFixedDelay(() -> {
+      Set<String> compactorGroups = ExternalCompactionUtil.getCompactorAddrs(ctx).keySet();
+      for (CompactorGroupId groupId : TIME_COMPACTOR_LAST_CHECKED.keySet()) {
+        if (!compactorGroups.contains(groupId.canonical())) {
+          TIME_COMPACTOR_LAST_CHECKED.remove(groupId);
+        }
+      }
+    }, 5, 5, TimeUnit.MINUTES);
+    ThreadPools.watchNonCriticalScheduledTask(future);
+  }
+
   private void idleCompactionWarning() {
 
     long now = System.currentTimeMillis();
@@ -295,6 +306,8 @@ public void run() {
 
     startIdleCompactionWatcher();
 
+    startGroupCheckThread();
+
     try {
       shutdown.await();
     } catch (InterruptedException e) {

server/manager/src/test/java/org/apache/accumulo/manager/compaction/CompactionCoordinatorTest.java

@@ -115,6 +115,9 @@ public TestCoordinator(ServerContext ctx, SecurityOperation security,
     @Override
     protected void startDeadCompactionDetector() {}
 
+    @Override
+    protected void startGroupCheckThread() {}
+
     @Override
     protected long getTServerCheckInterval() {
       return 5000L;