fixes clearing suspension for offline tables (apache#4295)

There was code in the manager that seemed to have the intent of clearing
suspension markers in tablets for an offline table.  This code was not
working or tested.  This commit fixes this code and adds a tests that
validates that suspension markers are removed when a table is taken
offline.

fixes apache#3314

Author: keith-turner

server/base/src/main/java/org/apache/accumulo/server/manager/state/MetaDataStateStore.java

@@ -150,9 +150,8 @@ public void unsuspend(Collection<TabletLocationState> tablets) throws Distribute
     try (var tabletsMutator = ample.mutateTablets()) {
       for (TabletLocationState tls : tablets) {
         if (tls.suspend != null) {
-          continue;
+          tabletsMutator.mutateTablet(tls.extent).deleteSuspension().mutate();
         }
-        tabletsMutator.mutateTablet(tls.extent).deleteSuspension().mutate();
       }
     } catch (RuntimeException ex) {
       throw new DistributedStoreException(ex);

server/base/src/main/java/org/apache/accumulo/server/manager/state/TabletStateChangeIterator.java

@@ -209,6 +209,9 @@ protected void consume() throws IOException {
         case ASSIGNED_TO_DEAD_SERVER:
           return;
         case SUSPENDED:
+          // Always return data about suspended tablets. Need to clear the suspension stats when the
+          // tablet is offline. May need to assign the tablet when the tablet is online.
+          return;
         case UNASSIGNED:
           if (shouldBeOnline) {
             return;

test/src/main/java/org/apache/accumulo/test/manager/SuspendedTabletsIT.java

@@ -83,7 +83,6 @@ public class SuspendedTabletsIT extends ConfigurableMacBase {
   private static ExecutorService THREAD_POOL;
 
   public static final int TSERVERS = 3;
-  public static final long SUSPEND_DURATION = 80;
   public static final int TABLETS = 30;
 
   private ProcessReference metadataTserverProcess;
@@ -95,7 +94,6 @@ protected Duration defaultTimeout() {
 
   @Override
   public void configure(MiniAccumuloConfigImpl cfg, Configuration fsConf) {
-    cfg.setProperty(Property.TABLE_SUSPEND_DURATION, SUSPEND_DURATION + "s");
     cfg.setClientProperty(ClientProperty.INSTANCE_ZOOKEEPERS_TIMEOUT, "5s");
     cfg.setProperty(Property.INSTANCE_ZK_TIMEOUT, "5s");
     // Start with 1 tserver, we'll increase that later
@@ -149,99 +147,54 @@ public void setUp() throws Exception {
     getCluster().start();
   }
 
+  enum AfterSuspendAction {
+    RESUME("80s"),
+    // Set a long suspend time for testing offline table, want the suspension to be cleared because
+    // the tablet went offline and not the because the suspension timed out.
+    OFFLINE("800s");
+
+    public final String suspendTime;
+
+    AfterSuspendAction(String suspendTime) {
+      this.suspendTime = suspendTime;
+    }
+  }
+
   @Test
   public void crashAndResumeTserver() throws Exception {
     // Run the test body. When we get to the point where we need a tserver to go away, get rid of it
     // via crashing
-    suspensionTestBody((ctx, locs, count) -> {
-      // Exclude the tablet server hosting the metadata table from the list and only
-      // kill tablet servers that are not hosting the metadata table.
-      List<ProcessReference> procs = getCluster().getProcesses().get(ServerType.TABLET_SERVER)
-          .stream().filter(p -> !metadataTserverProcess.equals(p)).collect(Collectors.toList());
-      Collections.shuffle(procs, random);
-      assertEquals(TSERVERS - 1, procs.size(), "Not enough tservers exist");
-      assertTrue(procs.size() >= count, "Attempting to kill more tservers (" + count
-          + ") than exist in the cluster (" + procs.size() + ")");
+    suspensionTestBody(new CrashTserverKiller(), AfterSuspendAction.RESUME);
+  }
 
-      for (int i = 0; i < count; ++i) {
-        ProcessReference pr = procs.get(i);
-        log.info("Crashing {}", pr.getProcess());
-        getCluster().killProcess(ServerType.TABLET_SERVER, pr);
-      }
-    });
+  @Test
+  public void crashAndOffline() throws Exception {
+    // Test to ensure that taking a table offline causes the suspension markers to be cleared.
+    // Suspension markers can prevent balancing and possibly cause other problems, so its good to
+    // clear them for offline tables.
+    suspensionTestBody(new CrashTserverKiller(), AfterSuspendAction.OFFLINE);
   }
 
   @Test
   public void shutdownAndResumeTserver() throws Exception {
     // Run the test body. When we get to the point where we need tservers to go away, stop them via
     // a clean shutdown.
-    suspensionTestBody((ctx, locs, count) -> {
-      Set<TServerInstance> tserverSet = new HashSet<>();
-      Set<TServerInstance> metadataServerSet = new HashSet<>();
-
-      TabletLocator tl = TabletLocator.getLocator(ctx, MetadataTable.ID);
-      for (TabletLocationState tls : locs.locationStates.values()) {
-        if (tls.current != null) {
-          // add to set of all servers
-          tserverSet.add(tls.current.getServerInstance());
-
-          // get server that the current tablets metadata is on
-          TabletLocator.TabletLocation tab =
-              tl.locateTablet(ctx, tls.extent.toMetaRow(), false, false);
-          // add it to the set of servers with metadata
-          metadataServerSet
-              .add(new TServerInstance(tab.tablet_location, Long.valueOf(tab.tablet_session, 16)));
-        }
-      }
-
-      // remove servers with metadata on them from the list of servers to be shutdown
-      assertEquals(1, metadataServerSet.size(), "Expecting a single tServer in metadataServerSet");
-      tserverSet.removeAll(metadataServerSet);
-
-      assertEquals(TSERVERS - 1, tserverSet.size(),
-          "Expecting " + (TSERVERS - 1) + " tServers in shutdown-list");
-
-      List<TServerInstance> tserversList = new ArrayList<>(tserverSet);
-      Collections.shuffle(tserversList, random);
-
-      for (int i1 = 0; i1 < count; ++i1) {
-        final String tserverName = tserversList.get(i1).getHostPortSession();
-        ThriftClientTypes.MANAGER.executeVoid(ctx, client -> {
-          log.info("Sending shutdown command to {} via ManagerClientService", tserverName);
-          client.shutdownTabletServer(null, ctx.rpcCreds(), tserverName, false);
-        });
-      }
+    suspensionTestBody(new ShutdownTserverKiller(), AfterSuspendAction.RESUME);
+  }
 
-      log.info("Waiting for tserver process{} to die", count == 1 ? "" : "es");
-      for (int i2 = 0; i2 < 10; ++i2) {
-        List<ProcessReference> deadProcs = new ArrayList<>();
-        for (ProcessReference pr1 : getCluster().getProcesses().get(ServerType.TABLET_SERVER)) {
-          Process p = pr1.getProcess();
-          if (!p.isAlive()) {
-            deadProcs.add(pr1);
-          }
-        }
-        for (ProcessReference pr2 : deadProcs) {
-          log.info("Process {} is dead, informing cluster control about this", pr2.getProcess());
-          getCluster().getClusterControl().killProcess(ServerType.TABLET_SERVER, pr2);
-          --count;
-        }
-        if (count == 0) {
-          return;
-        } else {
-          Thread.sleep(SECONDS.toMillis(2));
-        }
-      }
-      throw new IllegalStateException("Tablet servers didn't die!");
-    });
+  @Test
+  public void shutdownAndOffline() throws Exception {
+    // Test to ensure that taking a table offline causes the suspension markers to be cleared.
+    suspensionTestBody(new ShutdownTserverKiller(), AfterSuspendAction.OFFLINE);
   }
 
   /**
    * Main test body for suspension tests.
    *
    * @param serverStopper callback which shuts down some tablet servers.
    */
-  private void suspensionTestBody(TServerKiller serverStopper) throws Exception {
+  private void suspensionTestBody(TServerKiller serverStopper, AfterSuspendAction action)
+      throws Exception {
     try (AccumuloClient client = Accumulo.newClient().from(getClientProperties()).build()) {
       ClientContext ctx = (ClientContext) client;
 
@@ -253,6 +206,7 @@ private void suspensionTestBody(TServerKiller serverStopper) throws Exception {
       }
       log.info("Creating table " + tableName);
       NewTableConfiguration ntc = new NewTableConfiguration().withSplits(splitPoints);
+      ntc.setProperties(Map.of(Property.TABLE_SUSPEND_DURATION.getKey(), action.suspendTime));
       ctx.tableOperations().create(tableName, ntc);
 
       // Wait for all of the tablets to hosted ...
@@ -300,27 +254,43 @@ private void suspensionTestBody(TServerKiller serverStopper) throws Exception {
         assertEquals(beforeDeathState.hosted.get(server), deadTabletsByServer.get(server));
       }
       assertEquals(TABLETS, ds.hostedCount + ds.suspendedCount);
-      // Restart the first tablet server, making sure it ends up on the same port
-      HostAndPort restartedServer = deadTabletsByServer.keySet().iterator().next();
-      log.info("Restarting " + restartedServer);
-      getCluster().getClusterControl().start(ServerType.TABLET_SERVER,
-          Map.of(Property.TSERV_CLIENTPORT.getKey(), "" + restartedServer.getPort(),
-              Property.TSERV_PORTSEARCH.getKey(), "false"),
-          1);
-
-      // Eventually, the suspended tablets should be reassigned to the newly alive tserver.
-      log.info("Awaiting tablet unsuspension for tablets belonging to " + restartedServer);
-      while (ds.suspended.containsKey(restartedServer) || ds.assignedCount != 0) {
-        Thread.sleep(1000);
-        ds = TabletLocations.retrieve(ctx, tableName);
-      }
-      assertEquals(deadTabletsByServer.get(restartedServer), ds.hosted.get(restartedServer));
 
-      // Finally, after much longer, remaining suspended tablets should be reassigned.
-      log.info("Awaiting tablet reassignment for remaining tablets");
-      while (ds.hostedCount != TABLETS) {
-        Thread.sleep(1000);
-        ds = TabletLocations.retrieve(ctx, tableName);
+      assertTrue(ds.suspendedCount > 0);
+
+      if (action == AfterSuspendAction.OFFLINE) {
+        client.tableOperations().offline(tableName, true);
+
+        while (ds.suspendedCount > 0) {
+          Thread.sleep(1000);
+          ds = TabletLocations.retrieve(ctx, tableName);
+          log.info("Waiting for suspended {}", ds.suspended);
+        }
+      } else if (action == AfterSuspendAction.RESUME) {
+        // Restart the first tablet server, making sure it ends up on the same port
+        HostAndPort restartedServer = deadTabletsByServer.keySet().iterator().next();
+        log.info("Restarting " + restartedServer);
+        getCluster().getClusterControl()
+            .start(
+                ServerType.TABLET_SERVER, Map.of(Property.TSERV_CLIENTPORT.getKey(),
+                    "" + restartedServer.getPort(), Property.TSERV_PORTSEARCH.getKey(), "false"),
+                1);
+
+        // Eventually, the suspended tablets should be reassigned to the newly alive tserver.
+        log.info("Awaiting tablet unsuspension for tablets belonging to " + restartedServer);
+        while (ds.suspended.containsKey(restartedServer) || ds.assignedCount != 0) {
+          Thread.sleep(1000);
+          ds = TabletLocations.retrieve(ctx, tableName);
+        }
+        assertEquals(deadTabletsByServer.get(restartedServer), ds.hosted.get(restartedServer));
+
+        // Finally, after much longer, remaining suspended tablets should be reassigned.
+        log.info("Awaiting tablet reassignment for remaining tablets");
+        while (ds.hostedCount != TABLETS) {
+          Thread.sleep(1000);
+          ds = TabletLocations.retrieve(ctx, tableName);
+        }
+      } else {
+        throw new IllegalStateException("Unknown action " + action);
       }
     }
   }
@@ -330,6 +300,95 @@ void eliminateTabletServers(ClientContext ctx, TabletLocations locs, int count)
         throws Exception;
   }
 
+  private class ShutdownTserverKiller implements TServerKiller {
+
+    @Override
+    public void eliminateTabletServers(ClientContext ctx, TabletLocations locs, int count)
+        throws Exception {
+
+      Set<TServerInstance> tserverSet = new HashSet<>();
+      Set<TServerInstance> metadataServerSet = new HashSet<>();
+
+      TabletLocator tl = TabletLocator.getLocator(ctx, MetadataTable.ID);
+      for (TabletLocationState tls : locs.locationStates.values()) {
+        if (tls.current != null) {
+          // add to set of all servers
+          tserverSet.add(tls.current.getServerInstance());
+
+          // get server that the current tablets metadata is on
+          TabletLocator.TabletLocation tab =
+              tl.locateTablet(ctx, tls.extent.toMetaRow(), false, false);
+          // add it to the set of servers with metadata
+          metadataServerSet
+              .add(new TServerInstance(tab.tablet_location, Long.valueOf(tab.tablet_session, 16)));
+        }
+      }
+
+      // remove servers with metadata on them from the list of servers to be shutdown
+      assertEquals(1, metadataServerSet.size(), "Expecting a single tServer in metadataServerSet");
+      tserverSet.removeAll(metadataServerSet);
+
+      assertEquals(TSERVERS - 1, tserverSet.size(),
+          "Expecting " + (TSERVERS - 1) + " tServers in shutdown-list");
+
+      List<TServerInstance> tserversList = new ArrayList<>(tserverSet);
+      Collections.shuffle(tserversList, random);
+
+      for (int i1 = 0; i1 < count; ++i1) {
+        final String tserverName = tserversList.get(i1).getHostPortSession();
+        ThriftClientTypes.MANAGER.executeVoid(ctx, client -> {
+          log.info("Sending shutdown command to {} via ManagerClientService", tserverName);
+          client.shutdownTabletServer(null, ctx.rpcCreds(), tserverName, false);
+        });
+      }
+
+      log.info("Waiting for tserver process{} to die", count == 1 ? "" : "es");
+      for (int i2 = 0; i2 < 10; ++i2) {
+        List<ProcessReference> deadProcs = new ArrayList<>();
+        for (ProcessReference pr1 : getCluster().getProcesses().get(ServerType.TABLET_SERVER)) {
+          Process p = pr1.getProcess();
+          if (!p.isAlive()) {
+            deadProcs.add(pr1);
+          }
+        }
+        for (ProcessReference pr2 : deadProcs) {
+          log.info("Process {} is dead, informing cluster control about this", pr2.getProcess());
+          getCluster().getClusterControl().killProcess(ServerType.TABLET_SERVER, pr2);
+          --count;
+        }
+        if (count == 0) {
+          return;
+        } else {
+          Thread.sleep(SECONDS.toMillis(2));
+        }
+      }
+      throw new IllegalStateException("Tablet servers didn't die!");
+
+    }
+  }
+
+  private class CrashTserverKiller implements TServerKiller {
+
+    @Override
+    public void eliminateTabletServers(ClientContext ctx, TabletLocations locs, int count)
+        throws Exception {
+      // Exclude the tablet server hosting the metadata table from the list and only
+      // kill tablet servers that are not hosting the metadata table.
+      List<ProcessReference> procs = getCluster().getProcesses().get(ServerType.TABLET_SERVER)
+          .stream().filter(p -> !metadataTserverProcess.equals(p)).collect(Collectors.toList());
+      Collections.shuffle(procs, random);
+      assertEquals(TSERVERS - 1, procs.size(), "Not enough tservers exist");
+      assertTrue(procs.size() >= count, "Attempting to kill more tservers (" + count
+          + ") than exist in the cluster (" + procs.size() + ")");
+
+      for (int i = 0; i < count; ++i) {
+        ProcessReference pr = procs.get(i);
+        log.info("Crashing {}", pr.getProcess());
+        getCluster().killProcess(ServerType.TABLET_SERVER, pr);
+      }
+    }
+  }
+
   private static final AtomicInteger threadCounter = new AtomicInteger(0);
 
   @BeforeAll