HDDS-13833. Add transactionInfo field in SnapshotLocalData and update the value on SnapshotPurgeRequest

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

@@ -221,6 +221,7 @@ public final class OzoneConsts {
   public static final String OM_SST_FILE_INFO_START_KEY = "startKey";
   public static final String OM_SST_FILE_INFO_END_KEY = "endKey";
   public static final String OM_SST_FILE_INFO_COL_FAMILY = "columnFamily";
+  public static final String OM_SLD_TXN_INFO = "transactionInfo";
 
   // YAML fields for .container files
   public static final String CONTAINER_ID = "containerID";

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

@@ -30,6 +30,7 @@
 import java.util.UUID;
 import java.util.stream.Collectors;
 import org.apache.commons.codec.digest.DigestUtils;
+import org.apache.hadoop.hdds.utils.TransactionInfo;
 import org.apache.hadoop.hdds.utils.db.CopyObject;
 import org.apache.hadoop.ozone.util.WithChecksum;
 import org.apache.ozone.compaction.log.SstFileInfo;
@@ -63,6 +64,9 @@ public class OmSnapshotLocalData implements WithChecksum<OmSnapshotLocalData> {
   // Previous snapshotId based on which the snapshot local data is built.
   private UUID previousSnapshotId;
 
+  // Stores the transactionInfo corresponding to OM when the snaphot is purged.
+  private TransactionInfo transactionInfo;
+
   // Map of version to VersionMeta, using linkedHashMap since the order of the map needs to be deterministic for
   // checksum computation.
   private final LinkedHashMap<Integer, VersionMeta> versionSstFileInfos;
@@ -73,7 +77,8 @@ public class OmSnapshotLocalData implements WithChecksum<OmSnapshotLocalData> {
   /**
    * Creates a OmSnapshotLocalData object with default values.
    */
-  public OmSnapshotLocalData(UUID snapshotId, List<LiveFileMetaData> notDefraggedSSTFileList, UUID previousSnapshotId) {
+  public OmSnapshotLocalData(UUID snapshotId, List<LiveFileMetaData> notDefraggedSSTFileList, UUID previousSnapshotId,
+      TransactionInfo transactionInfo) {
     this.snapshotId = snapshotId;
     this.isSSTFiltered = false;
     this.lastDefragTime = 0L;
@@ -83,6 +88,7 @@ public OmSnapshotLocalData(UUID snapshotId, List<LiveFileMetaData> notDefraggedS
         new VersionMeta(0, notDefraggedSSTFileList.stream().map(SstFileInfo::new).collect(Collectors.toList())));
     this.version = 0;
     this.previousSnapshotId = previousSnapshotId;
+    this.transactionInfo = transactionInfo;
     setChecksumTo0ByteArray();
   }
 
@@ -101,6 +107,15 @@ public OmSnapshotLocalData(OmSnapshotLocalData source) {
     this.previousSnapshotId = source.previousSnapshotId;
     this.versionSstFileInfos = new LinkedHashMap<>();
     setVersionSstFileInfos(source.versionSstFileInfos);
+    this.transactionInfo = source.transactionInfo;
+  }
+
+  public TransactionInfo getTransactionInfo() {
+    return transactionInfo;
+  }
+
+  public void setTransactionInfo(TransactionInfo transactionInfo) {
+    this.transactionInfo = transactionInfo;
   }
 
   /**

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

@@ -24,6 +24,7 @@
 import org.apache.commons.pool2.BasePooledObjectFactory;
 import org.apache.commons.pool2.PooledObject;
 import org.apache.commons.pool2.impl.DefaultPooledObject;
+import org.apache.hadoop.hdds.utils.TransactionInfo;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.om.OmSnapshotLocalData.VersionMeta;
 import org.apache.ozone.compaction.log.SstFileInfo;
@@ -71,6 +72,8 @@ private static class OmSnapshotLocalDataRepresenter extends Representer {
       this.addClassTag(SstFileInfo.class, SST_FILE_INFO_TAG);
       representers.put(SstFileInfo.class, new RepresentSstFileInfo());
       representers.put(VersionMeta.class, new RepresentVersionMeta());
+      representers.put(TransactionInfo.class, data -> new ScalarNode(Tag.STR, data.toString(), null, null,
+          DumperOptions.ScalarStyle.PLAIN));
       representers.put(UUID.class, data ->
           new ScalarNode(Tag.STR, data.toString(), null, null, DumperOptions.ScalarStyle.PLAIN));
     }
@@ -168,7 +171,10 @@ public Object construct(Node node) {
         UUID snapId = UUID.fromString(snapIdStr);
         final String prevSnapIdStr = (String) nodes.get(OzoneConsts.OM_SLD_PREV_SNAP_ID);
         UUID prevSnapId = prevSnapIdStr != null ? UUID.fromString(prevSnapIdStr) : null;
-        OmSnapshotLocalData snapshotLocalData = new OmSnapshotLocalData(snapId, Collections.emptyList(), prevSnapId);
+        final String purgeTxInfoStr = (String) nodes.get(OzoneConsts.OM_SLD_TXN_INFO);
+        TransactionInfo transactionInfo = purgeTxInfoStr != null ? TransactionInfo.valueOf(purgeTxInfoStr) : null;
+        OmSnapshotLocalData snapshotLocalData = new OmSnapshotLocalData(snapId, Collections.emptyList(), prevSnapId,
+            transactionInfo);
 
         // Set version from YAML
         Integer version = (Integer) nodes.get(OzoneConsts.OM_SLD_VERSION);

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/snapshot/OMSnapshotPurgeRequest.java

@@ -91,6 +91,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
 
     List<String> snapshotDbKeys = snapshotPurgeRequest
         .getSnapshotDBKeysList();
+    TransactionInfo transactionInfo = TransactionInfo.valueOf(context.getTermIndex());
     try {
 
       // Each snapshot purge operation does three things:
@@ -123,12 +124,13 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
       }
       // Update the snapshotInfo lastTransactionInfo.
       for (SnapshotInfo snapshotInfo : updatedSnapshotInfos.values()) {
-        snapshotInfo.setLastTransactionInfo(TransactionInfo.valueOf(context.getTermIndex()).toByteString());
+        snapshotInfo.setLastTransactionInfo(transactionInfo.toByteString());
         omMetadataManager.getSnapshotInfoTable().addCacheEntry(new CacheKey<>(snapshotInfo.getTableKey()),
             CacheValue.get(context.getIndex(), snapshotInfo));
       }
 
-      omClientResponse = new OMSnapshotPurgeResponse(omResponse.build(), snapshotDbKeys, updatedSnapshotInfos);
+      omClientResponse = new OMSnapshotPurgeResponse(omResponse.build(), snapshotDbKeys, updatedSnapshotInfos,
+          transactionInfo);
 
       omSnapshotIntMetrics.incNumSnapshotPurges();
       LOG.info("Successfully executed snapshotPurgeRequest: {{}} along with updating snapshots:{}.",

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/snapshot/OMSnapshotPurgeResponse.java

@@ -27,6 +27,7 @@
 import java.util.List;
 import java.util.Map;
 import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.hdds.utils.TransactionInfo;
 import org.apache.hadoop.hdds.utils.db.BatchOperation;
 import org.apache.hadoop.ozone.om.OMMetadataManager;
 import org.apache.hadoop.ozone.om.OmMetadataManagerImpl;
@@ -36,6 +37,7 @@
 import org.apache.hadoop.ozone.om.response.CleanupTableInfo;
 import org.apache.hadoop.ozone.om.response.OMClientResponse;
 import org.apache.hadoop.ozone.om.snapshot.OmSnapshotLocalDataManager;
+import org.apache.hadoop.ozone.om.snapshot.OmSnapshotLocalDataManager.WritableOmSnapshotLocalDataProvider;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -49,15 +51,18 @@ public class OMSnapshotPurgeResponse extends OMClientResponse {
       LoggerFactory.getLogger(OMSnapshotPurgeResponse.class);
   private final List<String> snapshotDbKeys;
   private final Map<String, SnapshotInfo> updatedSnapInfos;
+  private final TransactionInfo transactionInfo;
 
   public OMSnapshotPurgeResponse(
       @Nonnull OMResponse omResponse,
       @Nonnull List<String> snapshotDbKeys,
-      Map<String, SnapshotInfo> updatedSnapInfos
+      Map<String, SnapshotInfo> updatedSnapInfos,
+      TransactionInfo transactionInfo
   ) {
     super(omResponse);
     this.snapshotDbKeys = snapshotDbKeys;
     this.updatedSnapInfos = updatedSnapInfos;
+    this.transactionInfo = transactionInfo;
   }
 
   /**
@@ -69,6 +74,7 @@ public OMSnapshotPurgeResponse(@Nonnull OMResponse omResponse) {
     checkStatusNotOK();
     this.snapshotDbKeys = null;
     this.updatedSnapInfos = null;
+    this.transactionInfo = null;
   }
 
   @Override
@@ -96,10 +102,14 @@ protected void addToDBBatch(OMMetadataManager omMetadataManager,
       // Remove the snapshot from snapshotId to snapshotTableKey map.
       ((OmMetadataManagerImpl) omMetadataManager).getSnapshotChainManager()
           .removeFromSnapshotIdToTable(snapshotInfo.getSnapshotId());
-      // Delete Snapshot checkpoint directory.
+
       OmSnapshotLocalDataManager snapshotLocalDataManager = ((OmMetadataManagerImpl) omMetadataManager)
           .getOzoneManager().getOmSnapshotManager().getSnapshotLocalDataManager();
-      deleteCheckpointDirectory(snapshotLocalDataManager, omMetadataManager, snapshotInfo);
+      // Update snapshot local data to update purge transaction info. This would be used to check whether the
+      // snapshot purged txn is flushed to rocksdb.
+      updateLocalData(snapshotLocalDataManager, snapshotInfo);
+      // Delete Snapshot checkpoint directory.
+      deleteCheckpointDirectory(omMetadataManager, snapshotInfo);
       // Delete snapshotInfo from the table.
       omMetadataManager.getSnapshotInfoTable().deleteWithBatch(batchOperation, dbKey);
     }
@@ -115,11 +125,18 @@ private void updateSnapInfo(OmMetadataManagerImpl metadataManager,
     }
   }
 
+  private void updateLocalData(OmSnapshotLocalDataManager localDataManager, SnapshotInfo snapshotInfo)
+      throws IOException {
+    try (WritableOmSnapshotLocalDataProvider snap = localDataManager.getWritableOmSnapshotLocalData(snapshotInfo)) {
+      snap.setTransactionInfo(this.transactionInfo);
+      snap.commit();
+    }
+  }
+
   /**
    * Deletes the checkpoint directory for a snapshot.
    */
-  private void deleteCheckpointDirectory(OmSnapshotLocalDataManager snapshotLocalDataManager,
-      OMMetadataManager omMetadataManager, SnapshotInfo snapshotInfo) {
+  private void deleteCheckpointDirectory(OMMetadataManager omMetadataManager, SnapshotInfo snapshotInfo) {
     // Acquiring write lock to avoid race condition with sst filtering service which creates a sst filtered file
     // inside the snapshot directory. Any operation apart which doesn't create/delete files under this snapshot
     // directory can run in parallel along with this operation.

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/OmSnapshotLocalDataManager.java

@@ -45,6 +45,7 @@
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.stream.Collectors;
 import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.hdds.utils.TransactionInfo;
 import org.apache.hadoop.hdds.utils.db.RDBStore;
 import org.apache.hadoop.ozone.om.OMMetadataManager;
 import org.apache.hadoop.ozone.om.OmSnapshotLocalData;
@@ -155,10 +156,11 @@ public String getSnapshotLocalPropertyYamlPath(UUID snapshotId) {
    */
   public void createNewOmSnapshotLocalDataFile(RDBStore snapshotStore, SnapshotInfo snapshotInfo) throws IOException {
     try (WritableOmSnapshotLocalDataProvider snapshotLocalData =
-        new WritableOmSnapshotLocalDataProvider(snapshotInfo.getSnapshotId(),
-            () -> Pair.of(new OmSnapshotLocalData(snapshotInfo.getSnapshotId(),
-                OmSnapshotManager.getSnapshotSSTFileList(snapshotStore), snapshotInfo.getPathPreviousSnapshotId()),
-                null))) {
+             new WritableOmSnapshotLocalDataProvider(snapshotInfo.getSnapshotId(),
+                 () -> Pair.of(new OmSnapshotLocalData(snapshotInfo.getSnapshotId(),
+                         OmSnapshotManager.getSnapshotSSTFileList(snapshotStore),
+                         snapshotInfo.getPathPreviousSnapshotId(), null),
+                     null))) {
       snapshotLocalData.commit();
     }
   }
@@ -660,6 +662,12 @@ public void removeVersion(int version) {
       setDirty();
     }
 
+    public void setTransactionInfo(TransactionInfo transactionInfo) {
+      this.getSnapshotLocalData().setTransactionInfo(transactionInfo);
+      // Set Dirty when the transactionInfo is set.
+      setDirty();
+    }
+
     public synchronized void commit() throws IOException {
       // Validate modification and commit the changes.
       SnapshotVersionsMeta localDataVersionNodes = validateModification(super.snapshotLocalData);

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestOmSnapshotLocalDataYaml.java

@@ -37,11 +37,13 @@
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.fs.FileSystemTestHelper;
 import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.hdds.StringUtils;
+import org.apache.hadoop.hdds.utils.TransactionInfo;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.om.OmSnapshotLocalData.VersionMeta;
 import org.apache.hadoop.ozone.util.ObjectSerializer;
@@ -106,7 +108,8 @@ private LiveFileMetaData createLiveFileMetaData(String fileName, String table, S
   /**
    * Creates a snapshot local data YAML file.
    */
-  private Pair<File, UUID> writeToYaml(UUID snapshotId, String snapshotName) throws IOException {
+  private Pair<File, UUID> writeToYaml(UUID snapshotId, String snapshotName, TransactionInfo transactionInfo)
+      throws IOException {
     String yamlFilePath = snapshotName + ".yaml";
     UUID previousSnapshotId = UUID.randomUUID();
     // Create snapshot data with not defragged SST files
@@ -115,7 +118,7 @@ private Pair<File, UUID> writeToYaml(UUID snapshotId, String snapshotName) throw
         createLiveFileMetaData("sst2", "table1", "k3", "k4"),
         createLiveFileMetaData("sst3", "table2", "k4", "k5"));
     OmSnapshotLocalData dataYaml = new OmSnapshotLocalData(snapshotId, notDefraggedSSTFileList,
-        previousSnapshotId);
+        previousSnapshotId, transactionInfo);
 
     // Set version
     dataYaml.setVersion(42);
@@ -150,7 +153,9 @@ private Pair<File, UUID> writeToYaml(UUID snapshotId, String snapshotName) throw
   @Test
   public void testWriteToYaml() throws IOException {
     UUID snapshotId = UUID.randomUUID();
-    Pair<File, UUID> yamlFilePrevIdPair = writeToYaml(snapshotId, "snapshot1");
+    TransactionInfo transactionInfo = TransactionInfo.valueOf(ThreadLocalRandom.current().nextLong(),
+        ThreadLocalRandom.current().nextLong());
+    Pair<File, UUID> yamlFilePrevIdPair = writeToYaml(snapshotId, "snapshot1", transactionInfo);
     File yamlFile = yamlFilePrevIdPair.getLeft();
     UUID prevSnapId = yamlFilePrevIdPair.getRight();
 
@@ -160,6 +165,7 @@ public void testWriteToYaml() throws IOException {
     // Verify fields
     assertEquals(44, snapshotData.getVersion());
     assertTrue(snapshotData.getSstFiltered());
+    assertEquals(transactionInfo, snapshotData.getTransactionInfo());
 
     VersionMeta notDefraggedSSTFiles = snapshotData.getVersionSstFileInfos().get(0);
     assertEquals(new VersionMeta(0,
@@ -192,17 +198,19 @@ public void testWriteToYaml() throws IOException {
   @Test
   public void testUpdateSnapshotDataFile() throws IOException {
     UUID snapshotId = UUID.randomUUID();
-    Pair<File, UUID> yamlFilePrevIdPair = writeToYaml(snapshotId, "snapshot2");
+    Pair<File, UUID> yamlFilePrevIdPair = writeToYaml(snapshotId, "snapshot2", null);
     File yamlFile = yamlFilePrevIdPair.getLeft();
     // Read from YAML file
     OmSnapshotLocalData dataYaml =
         omSnapshotLocalDataSerializer.load(yamlFile);
-
+    TransactionInfo transactionInfo = TransactionInfo.valueOf(ThreadLocalRandom.current().nextLong(),
+        ThreadLocalRandom.current().nextLong());
     // Update snapshot data
     dataYaml.setSstFiltered(false);
     dataYaml.setNeedsDefrag(false);
     dataYaml.addVersionSSTFileInfos(
         singletonList(createLiveFileMetaData("defragged-sst4", "table3", "k5", "k6")), 5);
+    dataYaml.setTransactionInfo(transactionInfo);
 
     // Write updated data back to file
     omSnapshotLocalDataSerializer.save(yamlFile, dataYaml);
@@ -213,6 +221,7 @@ public void testUpdateSnapshotDataFile() throws IOException {
     // Verify updated data
     assertThat(dataYaml.getSstFiltered()).isFalse();
     assertThat(dataYaml.getNeedsDefrag()).isFalse();
+    assertEquals(transactionInfo, dataYaml.getTransactionInfo());
 
     Map<Integer, VersionMeta> defraggedFiles = dataYaml.getVersionSstFileInfos();
     assertEquals(4, defraggedFiles.size());
@@ -234,7 +243,9 @@ public void testEmptyFile() throws IOException {
   @Test
   public void testChecksum() throws IOException {
     UUID snapshotId = UUID.randomUUID();
-    Pair<File, UUID> yamlFilePrevIdPair = writeToYaml(snapshotId, "snapshot3");
+    TransactionInfo transactionInfo = TransactionInfo.valueOf(ThreadLocalRandom.current().nextLong(),
+        ThreadLocalRandom.current().nextLong());
+    Pair<File, UUID> yamlFilePrevIdPair = writeToYaml(snapshotId, "snapshot3", transactionInfo);
     File yamlFile = yamlFilePrevIdPair.getLeft();
     // Read from YAML file
     OmSnapshotLocalData snapshotData = omSnapshotLocalDataSerializer.load(yamlFile);
@@ -251,7 +262,9 @@ public void testChecksum() throws IOException {
   @Test
   public void testYamlContainsAllFields() throws IOException {
     UUID snapshotId = UUID.randomUUID();
-    Pair<File, UUID> yamlFilePrevIdPair = writeToYaml(snapshotId, "snapshot4");
+    TransactionInfo transactionInfo = TransactionInfo.valueOf(ThreadLocalRandom.current().nextLong(),
+        ThreadLocalRandom.current().nextLong());
+    Pair<File, UUID> yamlFilePrevIdPair = writeToYaml(snapshotId, "snapshot4", transactionInfo);
     File yamlFile = yamlFilePrevIdPair.getLeft();
     String content = FileUtils.readFileToString(yamlFile, Charset.defaultCharset());
 
@@ -264,5 +277,6 @@ public void testYamlContainsAllFields() throws IOException {
     assertThat(content).contains(OzoneConsts.OM_SLD_VERSION_SST_FILE_INFO);
     assertThat(content).contains(OzoneConsts.OM_SLD_SNAP_ID);
     assertThat(content).contains(OzoneConsts.OM_SLD_PREV_SNAP_ID);
+    assertThat(content).contains(OzoneConsts.OM_SLD_TXN_INFO);
   }
 }

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/snapshot/TestOMSnapshotPurgeRequestAndResponse.java

@@ -52,6 +52,7 @@
 import org.apache.hadoop.ozone.om.request.OMRequestTestUtils;
 import org.apache.hadoop.ozone.om.response.snapshot.OMSnapshotPurgeResponse;
 import org.apache.hadoop.ozone.om.snapshot.OmSnapshotLocalDataManager;
+import org.apache.hadoop.ozone.om.snapshot.OmSnapshotLocalDataManager.ReadableOmSnapshotLocalDataProvider;
 import org.apache.hadoop.ozone.om.snapshot.TestSnapshotRequestAndResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SnapshotPurgeRequest;
@@ -159,6 +160,10 @@ public void testValidateAndUpdateCache() throws Exception {
 
     List<String> snapshotDbKeysToPurge = createSnapshots(10);
     assertFalse(getOmMetadataManager().getSnapshotInfoTable().isEmpty());
+    List<SnapshotInfo> snapshotInfos = new ArrayList<>();
+    for (String snapshotKey : snapshotDbKeysToPurge) {
+      snapshotInfos.add(getOmMetadataManager().getSnapshotInfoTable().get(snapshotKey));
+    }
 
     // Check if all the checkpoints are created.
     for (Path checkpoint : checkpointPaths) {
@@ -171,9 +176,9 @@ public void testValidateAndUpdateCache() throws Exception {
         snapshotDbKeysToPurge);
 
     OMSnapshotPurgeRequest omSnapshotPurgeRequest = preExecute(snapshotPurgeRequest);
-
+    TransactionInfo transactionInfo = TransactionInfo.valueOf(TransactionInfo.getTermIndex(200L));
     OMSnapshotPurgeResponse omSnapshotPurgeResponse = (OMSnapshotPurgeResponse)
-        omSnapshotPurgeRequest.validateAndUpdateCache(getOzoneManager(), 200L);
+        omSnapshotPurgeRequest.validateAndUpdateCache(getOzoneManager(), transactionInfo.getTransactionIndex());
 
     for (String snapshotTableKey: snapshotDbKeysToPurge) {
       assertNull(getOmMetadataManager().getSnapshotInfoTable().get(snapshotTableKey));
@@ -191,6 +196,15 @@ public void testValidateAndUpdateCache() throws Exception {
     for (Path checkpoint : checkpointPaths) {
       assertFalse(Files.exists(checkpoint));
     }
+    OmSnapshotLocalDataManager snapshotLocalDataManager =
+        getOzoneManager().getOmSnapshotManager().getSnapshotLocalDataManager();
+    for (SnapshotInfo snapshotInfo : snapshotInfos) {
+      try (ReadableOmSnapshotLocalDataProvider snapProvider =
+               snapshotLocalDataManager.getOmSnapshotLocalData(snapshotInfo)) {
+        assertEquals(transactionInfo, snapProvider.getSnapshotLocalData().getTransactionInfo());
+      }
+    }
+
     assertEquals(initialSnapshotPurgeCount + 1, getOmSnapshotIntMetrics().getNumSnapshotPurges());
     assertEquals(initialSnapshotPurgeFailCount, getOmSnapshotIntMetrics().getNumSnapshotPurgeFails());
   }