Merge branch '2.1'

keith-turner · keith-turner · commit ea5afb9d85e3 · 2024-02-22T19:00:28.000-05:00
diff --git a/server/base/src/main/java/org/apache/accumulo/server/manager/state/MetaDataStateStore.java b/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);
diff --git a/server/base/src/main/java/org/apache/accumulo/server/manager/state/TabletStateChangeIterator.java b/server/base/src/main/java/org/apache/accumulo/server/manager/state/TabletStateChangeIterator.java
@@ -211,6 +211,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;
diff --git a/test/src/main/java/org/apache/accumulo/test/manager/SuspendedTabletsIT.java b/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
@@ -152,99 +150,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.get());
-      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, AccumuloTable.METADATA.tableId());
-      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.getTserverLocation(),
-              Long.valueOf(tab.getTserverSession(), 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.get());
-
-      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;
 
@@ -256,6 +209,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 ...
@@ -303,27 +257,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);
       }
     }
   }
@@ -333,6 +303,93 @@ 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, AccumuloTable.METADATA.tableId());
+      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.getTserverLocation(),
+              Long.valueOf(tab.getTserverSession(), 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.get());
+
+      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.get());
+      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

Original file line number	Diff line number	Diff line change
`@@ -150,9 +150,8 @@ public void unsuspend(Collection<TabletLocationState> tablets) throws Distribute`
`150`	`150`	`try (var tabletsMutator = ample.mutateTablets()) {`
`151`	`151`	`for (TabletLocationState tls : tablets) {`
`152`	`152`	`if (tls.suspend != null) {`
`153`		`- continue;`
	`153`	`+ tabletsMutator.mutateTablet(tls.extent).deleteSuspension().mutate();`
`154`	`154`	`}`
`155`		`- tabletsMutator.mutateTablet(tls.extent).deleteSuspension().mutate();`
`156`	`155`	`}`
`157`	`156`	`} catch (RuntimeException ex) {`
`158`	`157`	`throw new DistributedStoreException(ex);`