hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/TestSnapshotChain.java - ozone - Git at Google

 /*
  * 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
  * <p>
  * http://www.apache.org/licenses/LICENSE-2.0
  * <p>
  * 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.snapshot;

 import com.google.common.collect.ImmutableMap;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.utils.db.Table;
 import org.apache.hadoop.ozone.om.OMMetadataManager;
 import org.apache.hadoop.ozone.om.OmMetadataManagerImpl;
 import org.apache.hadoop.ozone.om.SnapshotChainManager;
 import org.apache.hadoop.ozone.om.helpers.SnapshotInfo;
 import org.apache.hadoop.util.Time;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Named;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
 import org.junit.jupiter.params.provider.ValueSource;

 import java.io.File;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.UUID;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;

 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_DB_DIRS;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;

 /**
  * Tests SnapshotChain that stores in chronological order
  * Ozone object storage snapshots.  There exist two types
  * of chains provided by the SnapshotChainManager
  * i.) path based snapshots - a list of snapshots taken for a given bucket
  * ii.) global snapshots - a list of every snapshot taken in chrono order
  */
 public class TestSnapshotChain {
   private OMMetadataManager omMetadataManager;
   private Map<UUID, SnapshotInfo> snapshotIdToSnapshotInfoMap;
   private SnapshotChainManager chainManager;

   @TempDir
   private File folder;

   @BeforeEach
   public void setup() throws Exception {
     OzoneConfiguration conf = new OzoneConfiguration();
     conf.set(OZONE_OM_DB_DIRS, folder.toString());
     omMetadataManager = new OmMetadataManagerImpl(conf, null);
     snapshotIdToSnapshotInfoMap = new HashMap<>();
     chainManager = new SnapshotChainManager(omMetadataManager);
   }

   private SnapshotInfo createSnapshotInfo(UUID snapshotID,
                                           UUID pathPrevID,
                                           UUID globalPrevID,
                                           long creationTime) {
     return new SnapshotInfo.Builder()
         .setSnapshotId(snapshotID)
         .setName("test")
         .setVolumeName("vol1")
         .setBucketName("bucket1")
         .setSnapshotStatus(SnapshotInfo.SnapshotStatus.SNAPSHOT_ACTIVE)
         .setCreationTime(creationTime)
         .setDeletionTime(-1L)
         .setPathPreviousSnapshotId(pathPrevID)
         .setGlobalPreviousSnapshotId(globalPrevID)
         .setSnapshotPath(String.join("/", "vol1", "bucket1"))
         .setCheckpointDir("checkpoint.testdir")
         .build();
   }

   private void deleteSnapshot(UUID snapshotID) throws IOException {
     SnapshotInfo sinfo = null;
     final String snapshotPath = "vol1/bucket1";
     // reset the next snapshotInfo.globalPreviousSnapshotID
     // to the previous in the entry to be deleted.
     if (chainManager.hasNextGlobalSnapshot(snapshotID)) {
       sinfo = snapshotIdToSnapshotInfoMap.get(
           chainManager.nextGlobalSnapshot(snapshotID));
       if (chainManager.hasPreviousGlobalSnapshot(snapshotID)) {
         sinfo.setGlobalPreviousSnapshotId(
             chainManager.previousGlobalSnapshot(snapshotID));
       } else {
         sinfo.setGlobalPreviousSnapshotId(null);
       }
       snapshotIdToSnapshotInfoMap.put(sinfo.getSnapshotId(), sinfo);
     }
     // reset the next snapshotInfo.pathPreviousSnapshotID
     // to the previous in the entry to be deleted.
     if (chainManager.hasNextPathSnapshot(snapshotPath, snapshotID)) {
       sinfo = snapshotIdToSnapshotInfoMap.get(
           chainManager.nextPathSnapshot(snapshotPath, snapshotID));

       if (chainManager.hasPreviousPathSnapshot(snapshotPath, snapshotID)) {
         sinfo.setPathPreviousSnapshotId(
             chainManager.previousPathSnapshot(snapshotPath, snapshotID));
       } else {
         sinfo.setPathPreviousSnapshotId(null);
       }
       snapshotIdToSnapshotInfoMap.put(sinfo.getSnapshotId(), sinfo);
     }

     UUID latestGlobalSnapshot = chainManager.getLatestGlobalSnapshotId();
     // append snapshot to the sinfos (the end).
     if (latestGlobalSnapshot != null) {
       sinfo = snapshotIdToSnapshotInfoMap.get(snapshotID);
       sinfo.setGlobalPreviousSnapshotId(latestGlobalSnapshot);
       UUID latestPathSnapshot = chainManager.getLatestPathSnapshotId(
           String.join("/", "vol1", "bucket1"));
       if (latestPathSnapshot != null) {
         sinfo.setPathPreviousSnapshotId(latestPathSnapshot);
       }
       snapshotIdToSnapshotInfoMap.put(snapshotID, sinfo);
     }
     chainManager.deleteSnapshot(snapshotIdToSnapshotInfoMap.get(snapshotID));
   }

   @Test
   public void testAddSnapshot() throws Exception {
     final String snapshotPath = "vol1/bucket1";
     // add three snapshots
     UUID snapshotID1 = UUID.randomUUID();
     UUID snapshotID2 = UUID.randomUUID();
     UUID snapshotID3 = UUID.randomUUID();

     ArrayList<UUID> snapshotIDs = new ArrayList<>();
     snapshotIDs.add(snapshotID1);
     snapshotIDs.add(snapshotID2);
     snapshotIDs.add(snapshotID3);

     UUID prevSnapshotID = null;
     List<Integer> times = Arrays.asList(1, 2, 0);
     int timeIdx = 0;
     // add 3 snapshots
     for (UUID snapshotID : snapshotIDs) {
       chainManager.addSnapshot(createSnapshotInfo(
           snapshotID,
           prevSnapshotID,
           prevSnapshotID, times.get(timeIdx++)));
       prevSnapshotID = snapshotID;
     }

     assertEquals(snapshotID3, chainManager.getLatestGlobalSnapshotId());
     assertEquals(snapshotID3, chainManager.getLatestPathSnapshotId(
         String.join("/", "vol1", "bucket1")));

     int i = 0;
     UUID curID = snapshotIDs.get(i);
     while (chainManager.hasNextGlobalSnapshot(curID)) {
       i++;
       assertEquals(snapshotIDs.get(i), chainManager.nextGlobalSnapshot(curID));
       curID = snapshotIDs.get(i);
     }

     curID = snapshotIDs.get(i);
     while (chainManager.hasPreviousGlobalSnapshot(curID)) {
       i--;
       assertEquals(snapshotIDs.get(i),
           chainManager.previousGlobalSnapshot(curID));
       curID = snapshotIDs.get(i);
     }

     i = 0;
     curID = snapshotIDs.get(i);
     while (chainManager.hasNextPathSnapshot(snapshotPath, curID)) {
       i++;
       assertEquals(snapshotIDs.get(i),
           chainManager.nextPathSnapshot(snapshotPath, curID));
       curID = snapshotIDs.get(i);
     }

     curID = snapshotIDs.get(i);
     while (chainManager.hasPreviousPathSnapshot(snapshotPath,
         curID)) {
       i--;
       assertEquals(snapshotIDs.get(i),
           chainManager.previousPathSnapshot(snapshotPath, curID));
       curID = snapshotIDs.get(i);
     }
   }

   @Test
   public void testDeleteSnapshot() throws Exception {
     // add three snapshots
     UUID snapshotID1 = UUID.randomUUID();
     UUID snapshotID2 = UUID.randomUUID();
     UUID snapshotID3 = UUID.randomUUID();

     ArrayList<UUID> snapshotIDs = new ArrayList<>();
     snapshotIDs.add(snapshotID1);
     snapshotIDs.add(snapshotID2);
     snapshotIDs.add(snapshotID3);

     UUID prevSnapshotID = null;
     List<Integer> times = Arrays.asList(1, 2, 0);
     int timeIdx = 0;
     // add 3 snapshots
     for (UUID snapshotID : snapshotIDs) {
       snapshotIdToSnapshotInfoMap.put(snapshotID,
           createSnapshotInfo(
               snapshotID,
               prevSnapshotID,
               prevSnapshotID, times.get(timeIdx++)));

       chainManager.addSnapshot(snapshotIdToSnapshotInfoMap.get(snapshotID));
       prevSnapshotID = snapshotID;
     }

     // delete snapshotID2
     // should have snapshots ID1 and ID3 in chain

     deleteSnapshot(snapshotID2);
     // start with first snapshot in snapshot chain
     assertFalse(chainManager.hasPreviousGlobalSnapshot(snapshotID1));
     assertTrue(chainManager.hasNextGlobalSnapshot(snapshotID1));
     assertFalse(chainManager.hasNextGlobalSnapshot(snapshotID3));
     assertEquals(snapshotID3,
         chainManager.nextGlobalSnapshot(snapshotID1));

     // add snapshotID2 and delete snapshotID1
     // should have snapshotID3 and snapshotID2
     deleteSnapshot(snapshotID1);
     chainManager.addSnapshot(snapshotIdToSnapshotInfoMap.get(snapshotID2));

     assertFalse(chainManager.hasPreviousGlobalSnapshot(snapshotID3));
     assertTrue(chainManager.hasNextGlobalSnapshot(snapshotID3));
     assertEquals(snapshotID2, chainManager.nextGlobalSnapshot(snapshotID3));
     assertFalse(chainManager.hasNextGlobalSnapshot(snapshotID2));
     assertEquals(snapshotID3, chainManager.previousGlobalSnapshot(snapshotID2));
   }

   @ParameterizedTest
   @ValueSource(booleans = {true, false})
   public void testChainFromLoadFromTable(boolean increasingTIme)
       throws Exception {
     Table<String, SnapshotInfo> snapshotInfo =
             omMetadataManager.getSnapshotInfoTable();

     // add two snapshots to the snapshotInfo
     UUID snapshotID1 = UUID.randomUUID();
     UUID snapshotID2 = UUID.randomUUID();
     List<UUID> snapshotIDs = new ArrayList<>();
     snapshotIDs.add(snapshotID1);
     snapshotIDs.add(snapshotID2);
     List<SnapshotInfo> snapshotInfoList = new ArrayList<>();

     UUID prevSnapshotID = null;
     long time = System.currentTimeMillis();
     for (UUID snapshotID : snapshotIDs) {
       SnapshotInfo snapInfo = createSnapshotInfo(snapshotID, prevSnapshotID,
           prevSnapshotID, increasingTIme ? time++ : time--);
       snapshotInfo.put(snapshotID.toString(), snapInfo);
       prevSnapshotID = snapshotID;
       snapshotInfoList.add(snapInfo);
     }

     chainManager = new SnapshotChainManager(omMetadataManager);
     assertFalse(chainManager.isSnapshotChainCorrupted());
     // check if snapshots loaded correctly from snapshotInfoTable
     assertEquals(snapshotID2, chainManager.getLatestGlobalSnapshotId());
     assertEquals(snapshotID2, chainManager.nextGlobalSnapshot(snapshotID1));
     assertEquals(snapshotID1, chainManager.previousPathSnapshot(String
         .join("/", "vol1", "bucket1"), snapshotID2));
     assertThrows(NoSuchElementException.class,
         () -> chainManager.nextGlobalSnapshot(snapshotID2));
     assertThrows(NoSuchElementException.class,
         () -> chainManager.previousPathSnapshot(String
             .join("/", "vol1", "bucket1"), snapshotID1));

     UUID snapshotID3 = UUID.randomUUID();
     SnapshotInfo snapshotInfo3 = createSnapshotInfo(snapshotID3, prevSnapshotID,
         prevSnapshotID, Time.now());
     // Add and delete snapshot to make sure snapshot chain is correct.
     chainManager.addSnapshot(snapshotInfo3);
     chainManager.deleteSnapshot(snapshotInfoList.get(0));
     assertEquals(snapshotID3, chainManager.getLatestGlobalSnapshotId());
     assertThrows(NoSuchElementException.class,
         () -> chainManager.nextGlobalSnapshot(snapshotID1));
   }

   private static Stream<? extends Arguments> invalidSnapshotChain() {
     List<UUID> nodes = IntStream.range(0, 5)
         .mapToObj(i -> UUID.randomUUID())
         .collect(Collectors.toList());
     return Stream.of(
         Arguments.of(nodes, Named.of("Disconnected Snapshot Chain",
             ImmutableMap.of(
                 nodes.get(1), nodes.get(0),
                 nodes.get(2), nodes.get(1),
                 nodes.get(4), nodes.get(3)))),
         Arguments.of(nodes, Named.of("Complete Cyclic Snapshot Chain",
             ImmutableMap.of(
                 nodes.get(0), nodes.get(4),
                 nodes.get(1), nodes.get(0),
                 nodes.get(2), nodes.get(1),
                 nodes.get(3), nodes.get(2),
                 nodes.get(4), nodes.get(3)))),
         Arguments.of(nodes, Named.of("Partial Cyclic Snapshot Chain",
             ImmutableMap.of(
                 nodes.get(0), nodes.get(3),
                 nodes.get(1), nodes.get(0),
                 nodes.get(2), nodes.get(1),
                 nodes.get(3), nodes.get(2),
                 nodes.get(4), nodes.get(3)))),
         Arguments.of(nodes, Named.of("Diverged Snapshot Chain",
             ImmutableMap.of(nodes.get(1), nodes.get(0),
                 nodes.get(2), nodes.get(1),
                 nodes.get(3), nodes.get(2),
                 nodes.get(4), nodes.get(2))))
     );
   }

   @ParameterizedTest
   @MethodSource("invalidSnapshotChain")
   public void testInvalidGlobalChainFromLoadFromTable(
       List<UUID> snapshotIDs, Map<UUID, UUID> snapshotChain) throws Exception {
     Table<String, SnapshotInfo> snapshotInfo =
         omMetadataManager.getSnapshotInfoTable();
     for (UUID snapshotID : snapshotIDs) {
       snapshotInfo.put(snapshotID.toString(),
           createSnapshotInfo(snapshotID, snapshotChain.get(snapshotID),
               snapshotChain.get(snapshotID), System.currentTimeMillis()));
     }
     chainManager = new SnapshotChainManager(omMetadataManager);

     assertTrue(chainManager.isSnapshotChainCorrupted());

     SnapshotInfo snapInfo = createSnapshotInfo(UUID.randomUUID(),
         UUID.randomUUID(),
         UUID.randomUUID(),
         System.currentTimeMillis());
     IOException createException = assertThrows(IOException.class,
             () -> chainManager.addSnapshot(snapInfo));
     assertEquals("Snapshot chain is corrupted.", createException.getMessage());
     if (!snapshotIDs.isEmpty()) {
       IOException deleteException =
           assertThrows(IOException.class,
               () -> chainManager.deleteSnapshot(
                   snapshotInfo.get(snapshotIDs.get(0).toString())));
       assertEquals("Snapshot chain is corrupted.",
           deleteException.getMessage());
     }
   }

   @ParameterizedTest
   @MethodSource("invalidSnapshotChain")
   public void testInvalidChainFromLoadFromTable(List<UUID> snapshotIDs,
       Map<UUID, UUID> snapshotChain) throws Exception {
     Table<String, SnapshotInfo> snapshotInfo =
         omMetadataManager.getSnapshotInfoTable();
     UUID prevSnapshotId = null;
     for (UUID snapshotID : snapshotIDs) {
       snapshotInfo.put(snapshotID.toString(),
           createSnapshotInfo(snapshotID, snapshotChain.get(snapshotID),
               prevSnapshotId, System.currentTimeMillis()));
       prevSnapshotId = snapshotID;
     }
     chainManager = new SnapshotChainManager(omMetadataManager);
     assertTrue(chainManager.isSnapshotChainCorrupted());

     SnapshotInfo snapInfo = createSnapshotInfo(UUID.randomUUID(),
         UUID.randomUUID(),
         UUID.randomUUID(),
         System.currentTimeMillis());
     IOException createException = assertThrows(IOException.class,
             () -> chainManager.addSnapshot(snapInfo));
     assertEquals("Snapshot chain is corrupted.", createException.getMessage());
     if (!snapshotIDs.isEmpty()) {
       IOException deleteException =
           assertThrows(IOException.class,
               () -> chainManager.deleteSnapshot(
                   snapshotInfo.get(snapshotIDs.get(0).toString())));
       assertEquals("Snapshot chain is corrupted.",
           deleteException.getMessage());
     }
   }
 }
	/*
	* 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
	* <p>
	* http://www.apache.org/licenses/LICENSE-2.0
	* <p>
	* 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.snapshot;

	import com.google.common.collect.ImmutableMap;
	import org.apache.hadoop.hdds.conf.OzoneConfiguration;
	import org.apache.hadoop.hdds.utils.db.Table;
	import org.apache.hadoop.ozone.om.OMMetadataManager;
	import org.apache.hadoop.ozone.om.OmMetadataManagerImpl;
	import org.apache.hadoop.ozone.om.SnapshotChainManager;
	import org.apache.hadoop.ozone.om.helpers.SnapshotInfo;
	import org.apache.hadoop.util.Time;
	import org.junit.jupiter.api.BeforeEach;
	import org.junit.jupiter.api.Named;
	import org.junit.jupiter.api.Test;
	import org.junit.jupiter.api.io.TempDir;
	import org.junit.jupiter.params.ParameterizedTest;
	import org.junit.jupiter.params.provider.Arguments;
	import org.junit.jupiter.params.provider.MethodSource;
	import org.junit.jupiter.params.provider.ValueSource;

	import java.io.File;
	import java.io.IOException;
	import java.util.ArrayList;
	import java.util.Arrays;
	import java.util.HashMap;
	import java.util.List;
	import java.util.Map;
	import java.util.NoSuchElementException;
	import java.util.UUID;
	import java.util.stream.Collectors;
	import java.util.stream.IntStream;
	import java.util.stream.Stream;

	import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_DB_DIRS;
	import static org.junit.jupiter.api.Assertions.assertEquals;
	import static org.junit.jupiter.api.Assertions.assertFalse;
	import static org.junit.jupiter.api.Assertions.assertThrows;
	import static org.junit.jupiter.api.Assertions.assertTrue;

	/**
	* Tests SnapshotChain that stores in chronological order
	* Ozone object storage snapshots. There exist two types
	* of chains provided by the SnapshotChainManager
	* i.) path based snapshots - a list of snapshots taken for a given bucket
	* ii.) global snapshots - a list of every snapshot taken in chrono order
	*/
	public class TestSnapshotChain {
	private OMMetadataManager omMetadataManager;
	private Map<UUID, SnapshotInfo> snapshotIdToSnapshotInfoMap;
	private SnapshotChainManager chainManager;

	@TempDir
	private File folder;

	@BeforeEach
	public void setup() throws Exception {
	OzoneConfiguration conf = new OzoneConfiguration();
	conf.set(OZONE_OM_DB_DIRS, folder.toString());
	omMetadataManager = new OmMetadataManagerImpl(conf, null);
	snapshotIdToSnapshotInfoMap = new HashMap<>();
	chainManager = new SnapshotChainManager(omMetadataManager);
	}

	private SnapshotInfo createSnapshotInfo(UUID snapshotID,
	UUID pathPrevID,
	UUID globalPrevID,
	long creationTime) {
	return new SnapshotInfo.Builder()
	.setSnapshotId(snapshotID)
	.setName("test")
	.setVolumeName("vol1")
	.setBucketName("bucket1")
	.setSnapshotStatus(SnapshotInfo.SnapshotStatus.SNAPSHOT_ACTIVE)
	.setCreationTime(creationTime)
	.setDeletionTime(-1L)
	.setPathPreviousSnapshotId(pathPrevID)
	.setGlobalPreviousSnapshotId(globalPrevID)
	.setSnapshotPath(String.join("/", "vol1", "bucket1"))
	.setCheckpointDir("checkpoint.testdir")
	.build();
	}

	private void deleteSnapshot(UUID snapshotID) throws IOException {
	SnapshotInfo sinfo = null;
	final String snapshotPath = "vol1/bucket1";
	// reset the next snapshotInfo.globalPreviousSnapshotID
	// to the previous in the entry to be deleted.
	if (chainManager.hasNextGlobalSnapshot(snapshotID)) {
	sinfo = snapshotIdToSnapshotInfoMap.get(
	chainManager.nextGlobalSnapshot(snapshotID));
	if (chainManager.hasPreviousGlobalSnapshot(snapshotID)) {
	sinfo.setGlobalPreviousSnapshotId(
	chainManager.previousGlobalSnapshot(snapshotID));
	} else {
	sinfo.setGlobalPreviousSnapshotId(null);
	}
	snapshotIdToSnapshotInfoMap.put(sinfo.getSnapshotId(), sinfo);
	}
	// reset the next snapshotInfo.pathPreviousSnapshotID
	// to the previous in the entry to be deleted.
	if (chainManager.hasNextPathSnapshot(snapshotPath, snapshotID)) {
	sinfo = snapshotIdToSnapshotInfoMap.get(
	chainManager.nextPathSnapshot(snapshotPath, snapshotID));

	if (chainManager.hasPreviousPathSnapshot(snapshotPath, snapshotID)) {
	sinfo.setPathPreviousSnapshotId(
	chainManager.previousPathSnapshot(snapshotPath, snapshotID));
	} else {
	sinfo.setPathPreviousSnapshotId(null);
	}
	snapshotIdToSnapshotInfoMap.put(sinfo.getSnapshotId(), sinfo);
	}

	UUID latestGlobalSnapshot = chainManager.getLatestGlobalSnapshotId();
	// append snapshot to the sinfos (the end).
	if (latestGlobalSnapshot != null) {
	sinfo = snapshotIdToSnapshotInfoMap.get(snapshotID);
	sinfo.setGlobalPreviousSnapshotId(latestGlobalSnapshot);
	UUID latestPathSnapshot = chainManager.getLatestPathSnapshotId(
	String.join("/", "vol1", "bucket1"));
	if (latestPathSnapshot != null) {
	sinfo.setPathPreviousSnapshotId(latestPathSnapshot);
	}
	snapshotIdToSnapshotInfoMap.put(snapshotID, sinfo);
	}
	chainManager.deleteSnapshot(snapshotIdToSnapshotInfoMap.get(snapshotID));
	}

	@Test
	public void testAddSnapshot() throws Exception {
	final String snapshotPath = "vol1/bucket1";
	// add three snapshots
	UUID snapshotID1 = UUID.randomUUID();
	UUID snapshotID2 = UUID.randomUUID();
	UUID snapshotID3 = UUID.randomUUID();

	ArrayList<UUID> snapshotIDs = new ArrayList<>();
	snapshotIDs.add(snapshotID1);
	snapshotIDs.add(snapshotID2);
	snapshotIDs.add(snapshotID3);

	UUID prevSnapshotID = null;
	List<Integer> times = Arrays.asList(1, 2, 0);
	int timeIdx = 0;
	// add 3 snapshots
	for (UUID snapshotID : snapshotIDs) {
	chainManager.addSnapshot(createSnapshotInfo(
	snapshotID,
	prevSnapshotID,
	prevSnapshotID, times.get(timeIdx++)));
	prevSnapshotID = snapshotID;
	}

	assertEquals(snapshotID3, chainManager.getLatestGlobalSnapshotId());
	assertEquals(snapshotID3, chainManager.getLatestPathSnapshotId(
	String.join("/", "vol1", "bucket1")));

	int i = 0;
	UUID curID = snapshotIDs.get(i);
	while (chainManager.hasNextGlobalSnapshot(curID)) {
	i++;
	assertEquals(snapshotIDs.get(i), chainManager.nextGlobalSnapshot(curID));
	curID = snapshotIDs.get(i);
	}

	curID = snapshotIDs.get(i);
	while (chainManager.hasPreviousGlobalSnapshot(curID)) {
	i--;
	assertEquals(snapshotIDs.get(i),
	chainManager.previousGlobalSnapshot(curID));
	curID = snapshotIDs.get(i);
	}

	i = 0;
	curID = snapshotIDs.get(i);
	while (chainManager.hasNextPathSnapshot(snapshotPath, curID)) {
	i++;
	assertEquals(snapshotIDs.get(i),
	chainManager.nextPathSnapshot(snapshotPath, curID));
	curID = snapshotIDs.get(i);
	}

	curID = snapshotIDs.get(i);
	while (chainManager.hasPreviousPathSnapshot(snapshotPath,
	curID)) {
	i--;
	assertEquals(snapshotIDs.get(i),
	chainManager.previousPathSnapshot(snapshotPath, curID));
	curID = snapshotIDs.get(i);
	}
	}

	@Test
	public void testDeleteSnapshot() throws Exception {
	// add three snapshots
	UUID snapshotID1 = UUID.randomUUID();
	UUID snapshotID2 = UUID.randomUUID();
	UUID snapshotID3 = UUID.randomUUID();

	ArrayList<UUID> snapshotIDs = new ArrayList<>();
	snapshotIDs.add(snapshotID1);
	snapshotIDs.add(snapshotID2);
	snapshotIDs.add(snapshotID3);

	UUID prevSnapshotID = null;
	List<Integer> times = Arrays.asList(1, 2, 0);
	int timeIdx = 0;
	// add 3 snapshots
	for (UUID snapshotID : snapshotIDs) {
	snapshotIdToSnapshotInfoMap.put(snapshotID,
	createSnapshotInfo(
	snapshotID,
	prevSnapshotID,
	prevSnapshotID, times.get(timeIdx++)));

	chainManager.addSnapshot(snapshotIdToSnapshotInfoMap.get(snapshotID));
	prevSnapshotID = snapshotID;
	}

	// delete snapshotID2
	// should have snapshots ID1 and ID3 in chain

	deleteSnapshot(snapshotID2);
	// start with first snapshot in snapshot chain
	assertFalse(chainManager.hasPreviousGlobalSnapshot(snapshotID1));
	assertTrue(chainManager.hasNextGlobalSnapshot(snapshotID1));
	assertFalse(chainManager.hasNextGlobalSnapshot(snapshotID3));
	assertEquals(snapshotID3,
	chainManager.nextGlobalSnapshot(snapshotID1));

	// add snapshotID2 and delete snapshotID1
	// should have snapshotID3 and snapshotID2
	deleteSnapshot(snapshotID1);
	chainManager.addSnapshot(snapshotIdToSnapshotInfoMap.get(snapshotID2));

	assertFalse(chainManager.hasPreviousGlobalSnapshot(snapshotID3));
	assertTrue(chainManager.hasNextGlobalSnapshot(snapshotID3));
	assertEquals(snapshotID2, chainManager.nextGlobalSnapshot(snapshotID3));
	assertFalse(chainManager.hasNextGlobalSnapshot(snapshotID2));
	assertEquals(snapshotID3, chainManager.previousGlobalSnapshot(snapshotID2));
	}

	@ParameterizedTest
	@ValueSource(booleans = {true, false})
	public void testChainFromLoadFromTable(boolean increasingTIme)
	throws Exception {
	Table<String, SnapshotInfo> snapshotInfo =
	omMetadataManager.getSnapshotInfoTable();

	// add two snapshots to the snapshotInfo
	UUID snapshotID1 = UUID.randomUUID();
	UUID snapshotID2 = UUID.randomUUID();
	List<UUID> snapshotIDs = new ArrayList<>();
	snapshotIDs.add(snapshotID1);
	snapshotIDs.add(snapshotID2);
	List<SnapshotInfo> snapshotInfoList = new ArrayList<>();

	UUID prevSnapshotID = null;
	long time = System.currentTimeMillis();
	for (UUID snapshotID : snapshotIDs) {
	SnapshotInfo snapInfo = createSnapshotInfo(snapshotID, prevSnapshotID,
	prevSnapshotID, increasingTIme ? time++ : time--);
	snapshotInfo.put(snapshotID.toString(), snapInfo);
	prevSnapshotID = snapshotID;
	snapshotInfoList.add(snapInfo);
	}

	chainManager = new SnapshotChainManager(omMetadataManager);
	assertFalse(chainManager.isSnapshotChainCorrupted());
	// check if snapshots loaded correctly from snapshotInfoTable
	assertEquals(snapshotID2, chainManager.getLatestGlobalSnapshotId());
	assertEquals(snapshotID2, chainManager.nextGlobalSnapshot(snapshotID1));
	assertEquals(snapshotID1, chainManager.previousPathSnapshot(String
	.join("/", "vol1", "bucket1"), snapshotID2));
	assertThrows(NoSuchElementException.class,
	() -> chainManager.nextGlobalSnapshot(snapshotID2));
	assertThrows(NoSuchElementException.class,
	() -> chainManager.previousPathSnapshot(String
	.join("/", "vol1", "bucket1"), snapshotID1));

	UUID snapshotID3 = UUID.randomUUID();
	SnapshotInfo snapshotInfo3 = createSnapshotInfo(snapshotID3, prevSnapshotID,
	prevSnapshotID, Time.now());
	// Add and delete snapshot to make sure snapshot chain is correct.
	chainManager.addSnapshot(snapshotInfo3);
	chainManager.deleteSnapshot(snapshotInfoList.get(0));
	assertEquals(snapshotID3, chainManager.getLatestGlobalSnapshotId());
	assertThrows(NoSuchElementException.class,
	() -> chainManager.nextGlobalSnapshot(snapshotID1));
	}

	private static Stream<? extends Arguments> invalidSnapshotChain() {
	List<UUID> nodes = IntStream.range(0, 5)
	.mapToObj(i -> UUID.randomUUID())
	.collect(Collectors.toList());
	return Stream.of(
	Arguments.of(nodes, Named.of("Disconnected Snapshot Chain",
	ImmutableMap.of(
	nodes.get(1), nodes.get(0),
	nodes.get(2), nodes.get(1),
	nodes.get(4), nodes.get(3)))),
	Arguments.of(nodes, Named.of("Complete Cyclic Snapshot Chain",
	ImmutableMap.of(
	nodes.get(0), nodes.get(4),
	nodes.get(1), nodes.get(0),
	nodes.get(2), nodes.get(1),
	nodes.get(3), nodes.get(2),
	nodes.get(4), nodes.get(3)))),
	Arguments.of(nodes, Named.of("Partial Cyclic Snapshot Chain",
	ImmutableMap.of(
	nodes.get(0), nodes.get(3),
	nodes.get(1), nodes.get(0),
	nodes.get(2), nodes.get(1),
	nodes.get(3), nodes.get(2),
	nodes.get(4), nodes.get(3)))),
	Arguments.of(nodes, Named.of("Diverged Snapshot Chain",
	ImmutableMap.of(nodes.get(1), nodes.get(0),
	nodes.get(2), nodes.get(1),
	nodes.get(3), nodes.get(2),
	nodes.get(4), nodes.get(2))))
	);
	}

	@ParameterizedTest
	@MethodSource("invalidSnapshotChain")
	public void testInvalidGlobalChainFromLoadFromTable(
	List<UUID> snapshotIDs, Map<UUID, UUID> snapshotChain) throws Exception {
	Table<String, SnapshotInfo> snapshotInfo =
	omMetadataManager.getSnapshotInfoTable();
	for (UUID snapshotID : snapshotIDs) {
	snapshotInfo.put(snapshotID.toString(),
	createSnapshotInfo(snapshotID, snapshotChain.get(snapshotID),
	snapshotChain.get(snapshotID), System.currentTimeMillis()));
	}
	chainManager = new SnapshotChainManager(omMetadataManager);

	assertTrue(chainManager.isSnapshotChainCorrupted());

	SnapshotInfo snapInfo = createSnapshotInfo(UUID.randomUUID(),
	UUID.randomUUID(),
	UUID.randomUUID(),
	System.currentTimeMillis());
	IOException createException = assertThrows(IOException.class,
	() -> chainManager.addSnapshot(snapInfo));
	assertEquals("Snapshot chain is corrupted.", createException.getMessage());
	if (!snapshotIDs.isEmpty()) {
	IOException deleteException =
	assertThrows(IOException.class,
	() -> chainManager.deleteSnapshot(
	snapshotInfo.get(snapshotIDs.get(0).toString())));
	assertEquals("Snapshot chain is corrupted.",
	deleteException.getMessage());
	}
	}

	@ParameterizedTest
	@MethodSource("invalidSnapshotChain")
	public void testInvalidChainFromLoadFromTable(List<UUID> snapshotIDs,
	Map<UUID, UUID> snapshotChain) throws Exception {
	Table<String, SnapshotInfo> snapshotInfo =
	omMetadataManager.getSnapshotInfoTable();
	UUID prevSnapshotId = null;
	for (UUID snapshotID : snapshotIDs) {
	snapshotInfo.put(snapshotID.toString(),
	createSnapshotInfo(snapshotID, snapshotChain.get(snapshotID),
	prevSnapshotId, System.currentTimeMillis()));
	prevSnapshotId = snapshotID;
	}
	chainManager = new SnapshotChainManager(omMetadataManager);
	assertTrue(chainManager.isSnapshotChainCorrupted());

	SnapshotInfo snapInfo = createSnapshotInfo(UUID.randomUUID(),
	UUID.randomUUID(),
	UUID.randomUUID(),
	System.currentTimeMillis());
	IOException createException = assertThrows(IOException.class,
	() -> chainManager.addSnapshot(snapInfo));
	assertEquals("Snapshot chain is corrupted.", createException.getMessage());
	if (!snapshotIDs.isEmpty()) {
	IOException deleteException =
	assertThrows(IOException.class,
	() -> chainManager.deleteSnapshot(
	snapshotInfo.get(snapshotIDs.get(0).toString())));
	assertEquals("Snapshot chain is corrupted.",
	deleteException.getMessage());
	}
	}
	}