Resolved timeout issue in MemoryStarvedMajCIT (apache#4267)

It appears that the test was stopping the Compactors, but not
waiting for their ZooKeeper locks to be removed, before moving
on and checking the number of zookeeper locks. I think what was
happening is that the test code moved forward thinking that the
Compactors were stopped, but the Compactors were actually stopping
after this point and during the next step in the test.

Author: dlmarion

core/src/main/java/org/apache/accumulo/core/util/compaction/ExternalCompactionUtil.java

@@ -114,7 +114,6 @@ public static Map<String,List<HostAndPort>> getCompactorAddrs(ClientContext cont
       ZooReader zooReader = context.getZooReader();
       List<String> groups = zooReader.getChildren(compactorGroupsPath);
       for (String group : groups) {
-        groupsAndAddresses.putIfAbsent(group, new ArrayList<>());
         try {
           List<String> compactors = zooReader.getChildren(compactorGroupsPath + "/" + group);
           for (String compactor : compactors) {
@@ -124,6 +123,7 @@ public static Map<String,List<HostAndPort>> getCompactorAddrs(ClientContext cont
                 zooReader.getChildren(compactorGroupsPath + "/" + group + "/" + compactor);
             if (!children.isEmpty()) {
               LOG.trace("Found live compactor {} ", compactor);
+              groupsAndAddresses.putIfAbsent(group, new ArrayList<>());
               groupsAndAddresses.get(group).add(HostAndPort.fromString(compactor));
             }
           }

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

@@ -519,6 +519,10 @@ public void kill(ServerType server, String hostname) throws IOException {
     stop(server, hostname);
   }
 
+  public List<Process> getCompactors(String resourceGroup) {
+    return compactorProcesses.get(resourceGroup);
+  }
+
   public List<Process> getTabletServers(String resourceGroup) {
     return tabletServerProcesses.get(resourceGroup);
   }

test/src/main/java/org/apache/accumulo/test/functional/MemoryStarvedMajCIT.java

@@ -22,6 +22,7 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
 
 import java.util.List;
 import java.util.Map;
@@ -66,6 +67,7 @@ public void configureMiniCluster(MiniAccumuloConfigImpl cfg, Configuration coreS
       cfg.getClusterServerConfiguration().setNumDefaultTabletServers(1);
       cfg.getClusterServerConfiguration().setNumDefaultScanServers(0);
       cfg.getClusterServerConfiguration().setNumDefaultCompactors(1);
+      cfg.setProperty(Property.INSTANCE_ZK_TIMEOUT, "15s");
       cfg.setProperty(Property.GENERAL_LOW_MEM_DETECTOR_INTERVAL, "5s");
       cfg.setProperty(Property.GENERAL_LOW_MEM_DETECTOR_THRESHOLD,
           Double.toString(MemoryStarvedScanIT.FREE_MEMORY_THRESHOLD));
@@ -132,10 +134,23 @@ public void testMajCPauses() throws Exception {
 
       ClientContext ctx = (ClientContext) client;
 
-      // Stop the normal compactors and start the version that will consume
-      // and free memory when we need it to
-      getCluster().getClusterControl().stopAllServers(ServerType.COMPACTOR);
+      // Kill the normal compactors and wait until their addresses in ZK are cleared
+      getCluster().getConfig().getClusterServerConfiguration().getCompactorConfiguration().keySet()
+          .forEach(resourceGroup -> {
+            List<Process> procs = getCluster().getClusterControl().getCompactors(resourceGroup);
+            for (int i = 0; i < procs.size(); i++) {
+              LOG.info("Stopping compactor process: {}", procs.get(i).pid());
+              try {
+                procs.get(i).destroyForcibly().waitFor();
+              } catch (InterruptedException e) {
+                fail("Interrupted trying to stop compactor process");
+              }
+            }
+            getCluster().getClusterControl().getCompactors(resourceGroup).clear();
+          });
+      Wait.waitFor(() -> ExternalCompactionUtil.getCompactorAddrs(ctx).size() == 0, 60_000);
 
+      // Start the Compactors that will consume and free memory when we need it to
       getCluster().getClusterControl().start(ServerType.COMPACTOR, null, 1,
           MemoryConsumingCompactor.class);
       Wait.waitFor(() -> ExternalCompactionUtil.getCompactorAddrs(ctx).size() == 4, 60_000);