HDDS-14754. Add ZDU upgrade action to clean up leftover state

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java

@@ -407,17 +407,11 @@ public final class OzoneConsts {
   public static final String TRANSACTION_INFO_KEY = "#TRANSACTIONINFO";
   public static final String TRANSACTION_INFO_SPLIT_KEY = "#";
 
-  public static final String PREPARE_MARKER_KEY = "#PREPAREDINFO";
-
   public static final String CONTAINER_DB_TYPE_ROCKSDB = "RocksDB";
 
   // An on-disk transient marker file used when replacing DB with checkpoint
   public static final String DB_TRANSIENT_MARKER = "dbInconsistentMarker";
 
-  // An on-disk marker file used to indicate that the OM is in prepare and
-  // should remain prepared even after a restart.
-  public static final String PREPARE_MARKER = "prepareMarker";
-
   public static final String OZONE_RATIS_SNAPSHOT_DIR = "snapshot";
 
   public static final long DEFAULT_OM_UPDATE_ID = -1L;

hadoop-hdds/framework/src/main/java/org/apache/hadoop/ozone/upgrade/AbstractUpgradeActionProvider.java

@@ -80,8 +80,7 @@ public Map<ComponentVersion, T> load() {
           LOG.info("Registering Upgrade Action : {}", action.name());
           upgradeActions.put(version, action);
         } catch (Exception e) {
-          LOG.error("Cannot instantiate Upgrade Action class {}",
-              clazz.getSimpleName(), e);
+          throw new IllegalStateException("Cannot instantiate Upgrade Action class " + clazz.getName(), e);
         }
       } else {
         LOG.warn("Found upgrade action class not of type {} : {}",

hadoop-hdds/framework/src/test/java/org/apache/hadoop/ozone/upgrade/TestAbstractUpgradeActionProvider.java

@@ -0,0 +1,75 @@
+/*
+ * 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
+ *
+ *      http://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.hadoop.ozone.upgrade;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import com.google.common.collect.ImmutableSet;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+import org.apache.hadoop.hdds.ComponentVersion;
+import org.apache.hadoop.hdds.upgrade.HDDSLayoutFeature;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for {@link AbstractUpgradeActionProvider#load()}.
+ */
+class TestAbstractUpgradeActionProvider {
+
+  /**
+   * An upgrade action whose no-arg constructor fails to instantiate must abort {@code load()} rather than
+   * silently drop the action from the returned map, which would finalize its version as a no-op.
+   */
+  @Test
+  public void testLoadFailsWhenActionCannotBeInstantiated() {
+    assertThrows(IllegalStateException.class, () -> new ThrowingActionProvider().load());
+  }
+
+  @Retention(RetentionPolicy.RUNTIME)
+  @Target(ElementType.TYPE)
+  private @interface TestUpgradeAction {
+  }
+
+  private interface TestAction extends UpgradeAction<Object> {
+  }
+
+  @TestUpgradeAction
+  public static class ThrowingTestAction implements TestAction {
+    ThrowingTestAction() {
+      throw new IllegalArgumentException("cannot construct test upgrade action");
+    }
+
+    @Override
+    public void execute(Object arg) {
+    }
+  }
+
+  private static final class ThrowingActionProvider extends AbstractUpgradeActionProvider<TestAction> {
+    ThrowingActionProvider() {
+      super(ImmutableSet.of(TestUpgradeAction.class), TestAction.class,
+          TestAbstractUpgradeActionProvider.class.getPackage().getName());
+    }
+
+    @Override
+    protected ComponentVersion extractVersion(Class<?> clazz) {
+      return HDDSLayoutFeature.INITIAL_VERSION;
+    }
+  }
+}

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/upgrade/ZduScmUpgradeActionForTest.java → hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/upgrade/ClearFinalizingStateScmUpgradeAction.java

@@ -20,16 +20,29 @@
 import static org.apache.hadoop.hdds.HDDSVersion.ZDU;
 
 import org.apache.hadoop.hdds.scm.server.OzoneStorageContainerManager;
+import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
 import org.apache.hadoop.hdds.upgrade.ScmUpgradeAction;
+import org.apache.hadoop.hdds.utils.db.Table;
 import org.apache.hadoop.ozone.upgrade.ScmUpgradeActionForVersion;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
- * No-op upgrade action used only to verify that {@link ScmUpgradeActionForVersion} is scanned by
- * {@link org.apache.hadoop.hdds.upgrade.ScmUpgradeActionProvider} in tests.
+ * Removes the "finalizing in progress" mark, which may be left over in SCM meta table from pre-ZDU
+ * code. Deleting an absent key is a no-op, making the action idempotent.
  */
 @ScmUpgradeActionForVersion(version = ZDU)
-public class ZduScmUpgradeActionForTest implements ScmUpgradeAction {
+public class ClearFinalizingStateScmUpgradeAction implements ScmUpgradeAction {
+  private static final Logger LOG = LoggerFactory.getLogger(ClearFinalizingStateScmUpgradeAction.class);
+
+  // Orphan "finalizing in progress" mark written by pre-ZDU SCM code; removed on ZDU finalization.
+  private static final String LEGACY_FINALIZING_KEY = "#FINALIZING";
+
   @Override
-  public void execute(OzoneStorageContainerManager arg) {
+  public void execute(OzoneStorageContainerManager context) throws Exception {
+    StorageContainerManager scm = (StorageContainerManager) context;
+    Table<String, String> metaTable = scm.getScmMetadataStore().getMetaTable();
+    scm.getScmHAManager().getDBTransactionBuffer().removeFromBuffer(metaTable, LEGACY_FINALIZING_KEY);
+    LOG.info("Removed leftover SCM finalizing mark {} during ZDU finalization.", LEGACY_FINALIZING_KEY);
   }
 }

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/ha/TestSCMStateMachine.java

@@ -17,15 +17,24 @@
 
 package org.apache.hadoop.hdds.scm.ha;
 
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import org.apache.hadoop.hdds.protocol.proto.SCMRatisProtocol.RequestType;
 import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMMetrics;
+import org.apache.hadoop.hdds.scm.ha.invoker.ScmInvoker;
 import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
 import org.apache.hadoop.hdds.utils.TransactionInfo;
+import org.apache.hadoop.ozone.upgrade.UpgradeException;
 import org.apache.ratis.proto.RaftProtos;
+import org.apache.ratis.proto.RaftProtos.LogEntryProto;
+import org.apache.ratis.proto.RaftProtos.StateMachineLogEntryProto;
 import org.apache.ratis.server.protocol.TermIndex;
+import org.apache.ratis.statemachine.TransactionContext;
+import org.apache.ratis.util.ExitUtils;
 import org.junit.jupiter.api.Test;
 
 /**
@@ -42,11 +51,56 @@ public void testRatisEventsRecording() throws Exception {
     SCMHADBTransactionBuffer buffer = mock(SCMHADBTransactionBuffer.class);
     when(buffer.getLatestTrxInfo()).thenReturn(TransactionInfo.valueOf(TermIndex.valueOf(0, 0)));
 
+    // The contents of the state machine are mocked, so SCMStateMachine#close is a no-op.
     SCMStateMachine stateMachine = new SCMStateMachine(scm, buffer);
-
     stateMachine.notifyConfigurationChanged(1, 1, RaftProtos.RaftConfigurationProto.getDefaultInstance());
+
     assertTrue(metrics.getRatisEvents().contains("Configuration changed at term index"));
 
     metrics.unRegister();
   }
+
+  /**
+   * A finalization step that throws an UpgradeException (an IOException, not an SCMException) must
+   * crash SCM rather than be returned to the Ratis client. UpgradeException skips the inner
+   * catch (SCMException) and hits the outer catch (Exception) -> ExitUtils.terminate.
+   */
+  @Test
+  public void testUpgradeExceptionDuringApplyTerminates() throws Exception {
+    ExitUtils.disableSystemExit();
+
+    StorageContainerManager scm = mock(StorageContainerManager.class);
+    SCMMetrics metrics = SCMMetrics.create();
+    when(scm.getMetrics()).thenReturn(metrics);
+
+    SCMHADBTransactionBuffer buffer = mock(SCMHADBTransactionBuffer.class);
+    when(buffer.getLatestTrxInfo()).thenReturn(TransactionInfo.valueOf(TermIndex.valueOf(0, 0)));
+
+    // The contents of the state machine are mocked, so SCMStateMachine#close is a no-op.
+    SCMStateMachine stateMachine = new SCMStateMachine(scm, buffer);
+    ScmInvoker<?> invoker = mock(ScmInvoker.class);
+    when(invoker.invokeLocal(any(), any())).thenThrow(
+        new UpgradeException(UpgradeException.ResultCodes.FINALIZE_UPGRADE_ACTION_FAILED));
+    stateMachine.registerInvoker(RequestType.FINALIZE, invoker);
+
+    SCMRatisRequest request = SCMRatisRequest.of(
+        RequestType.FINALIZE, "finalize", new Class<?>[]{});
+    StateMachineLogEntryProto smLogEntry = StateMachineLogEntryProto.newBuilder()
+        .setLogData(request.encode().getContent())
+        .build();
+    LogEntryProto logEntry = LogEntryProto.newBuilder()
+        .setTerm(1)
+        .setIndex(1)
+        .setStateMachineLogEntry(smLogEntry)
+        .build();
+    TransactionContext trx = mock(TransactionContext.class);
+    when(trx.getStateMachineLogEntry()).thenReturn(smLogEntry);
+    when(trx.getLogEntry()).thenReturn(logEntry);
+
+    // terminate throws ExitException when system exit is disabled
+    assertThrows(ExitUtils.ExitException.class,
+        () -> stateMachine.applyTransaction(trx));
+
+    metrics.unRegister();
+  }
 }

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/upgrade/TestClearFinalizingStateScmUpgradeAction.java

@@ -0,0 +1,83 @@
+/*
+ * 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
+ *
+ *      http://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.hadoop.hdds.scm.server.upgrade;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.apache.hadoop.hdds.scm.ha.SCMHAManager;
+import org.apache.hadoop.hdds.scm.metadata.DBTransactionBuffer;
+import org.apache.hadoop.hdds.scm.metadata.SCMMetadataStore;
+import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
+import org.apache.hadoop.hdds.utils.db.Table;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit tests for {@link ClearFinalizingStateScmUpgradeAction}.
+ */
+class TestClearFinalizingStateScmUpgradeAction {
+
+  private static final String LEGACY_FINALIZING_KEY = "#FINALIZING";
+
+  @Test
+  void testRemovesFinalizingKey() throws Exception {
+    @SuppressWarnings("unchecked")
+    Table<String, String> metaTable = mock(Table.class);
+    DBTransactionBuffer buffer = mock(DBTransactionBuffer.class);
+
+    SCMMetadataStore metadataStore = mock(SCMMetadataStore.class);
+    when(metadataStore.getMetaTable()).thenReturn(metaTable);
+
+    SCMHAManager haManager = mock(SCMHAManager.class);
+    when(haManager.getDBTransactionBuffer()).thenReturn(buffer);
+
+    StorageContainerManager scm = mock(StorageContainerManager.class);
+    when(scm.getScmMetadataStore()).thenReturn(metadataStore);
+    when(scm.getScmHAManager()).thenReturn(haManager);
+
+    new ClearFinalizingStateScmUpgradeAction().execute(scm);
+
+    verify(buffer).removeFromBuffer(metaTable, LEGACY_FINALIZING_KEY);
+  }
+
+  @Test
+  void testIdempotent() throws Exception {
+    // removeFromBuffer on an absent key is a no-op; call execute twice and expect no exception.
+    @SuppressWarnings("unchecked")
+    Table<String, String> metaTable = mock(Table.class);
+    DBTransactionBuffer buffer = mock(DBTransactionBuffer.class);
+
+    SCMMetadataStore metadataStore = mock(SCMMetadataStore.class);
+    when(metadataStore.getMetaTable()).thenReturn(metaTable);
+
+    SCMHAManager haManager = mock(SCMHAManager.class);
+    when(haManager.getDBTransactionBuffer()).thenReturn(buffer);
+
+    StorageContainerManager scm = mock(StorageContainerManager.class);
+    when(scm.getScmMetadataStore()).thenReturn(metadataStore);
+    when(scm.getScmHAManager()).thenReturn(haManager);
+
+    ClearFinalizingStateScmUpgradeAction action = new ClearFinalizingStateScmUpgradeAction();
+    action.execute(scm);
+    action.execute(scm);
+
+    // Called twice, once per execution.
+    verify(buffer, org.mockito.Mockito.times(2)).removeFromBuffer(metaTable, LEGACY_FINALIZING_KEY);
+  }
+}

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/upgrade/TestScmVersionManager.java

@@ -140,7 +140,7 @@ public void testClasspathScanDiscoversUpgradeActions() throws Exception {
       ScmUpgradeAction upgradeAction = versionManager.getUpgradeActionsForTesting().get(DATANODE_SCHEMA_V2);
       assertInstanceOf(ScmOnFinalizeActionForDatanodeSchemaV2.class, upgradeAction);
       ScmUpgradeAction zduAction = versionManager.getUpgradeActionsForTesting().get(HDDSVersion.ZDU);
-      assertInstanceOf(ZduScmUpgradeActionForTest.class, zduAction);
+      assertInstanceOf(ClearFinalizingStateScmUpgradeAction.class, zduAction);
     }
 
     try (ScmVersionManager versionManager = createManager(HDDSVersion.SOFTWARE_VERSION.serialize(),
@@ -149,7 +149,7 @@ public void testClasspathScanDiscoversUpgradeActions() throws Exception {
       ScmUpgradeAction upgradeAction = versionManager.getUpgradeActionsForTesting().get(DATANODE_SCHEMA_V2);
       assertInstanceOf(ScmOnFinalizeActionForDatanodeSchemaV2.class, upgradeAction);
       ScmUpgradeAction zduAction = versionManager.getUpgradeActionsForTesting().get(HDDSVersion.ZDU);
-      assertInstanceOf(ZduScmUpgradeActionForTest.class, zduAction);
+      assertInstanceOf(ClearFinalizingStateScmUpgradeAction.class, zduAction);
     }
   }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/upgrade/ClearPreparedStateOmUpgradeAction.java

@@ -0,0 +1,60 @@
+/*
+ * 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
+ *
+ *      http://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.hadoop.ozone.om.upgrade;
+
+import static org.apache.hadoop.ozone.OzoneManagerVersion.ZDU;
+
+import java.io.File;
+import java.io.IOException;
+import org.apache.hadoop.hdds.server.ServerUtils;
+import org.apache.hadoop.ozone.common.Storage;
+import org.apache.hadoop.ozone.om.OzoneManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Removes leftover OM "prepare for upgrade" state written by pre-ZDU code.
+ * It is idempotent (delete-if-exists), so it is a no-op on clusters that were never prepared and
+ * on fresh ZDU clusters that never finalize.
+ */
+@OmUpgradeActionForVersion(version = ZDU)
+public class ClearPreparedStateOmUpgradeAction implements OmUpgradeAction {
+  private static final Logger LOG = LoggerFactory.getLogger(ClearPreparedStateOmUpgradeAction.class);
+
+  // On-disk marker written by pre-ZDU OM prepare code; removed on ZDU finalization.
+  private static final String LEGACY_PREPARE_MARKER = "prepareMarker";
+  // transactionInfoTable key written by pre-ZDU OM prepare code.
+  private static final String LEGACY_PREPARE_MARKER_KEY = "#PREPAREDINFO";
+
+  @Override
+  public void execute(OzoneManager om) throws Exception {
+    // Reproduces the removed getPrepareMarkerFile() logic: <metadata dir>/current/prepareMarker.
+    File markerDir = new File(ServerUtils.getOzoneMetaDirPath(om.getConfiguration()),
+        Storage.STORAGE_DIR_CURRENT);
+    File marker = new File(markerDir, LEGACY_PREPARE_MARKER);
+    if (marker.exists()) {
+      if (!marker.delete()) {
+        throw new IOException("Failed to delete leftover OM prepare marker file " + marker);
+      }
+      LOG.info("Deleted leftover OM prepare marker file {}", marker);
+    }
+
+    // Direct RocksDB delete of the orphan prepare key, mirroring the removed startup cleanup.
+    om.getMetadataManager().getTransactionInfoTable().delete(LEGACY_PREPARE_MARKER_KEY);
+  }
+}