Merge branch '3.1'

Author: dlmarion

core/src/main/java/org/apache/accumulo/core/Constants.java

@@ -81,6 +81,8 @@ public class Constants {
   public static final String ZHDFS_RESERVATIONS = "/hdfs_reservations";
   public static final String ZRECOVERY = "/recovery";
 
+  public static final String ZUPGRADE_PROGRESS = "/upgrade_progress";
+
   /**
    * Base znode for storing secret keys that back delegation tokens
    */

server/manager/src/main/java/org/apache/accumulo/manager/Manager.java

@@ -128,7 +128,6 @@
 import org.apache.accumulo.manager.split.Splitter;
 import org.apache.accumulo.manager.state.TableCounts;
 import org.apache.accumulo.manager.tableOps.TraceRepo;
-import org.apache.accumulo.manager.upgrade.PreUpgradeValidation;
 import org.apache.accumulo.manager.upgrade.UpgradeCoordinator;
 import org.apache.accumulo.server.AbstractServer;
 import org.apache.accumulo.server.HighlyAvailableService;
@@ -334,21 +333,22 @@ synchronized void setManagerState(final ManagerState newState) {
         break;
       case HAVE_LOCK:
         if (isUpgrading()) {
-          new PreUpgradeValidation().validate(getContext(), nextEvent);
-          upgradeCoordinator.upgradeZookeeper(getContext(), nextEvent);
+          upgradeCoordinator.preUpgradeValidation();
+          upgradeCoordinator.startOrContinueUpgrade();
+          upgradeCoordinator.upgradeZookeeper(nextEvent);
         }
         break;
       case NORMAL:
         if (isUpgrading()) {
-          upgradeMetadataFuture = upgradeCoordinator.upgradeMetadata(getContext(), nextEvent);
+          upgradeMetadataFuture = upgradeCoordinator.upgradeMetadata(nextEvent);
         }
         break;
       default:
         break;
     }
   }
 
-  private final UpgradeCoordinator upgradeCoordinator = new UpgradeCoordinator();
+  private final UpgradeCoordinator upgradeCoordinator;
 
   private Future<Void> upgradeMetadataFuture;
 
@@ -462,6 +462,7 @@ protected Manager(ConfigOpts opts, Function<SiteConfiguration,ServerContext> ser
       String[] args) throws IOException {
     super(ServerId.Type.MANAGER, opts, serverContextFactory, args);
     ServerContext context = super.getContext();
+    upgradeCoordinator = new UpgradeCoordinator(context);
     balancerEnvironment = new BalancerEnvironmentImpl(context);
 
     AccumuloConfiguration aconf = context.getConfiguration();
@@ -1170,6 +1171,9 @@ public void run() {
     } catch (KeeperException | InterruptedException e) {
       throw new IllegalStateException("Exception getting manager lock", e);
     }
+    // Setting the Manager state to HAVE_LOCK has the side-effect of
+    // starting the upgrade process if necessary.
+    setManagerState(ManagerState.HAVE_LOCK);
 
     // Set the HostName **after** initially creating the lock. The lock data is
     // updated below with the correct address. This prevents clients from accessing
@@ -1564,7 +1568,6 @@ private ServiceLockData getManagerLock(final ServiceLockPath zManagerLoc)
     }
 
     this.getContext().setServiceLock(getManagerLock());
-    setManagerState(ManagerState.HAVE_LOCK);
     return sld;
   }
 

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

@@ -25,7 +25,6 @@
 
 import org.apache.accumulo.core.zookeeper.ZooSession;
 import org.apache.accumulo.core.zookeeper.ZooSession.ZKUtil;
-import org.apache.accumulo.manager.EventCoordinator;
 import org.apache.accumulo.server.AccumuloDataVersion;
 import org.apache.accumulo.server.ServerContext;
 import org.apache.zookeeper.KeeperException;
@@ -48,10 +47,11 @@ public class PreUpgradeValidation {
 
   private final static Logger log = LoggerFactory.getLogger(PreUpgradeValidation.class);
 
-  public void validate(final ServerContext context, final EventCoordinator eventCoordinator) {
-    int cv = AccumuloDataVersion.getCurrentVersion(context);
-    if (cv == AccumuloDataVersion.get()) {
-      log.debug("already at current data version: {}, skipping validation", cv);
+  public void validate(final ServerContext context) {
+    int storedVersion = AccumuloDataVersion.getCurrentVersion(context);
+    int currentVersion = AccumuloDataVersion.get();
+    if (storedVersion == currentVersion) {
+      log.debug("already at current data version: {}, skipping validation", currentVersion);
       return;
     }
     validateACLs(context);

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

@@ -133,12 +133,30 @@ public boolean isParentLevelUpgraded(Ample.DataLevel level) {
           Map.of(ROOT_TABLET_META_CHANGES, new Upgrader10to11(), REMOVE_DEPRECATIONS_FOR_VERSION_3,
               new Upgrader11to12(), METADATA_FILE_JSON_ENCODING, new Upgrader12to13())));
 
+  private final ServerContext context;
+  private final UpgradeProgressTracker progressTracker;
+  private final PreUpgradeValidation preUpgradeValidator;
+
   private volatile UpgradeStatus status;
 
-  public UpgradeCoordinator() {
+  public UpgradeCoordinator(ServerContext context) {
+    this.context = context;
+    progressTracker = new UpgradeProgressTracker(context);
+    preUpgradeValidator = new PreUpgradeValidation();
     status = UpgradeStatus.INITIAL;
   }
 
+  public void preUpgradeValidation() {
+    preUpgradeValidator.validate(context);
+  }
+
+  public void startOrContinueUpgrade() {
+    // The following check will fail if an upgrade is in progress
+    // but the target version is not the current version of the
+    // software.
+    progressTracker.startOrContinueUpgrade();
+  }
+
   private void setStatus(UpgradeStatus status, EventCoordinator eventCoordinator) {
     UpgradeStatus oldStatus = this.status;
     this.status = status;
@@ -156,8 +174,7 @@ private void handleFailure(Exception e) {
     System.exit(1);
   }
 
-  public synchronized void upgradeZookeeper(ServerContext context,
-      EventCoordinator eventCoordinator) {
+  public synchronized void upgradeZookeeper(EventCoordinator eventCoordinator) {
 
     Preconditions.checkState(status == UpgradeStatus.INITIAL,
         "Not currently in a suitable state to do zookeeper upgrade %s", status);
@@ -172,15 +189,24 @@ public synchronized void upgradeZookeeper(ServerContext context,
       }
 
       if (currentVersion < AccumuloDataVersion.get()) {
-        abortIfFateTransactions(context);
+        abortIfFateTransactions();
+
+        final UpgradeProgress progress = progressTracker.getProgress();
 
         for (int v = currentVersion; v < AccumuloDataVersion.get(); v++) {
+          if (progress.getZooKeeperVersion() >= currentVersion) {
+            log.info(
+                "ZooKeeper has already been upgraded to version {}, moving on to next upgrader",
+                currentVersion);
+            continue;
+          }
           log.info("Upgrading Zookeeper - current version {} as step towards target version {}", v,
               AccumuloDataVersion.get());
           var upgrader = upgraders.get(v);
           Objects.requireNonNull(upgrader,
               "upgrade ZooKeeper: failed to find upgrader for version " + currentVersion);
           upgrader.upgradeZookeeper(context);
+          progressTracker.updateZooKeeperVersion(v);
         }
       }
 
@@ -191,8 +217,7 @@ public synchronized void upgradeZookeeper(ServerContext context,
 
   }
 
-  public synchronized Future<Void> upgradeMetadata(ServerContext context,
-      EventCoordinator eventCoordinator) {
+  public synchronized Future<Void> upgradeMetadata(EventCoordinator eventCoordinator) {
     if (status == UpgradeStatus.COMPLETE) {
       return CompletableFuture.completedFuture(null);
     }
@@ -206,35 +231,51 @@ public synchronized Future<Void> upgradeMetadata(ServerContext context,
           .numMaxThreads(Integer.MAX_VALUE).withTimeOut(60L, SECONDS)
           .withQueue(new SynchronousQueue<>()).build().submit(() -> {
             try {
+              UpgradeProgress progress = progressTracker.getProgress();
               for (int v = currentVersion; v < AccumuloDataVersion.get(); v++) {
+                if (progress.getRootVersion() >= currentVersion) {
+                  log.info(
+                      "Root table has already been upgraded to version {}, moving on to next upgrader",
+                      currentVersion);
+                  continue;
+                }
                 log.info("Upgrading Root - current version {} as step towards target version {}", v,
                     AccumuloDataVersion.get());
                 var upgrader = upgraders.get(v);
                 Objects.requireNonNull(upgrader,
                     "upgrade root: failed to find root upgrader for version " + currentVersion);
                 upgraders.get(v).upgradeRoot(context);
+                progressTracker.updateRootVersion(v);
               }
               setStatus(UpgradeStatus.UPGRADED_ROOT, eventCoordinator);
 
               for (int v = currentVersion; v < AccumuloDataVersion.get(); v++) {
+                if (progress.getMetadataVersion() >= currentVersion) {
+                  log.info(
+                      "Metadata table has already been upgraded to version {}, moving on to next upgrader",
+                      currentVersion);
+                  continue;
+                }
                 log.info(
                     "Upgrading Metadata - current version {} as step towards target version {}", v,
                     AccumuloDataVersion.get());
                 var upgrader = upgraders.get(v);
                 Objects.requireNonNull(upgrader,
                     "upgrade metadata: failed to find upgrader for version " + currentVersion);
                 upgraders.get(v).upgradeMetadata(context);
+                progressTracker.updateMetadataVersion(v);
               }
               setStatus(UpgradeStatus.UPGRADED_METADATA, eventCoordinator);
 
               log.info("Validating configuration properties.");
-              validateProperties(context);
+              validateProperties();
 
               log.info("Updating persistent data version.");
               updateAccumuloVersion(context.getServerDirs(), context.getVolumeManager(),
                   currentVersion);
               log.info("Upgrade complete");
               setStatus(UpgradeStatus.COMPLETE, eventCoordinator);
+              progressTracker.upgradeComplete();
             } catch (Exception e) {
               handleFailure(e);
             }
@@ -245,7 +286,7 @@ public synchronized Future<Void> upgradeMetadata(ServerContext context,
     }
   }
 
-  private void validateProperties(ServerContext context) {
+  private void validateProperties() {
     ConfigCheckUtil.validate(context.getSiteConfiguration(), "site configuration");
     ConfigCheckUtil.validate(context.getConfiguration(), "system configuration");
     try {
@@ -312,7 +353,7 @@ public UpgradeStatus getStatus() {
    */
   @SuppressFBWarnings(value = "DM_EXIT",
       justification = "Want to immediately stop all manager threads on upgrade error")
-  private void abortIfFateTransactions(ServerContext context) {
+  private void abortIfFateTransactions() {
     try {
       // The current version of the code creates the new accumulo.fate table on upgrade, so no
       // attempt is made to read it here. Attempting to read it this point would likely cause a hang

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

@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.manager.upgrade;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.accumulo.core.util.LazySingletons.GSON;
+
+/**
+ * Track upgrade progress for each component. The version stored is the most recent version for
+ * which an upgrade has been completed.
+ */
+public class UpgradeProgress {
+
+  private int zooKeeperVersion;
+  private int rootVersion;
+  private int metadataVersion;
+  private int upgradeTargetVersion;
+
+  public UpgradeProgress() {}
+
+  public UpgradeProgress(int currentVersion, int targetVersion) {
+    zooKeeperVersion = currentVersion;
+    rootVersion = currentVersion;
+    metadataVersion = currentVersion;
+    upgradeTargetVersion = targetVersion;
+  }
+
+  public void setZooKeeperVersion(int version) {
+    zooKeeperVersion = version;
+  }
+
+  public int getZooKeeperVersion() {
+    return zooKeeperVersion;
+  }
+
+  public void setRootVersion(int version) {
+    rootVersion = version;
+  }
+
+  public int getRootVersion() {
+    return rootVersion;
+  }
+
+  public void setMetadataVersion(int version) {
+    metadataVersion = version;
+  }
+
+  public int getMetadataVersion() {
+    return metadataVersion;
+  }
+
+  public int getUpgradeTargetVersion() {
+    return upgradeTargetVersion;
+  }
+
+  public byte[] toJsonBytes() {
+    return GSON.get().toJson(this).getBytes(UTF_8);
+  }
+
+  public static UpgradeProgress fromJsonBytes(byte[] jsonData) {
+    return GSON.get().fromJson(new String(jsonData, UTF_8), UpgradeProgress.class);
+  }
+
+}