dombizita commented on code in PR #10832: URL: https://github.com/apache/ozone/pull/10832#discussion_r3629830182
########## hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOMUpgradeFinalization.java: ########## @@ -0,0 +1,190 @@ +/* + * 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; + +import static org.apache.hadoop.ozone.OzoneConsts.APPARENT_VERSION_KEY; +import static org.apache.hadoop.ozone.om.TestOzoneManagerHAWithStoppedNodes.createKey; +import static org.apache.hadoop.ozone.om.upgrade.OMLayoutFeature.INITIAL_VERSION; +import static org.apache.ozone.test.GenericTestUtils.waitFor; +import static org.apache.ozone.test.OzoneTestBase.uniqueObjectName; +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; + +import java.io.IOException; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.conf.StorageUnit; +import org.apache.hadoop.ozone.MiniOzoneCluster; +import org.apache.hadoop.ozone.MiniOzoneHAClusterImpl; +import org.apache.hadoop.ozone.OzoneManagerVersion; +import org.apache.hadoop.ozone.client.ObjectStore; +import org.apache.hadoop.ozone.client.OzoneBucket; +import org.apache.hadoop.ozone.client.OzoneClient; +import org.apache.hadoop.ozone.client.OzoneVolume; +import org.apache.hadoop.ozone.conf.OMClientConfig; +import org.apache.hadoop.ozone.om.protocol.OzoneManagerProtocol; +import org.apache.hadoop.ozone.om.ratis.OzoneManagerRatisServer; +import org.apache.hadoop.ozone.upgrade.RatisBasedVersionManager; +import org.apache.ozone.test.GenericTestUtils.LogCapturer; +import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Tests for OM upgrade finalization. + */ +class TestOMUpgradeFinalization { + + private static final Logger LOG = LoggerFactory.getLogger(TestOMUpgradeFinalization.class); + + // Force Ratis to take a snapshot and purge logs after only a few + // transactions, so a lagging OM has to install a snapshot on startup. + private static final long SNAPSHOT_THRESHOLD = 5; + private static final int LOG_PURGE_GAP = 5; + + /** + * A lagging OM that has to install a Ratis snapshot should finalize + * from that snapshot. + * When finalization runs on the active OMs while one OM is down, the down OM + * cannot replay the finalization from the (purged) logs, so it must pick up + * the finalized version from the checkpoint it downloads from the leader. + */ + @Test + void testFinalizationFromSnapshot() throws Exception { + OzoneConfiguration conf = new OzoneConfiguration(); + conf.set(OMConfigKeys.OZONE_OM_UPGRADE_FINALIZATION_CHECK_INTERVAL, "10ms"); + // Force frequent snapshots and log purge so the inactive OM must install a + // snapshot from the leader (rather than replay logs) when it starts. + conf.setInt(OMConfigKeys.OZONE_OM_RATIS_LOG_PURGE_GAP, LOG_PURGE_GAP); + conf.setStorageSize(OMConfigKeys.OZONE_OM_RATIS_SEGMENT_SIZE_KEY, 16, StorageUnit.KB); + conf.setStorageSize(OMConfigKeys.OZONE_OM_RATIS_SEGMENT_PREALLOCATED_SIZE_KEY, 16, StorageUnit.KB); + conf.setLong(OMConfigKeys.OZONE_OM_RATIS_SNAPSHOT_AUTO_TRIGGER_THRESHOLD_KEY, SNAPSHOT_THRESHOLD); + + // Use a short client RPC timeout so the first request fails over quickly to + // a responsive OM instead of blocking on the 15-minute default if it lands on the inactive one. + OMClientConfig clientConfig = conf.getObject(OMClientConfig.class); + clientConfig.setRpcTimeOut(TimeUnit.SECONDS.toMillis(5)); + conf.setFromObject(clientConfig); + + LogCapturer versionManagerLogCapture = LogCapturer.captureLogs(RatisBasedVersionManager.class); + LogCapturer omLogCapture = LogCapturer.captureLogs(OzoneManager.class); + + try (MiniOzoneHAClusterImpl cluster = newCluster(conf)) { + LOG.info("Waiting for cluster to be ready"); + cluster.waitForClusterToBeReady(); + OzoneManager followerOM = cluster.getInactiveOM().next(); + LOG.info("Cluster ready with inactive OM: {}", followerOM); + + try (OzoneClient client = cluster.newClient()) { + LOG.info("Client created"); + ObjectStore objectStore = client.getObjectStore(); + LOG.info("Object store created"); + + // The active OMs start un-finalized: the apparent version is the initial + // version and no version has been written to the DB yet. + for (OzoneManager om : cluster.getOzoneManagersList()) { + if (cluster.isOMActive(om.getOMNodeId())) { + assertEquals(INITIAL_VERSION, om.getVersionManager().getApparentVersion()); + assertNull(om.getMetadataManager().getMetaTable().get(APPARENT_VERSION_KEY)); + } + } + + // Finalize the running (active) OMs. + LOG.info("Finalizing OMs"); + OzoneManagerProtocol omClient = objectStore.getClientProxy().getOzoneManagerClient(); + omClient.finalizeUpgrade(); + OMUpgradeTestUtils.waitForFinalization(omClient); + LOG.info("Finalized active OMs"); + + // Write enough keys to advance the log index well past the purge gap so + // the leader's logs covering finalization are purged. This forces the + // follower to install a snapshot instead of replaying the logs. + OzoneManagerRatisServer leaderRatisServer = cluster.getOMLeader().getOmRatisServer(); Review Comment: Can we be sure that we have on OM leader at this time? `cluster.getOMLeader()` is returning `null` if there is not exactly one ready leader. If we decide to add a safety check it can be done at one place, it's also used on line 135. ########## hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOMUpgradeFinalization.java: ########## @@ -0,0 +1,190 @@ +/* + * 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; + +import static org.apache.hadoop.ozone.OzoneConsts.APPARENT_VERSION_KEY; +import static org.apache.hadoop.ozone.om.TestOzoneManagerHAWithStoppedNodes.createKey; +import static org.apache.hadoop.ozone.om.upgrade.OMLayoutFeature.INITIAL_VERSION; +import static org.apache.ozone.test.GenericTestUtils.waitFor; +import static org.apache.ozone.test.OzoneTestBase.uniqueObjectName; +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; + +import java.io.IOException; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.conf.StorageUnit; +import org.apache.hadoop.ozone.MiniOzoneCluster; +import org.apache.hadoop.ozone.MiniOzoneHAClusterImpl; +import org.apache.hadoop.ozone.OzoneManagerVersion; +import org.apache.hadoop.ozone.client.ObjectStore; +import org.apache.hadoop.ozone.client.OzoneBucket; +import org.apache.hadoop.ozone.client.OzoneClient; +import org.apache.hadoop.ozone.client.OzoneVolume; +import org.apache.hadoop.ozone.conf.OMClientConfig; +import org.apache.hadoop.ozone.om.protocol.OzoneManagerProtocol; +import org.apache.hadoop.ozone.om.ratis.OzoneManagerRatisServer; +import org.apache.hadoop.ozone.upgrade.RatisBasedVersionManager; +import org.apache.ozone.test.GenericTestUtils.LogCapturer; +import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Tests for OM upgrade finalization. + */ +class TestOMUpgradeFinalization { + + private static final Logger LOG = LoggerFactory.getLogger(TestOMUpgradeFinalization.class); + + // Force Ratis to take a snapshot and purge logs after only a few + // transactions, so a lagging OM has to install a snapshot on startup. + private static final long SNAPSHOT_THRESHOLD = 5; + private static final int LOG_PURGE_GAP = 5; + + /** + * A lagging OM that has to install a Ratis snapshot should finalize + * from that snapshot. + * When finalization runs on the active OMs while one OM is down, the down OM + * cannot replay the finalization from the (purged) logs, so it must pick up + * the finalized version from the checkpoint it downloads from the leader. + */ + @Test + void testFinalizationFromSnapshot() throws Exception { + OzoneConfiguration conf = new OzoneConfiguration(); + conf.set(OMConfigKeys.OZONE_OM_UPGRADE_FINALIZATION_CHECK_INTERVAL, "10ms"); + // Force frequent snapshots and log purge so the inactive OM must install a + // snapshot from the leader (rather than replay logs) when it starts. + conf.setInt(OMConfigKeys.OZONE_OM_RATIS_LOG_PURGE_GAP, LOG_PURGE_GAP); + conf.setStorageSize(OMConfigKeys.OZONE_OM_RATIS_SEGMENT_SIZE_KEY, 16, StorageUnit.KB); + conf.setStorageSize(OMConfigKeys.OZONE_OM_RATIS_SEGMENT_PREALLOCATED_SIZE_KEY, 16, StorageUnit.KB); + conf.setLong(OMConfigKeys.OZONE_OM_RATIS_SNAPSHOT_AUTO_TRIGGER_THRESHOLD_KEY, SNAPSHOT_THRESHOLD); + + // Use a short client RPC timeout so the first request fails over quickly to + // a responsive OM instead of blocking on the 15-minute default if it lands on the inactive one. + OMClientConfig clientConfig = conf.getObject(OMClientConfig.class); + clientConfig.setRpcTimeOut(TimeUnit.SECONDS.toMillis(5)); + conf.setFromObject(clientConfig); + + LogCapturer versionManagerLogCapture = LogCapturer.captureLogs(RatisBasedVersionManager.class); + LogCapturer omLogCapture = LogCapturer.captureLogs(OzoneManager.class); Review Comment: nit: maybe it's worth to add `.stopCapturing()` to the log capturers. ########## hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOMUpgradeFinalization.java: ########## @@ -0,0 +1,190 @@ +/* + * 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; + +import static org.apache.hadoop.ozone.OzoneConsts.APPARENT_VERSION_KEY; +import static org.apache.hadoop.ozone.om.TestOzoneManagerHAWithStoppedNodes.createKey; +import static org.apache.hadoop.ozone.om.upgrade.OMLayoutFeature.INITIAL_VERSION; +import static org.apache.ozone.test.GenericTestUtils.waitFor; +import static org.apache.ozone.test.OzoneTestBase.uniqueObjectName; +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; + +import java.io.IOException; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.conf.StorageUnit; +import org.apache.hadoop.ozone.MiniOzoneCluster; +import org.apache.hadoop.ozone.MiniOzoneHAClusterImpl; +import org.apache.hadoop.ozone.OzoneManagerVersion; +import org.apache.hadoop.ozone.client.ObjectStore; +import org.apache.hadoop.ozone.client.OzoneBucket; +import org.apache.hadoop.ozone.client.OzoneClient; +import org.apache.hadoop.ozone.client.OzoneVolume; +import org.apache.hadoop.ozone.conf.OMClientConfig; +import org.apache.hadoop.ozone.om.protocol.OzoneManagerProtocol; +import org.apache.hadoop.ozone.om.ratis.OzoneManagerRatisServer; +import org.apache.hadoop.ozone.upgrade.RatisBasedVersionManager; +import org.apache.ozone.test.GenericTestUtils.LogCapturer; +import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Tests for OM upgrade finalization. + */ +class TestOMUpgradeFinalization { + + private static final Logger LOG = LoggerFactory.getLogger(TestOMUpgradeFinalization.class); + + // Force Ratis to take a snapshot and purge logs after only a few + // transactions, so a lagging OM has to install a snapshot on startup. + private static final long SNAPSHOT_THRESHOLD = 5; + private static final int LOG_PURGE_GAP = 5; + + /** + * A lagging OM that has to install a Ratis snapshot should finalize + * from that snapshot. + * When finalization runs on the active OMs while one OM is down, the down OM + * cannot replay the finalization from the (purged) logs, so it must pick up + * the finalized version from the checkpoint it downloads from the leader. + */ + @Test + void testFinalizationFromSnapshot() throws Exception { + OzoneConfiguration conf = new OzoneConfiguration(); + conf.set(OMConfigKeys.OZONE_OM_UPGRADE_FINALIZATION_CHECK_INTERVAL, "10ms"); + // Force frequent snapshots and log purge so the inactive OM must install a + // snapshot from the leader (rather than replay logs) when it starts. + conf.setInt(OMConfigKeys.OZONE_OM_RATIS_LOG_PURGE_GAP, LOG_PURGE_GAP); + conf.setStorageSize(OMConfigKeys.OZONE_OM_RATIS_SEGMENT_SIZE_KEY, 16, StorageUnit.KB); + conf.setStorageSize(OMConfigKeys.OZONE_OM_RATIS_SEGMENT_PREALLOCATED_SIZE_KEY, 16, StorageUnit.KB); + conf.setLong(OMConfigKeys.OZONE_OM_RATIS_SNAPSHOT_AUTO_TRIGGER_THRESHOLD_KEY, SNAPSHOT_THRESHOLD); + + // Use a short client RPC timeout so the first request fails over quickly to + // a responsive OM instead of blocking on the 15-minute default if it lands on the inactive one. + OMClientConfig clientConfig = conf.getObject(OMClientConfig.class); + clientConfig.setRpcTimeOut(TimeUnit.SECONDS.toMillis(5)); + conf.setFromObject(clientConfig); + + LogCapturer versionManagerLogCapture = LogCapturer.captureLogs(RatisBasedVersionManager.class); + LogCapturer omLogCapture = LogCapturer.captureLogs(OzoneManager.class); + + try (MiniOzoneHAClusterImpl cluster = newCluster(conf)) { + LOG.info("Waiting for cluster to be ready"); + cluster.waitForClusterToBeReady(); + OzoneManager followerOM = cluster.getInactiveOM().next(); + LOG.info("Cluster ready with inactive OM: {}", followerOM); + + try (OzoneClient client = cluster.newClient()) { + LOG.info("Client created"); + ObjectStore objectStore = client.getObjectStore(); + LOG.info("Object store created"); + + // The active OMs start un-finalized: the apparent version is the initial + // version and no version has been written to the DB yet. + for (OzoneManager om : cluster.getOzoneManagersList()) { + if (cluster.isOMActive(om.getOMNodeId())) { + assertEquals(INITIAL_VERSION, om.getVersionManager().getApparentVersion()); + assertNull(om.getMetadataManager().getMetaTable().get(APPARENT_VERSION_KEY)); + } + } + + // Finalize the running (active) OMs. + LOG.info("Finalizing OMs"); + OzoneManagerProtocol omClient = objectStore.getClientProxy().getOzoneManagerClient(); + omClient.finalizeUpgrade(); + OMUpgradeTestUtils.waitForFinalization(omClient); + LOG.info("Finalized active OMs"); + + // Write enough keys to advance the log index well past the purge gap so + // the leader's logs covering finalization are purged. This forces the + // follower to install a snapshot instead of replaying the logs. + OzoneManagerRatisServer leaderRatisServer = cluster.getOMLeader().getOmRatisServer(); + OzoneBucket bucket = createVolumeAndBucket(objectStore); + long targetLogIndex = leaderRatisServer.getLastAppliedTermIndex().getIndex() + 100; + LOG.info("Writing keys to advance log index"); + writeKeysToIncreaseLogIndex(leaderRatisServer, targetLogIndex, bucket); + + // Start the inactive OM. It downloads the leader's checkpoint (which + // already contains the finalized apparent version) and finalizes from it. + LOG.info("Starting inactive OM"); + cluster.startInactiveOM(followerOM.getOMNodeId()); + + // Wait for the follower to finish installing the snapshot and catch up + // to the leader before reading its DB and shutting down, so shutdown + // does not race with the tail of the snapshot install. + LOG.info("Inactive OM started, waiting to catch up with leader"); + long leaderIndex = cluster.getOMLeader().getOmRatisServer().getLastAppliedTermIndex().getIndex(); + waitFor(() -> { + boolean running = followerOM.isRunning(); + // Make sure the snapshot install finishes completely so it does not block cluster shutdown on test cleanup. + boolean snapshotFinished = omLogCapture.getOutput().contains("Install Checkpoint is finished"); + long followerIndex = followerOM.getOmRatisServer().getLastAppliedTermIndex().getIndex(); + LOG.info("Waiting for follower to catch up with leader. Running? {} Snapshot finished? {}" + + " Follower index = {} leader index = {}", + running, snapshotFinished, followerIndex, leaderIndex); + return running && snapshotFinished && followerIndex >= leaderIndex; + }, 1000, 60000); + LOG.info("Inactive OM has caught up with leader, checking finalization state"); + + // The follower finalizes to the software version, picked up from the installed snapshot. + assertEquals(OzoneManagerVersion.SOFTWARE_VERSION, followerOM.getVersionManager().getApparentVersion()); + + String dbVersion = followerOM.getMetadataManager().getMetaTable().get(APPARENT_VERSION_KEY); + assertNotNull(dbVersion); + assertEquals(OzoneManagerVersion.SOFTWARE_VERSION.serialize(), Integer.parseInt(dbVersion)); + + // Confirm finalization happened via snapshot install, not log replay. + assertThat(versionManagerLogCapture.getOutput()).contains("New snapshot received with higher apparent version"); + } + } + } + + private static MiniOzoneHAClusterImpl newCluster(OzoneConfiguration conf) + throws IOException { + conf.setInt(OMStorage.TESTING_INIT_APPARENT_VERSION_KEY, INITIAL_VERSION.serialize()); + MiniOzoneHAClusterImpl.Builder builder = MiniOzoneCluster.newHABuilder(conf); + builder.setOMServiceId(UUID.randomUUID().toString()) + .setNumOfOzoneManagers(3) + .setNumOfActiveOMs(2) + .setNumDatanodes(1); + return builder.build(); + } + + private static OzoneBucket createVolumeAndBucket(ObjectStore objectStore) Review Comment: Can't `TestDataUtil.createVolumeAndBucket(client)` be re-used here? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
