Don't fully start some servers until upgrade is complete. (apache#5378)

Prevent the CompactionCoordinator, GarbageCollector, and
ScanServer from fully starting until the current version
of software matches the version stored on disk. Backported
AccumuloDataVersion.getCurrentVersion for this change.

Closes apache#5367

Author: dlmarion

server/base/src/main/java/org/apache/accumulo/server/AbstractServer.java

@@ -263,4 +263,11 @@ public void startServiceLockVerificationThread() {
   @Override
   public void close() {}
 
+  protected void waitForUpgrade() throws InterruptedException {
+    while (AccumuloDataVersion.getCurrentVersion(getContext()) < AccumuloDataVersion.get()) {
+      LOG.info("Waiting for upgrade to complete.");
+      Thread.sleep(1000);
+    }
+  }
+
 }

server/base/src/main/java/org/apache/accumulo/server/AccumuloDataVersion.java

@@ -78,4 +78,18 @@ public static int get() {
 
   public static final Set<Integer> CAN_RUN =
       Set.of(SHORTEN_RFILE_KEYS, CRYPTO_CHANGES, CURRENT_VERSION);
+
+  /**
+   * Get the stored, current working version.
+   *
+   * @param context the server context
+   * @return the stored data version
+   */
+  public static int getCurrentVersion(ServerContext context) {
+    int cv =
+        context.getServerDirs().getAccumuloPersistentVersion(context.getVolumeManager().getFirst());
+    ServerContext.ensureDataVersionCompatible(cv);
+    return cv;
+  }
+
 }

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

@@ -101,6 +101,7 @@
 import com.github.benmanes.caffeine.cache.LoadingCache;
 import com.google.common.collect.Sets;
 
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import io.micrometer.core.instrument.Tag;
 
 public class CompactionCoordinator extends AbstractServer implements
@@ -265,8 +266,16 @@ protected Collection<Tag> getServiceTags(HostAndPort clientAddress) {
   }
 
   @Override
+  @SuppressFBWarnings(value = "DM_EXIT", justification = "main class can call System.exit")
   public void run() {
 
+    try {
+      waitForUpgrade();
+    } catch (InterruptedException e) {
+      LOG.error("Interrupted while waiting for upgrade to complete, exiting...");
+      System.exit(1);
+    }
+
     ServerAddress coordinatorAddress = null;
     try {
       coordinatorAddress = startCoordinatorClientService();

server/compaction-coordinator/src/test/java/org/apache/accumulo/coordinator/CompactionCoordinatorTest.java

@@ -224,6 +224,10 @@ public void resetInternals() {
     public Collection<Tag> getServiceTags(HostAndPort clientAddr) {
       return List.of();
     }
+
+    @Override
+    protected void waitForUpgrade() throws InterruptedException {}
+
   }
 
   @Test

server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java

@@ -152,6 +152,14 @@ boolean inSafeMode() {
   @Override
   @SuppressFBWarnings(value = "DM_EXIT", justification = "main class can call System.exit")
   public void run() {
+
+    try {
+      waitForUpgrade();
+    } catch (InterruptedException e) {
+      LOG.error("Interrupted while waiting for upgrade to complete, exiting...");
+      System.exit(1);
+    }
+
     final VolumeManager fs = getContext().getVolumeManager();
 
     // Sleep for an initial period, giving the manager time to start up and

server/manager/src/main/java/org/apache/accumulo/manager/upgrade/UpgradeCoordinator.java

@@ -160,9 +160,7 @@ public synchronized void upgradeZookeeper(ServerContext context,
         "Not currently in a suitable state to do zookeeper upgrade %s", status);
 
     try {
-      int cv = context.getServerDirs()
-          .getAccumuloPersistentVersion(context.getVolumeManager().getFirst());
-      ServerContext.ensureDataVersionCompatible(cv);
+      int cv = AccumuloDataVersion.getCurrentVersion(context);
       this.currentVersion = cv;
 
       if (cv == AccumuloDataVersion.get()) {

server/tserver/src/main/java/org/apache/accumulo/tserver/ScanServer.java

@@ -127,6 +127,8 @@
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Sets;
 
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+
 public class ScanServer extends AbstractServer
     implements TabletScanClientService.Iface, TabletHostingServer, ServerProcessService.Iface {
 
@@ -375,15 +377,24 @@ private ServiceLock announceExistence() {
   }
 
   @Override
+  @SuppressFBWarnings(value = "DM_EXIT", justification = "main class can call System.exit")
   public void run() {
+
+    try {
+      waitForUpgrade();
+    } catch (InterruptedException e) {
+      LOG.error("Interrupted while waiting for upgrade to complete, exiting...");
+      System.exit(1);
+    }
+
     SecurityUtil.serverLogin(getConfiguration());
 
     ServerAddress address = null;
     try {
       address = startScanServerClientService();
       clientAddress = address.getAddress();
     } catch (UnknownHostException e1) {
-      throw new RuntimeException("Failed to start the compactor client service", e1);
+      throw new RuntimeException("Failed to start the scan server client service", e1);
     }
 
     MetricsInfo metricsInfo = getContext().getMetricsInfo();