Adding undelete record to persistent index (#1366)

Added behavioral changes for PersistentIndex for Undeletes

Added test for bad undelete, test infra for undelete in CuratedLogIndexState
Added undelete/lifeVersion awareness to findEntriesSince method
Co-authored-by: David Harju <david.a.harju@gmail.com>

Author: justinlin-linkedin

ambry-api/src/main/java/com.github.ambry/store/StoreErrorCodes.java

@@ -35,5 +35,9 @@ public enum StoreErrorCodes {
   Already_Updated,
   Update_Not_Allowed,
   File_Not_Found,
-  Channel_Closed
+  Channel_Closed,
+  Life_Version_Conflict,
+  ID_Not_Deleted,
+  ID_Undeleted,
+  ID_Deleted_Permanently
 }

ambry-store/src/main/java/com.github.ambry.store/IndexSegment.java

@@ -1010,10 +1010,9 @@ boolean getEntriesSince(StoreKey key, FindEntriesCondition findEntriesCondition,
         getIndexEntriesSince(key, findEntriesCondition, indexEntries, currentTotalSizeOfEntriesInBytes, true);
     for (IndexEntry indexEntry : indexEntries) {
       IndexValue value = indexEntry.getValue();
-      MessageInfo info =
-          new MessageInfo(indexEntry.getKey(), value.getSize(), value.isFlagSet(IndexValue.Flags.Delete_Index),
-              value.isFlagSet(IndexValue.Flags.Ttl_Update_Index), value.getExpiresAtMs(), value.getAccountId(),
-              value.getContainerId(), value.getOperationTimeInMs());
+      MessageInfo info = new MessageInfo(indexEntry.getKey(), value.getSize(), value.isDelete(), value.isTTLUpdate(),
+          value.isUndelete(), value.getExpiresAtMs(), null, value.getAccountId(), value.getContainerId(),
+          value.getOperationTimeInMs(), value.getLifeVersion());
       entries.add(info);
     }
     return areNewEntriesAdded;
@@ -1100,10 +1099,10 @@ boolean getIndexEntriesSince(StoreKey key, FindEntriesCondition findEntriesCondi
    */
   private void eliminateDuplicates(List<IndexEntry> entries) {
     Set<StoreKey> setToFindDuplicate = new HashSet<>();
-    // first choose PUTs over update entries (omitting DELETEs)
-    entries.removeIf(
-        entry -> !entry.getValue().isFlagSet(IndexValue.Flags.Delete_Index) && !setToFindDuplicate.add(entry.getKey()));
-    // then choose DELETEs over all other entries
+    // first choose PUTs over update entries (omitting DELETEs and UNDELETEs)
+    entries.removeIf(entry -> !entry.getValue().isDelete() && !entry.getValue().isUndelete() && !setToFindDuplicate.add(
+        entry.getKey()));
+    // then choose DELETEs/UNDELETEs over all other entries
     setToFindDuplicate.clear();
     ListIterator<IndexEntry> iterator = entries.listIterator(entries.size());
     while (iterator.hasPrevious()) {

ambry-store/src/main/java/com.github.ambry.store/IndexValue.java

@@ -55,6 +55,8 @@ enum Flags {
 
   final static byte FLAGS_DEFAULT_VALUE = (byte) 0;
   final static long UNKNOWN_ORIGINAL_MESSAGE_OFFSET = -1;
+  // The life version when the operation is trigger by the requests from frontend.
+  final static short LIFE_VERSION_FROM_FRONTEND = -1;
 
   private final static int BLOB_SIZE_IN_BYTES = 8;
   private final static int OFFSET_SIZE_IN_BYTES = 8;
@@ -186,6 +188,15 @@ enum Flags {
     this(size, offset, flags, expiresAtMs, offset.getOffset(), operationTimeInMs, accountId, containerId, lifeVersion);
   }
 
+  /**
+   * Constructor to copy all data from a given {@link IndexValue}.
+   * @param other the given {@link IndexValue}.
+   */
+  IndexValue(IndexValue other) {
+    this(other.getSize(), other.getOffset(), other.getFlags(), other.getExpiresAtMs(), other.getOriginalMessageOffset(),
+        other.getOperationTimeInMs(), other.getAccountId(), other.getContainerId(), other.getLifeVersion());
+  }
+
   /**
    * Constructs IndexValue based on the args passed
    * @param size the size of the blob that this index value refers to
@@ -243,6 +254,38 @@ boolean isFlagSet(Flags flag) {
     return ((getFlags() & (1 << flag.ordinal())) != 0);
   }
 
+  /**
+   * Helper function for isFlagSet(Flags.Ttl_Update_Index).
+   * @return true when the Ttl_Update_Index is set.
+   */
+  boolean isTTLUpdate() {
+    return isFlagSet(Flags.Ttl_Update_Index);
+  }
+
+  /**
+   * Helper function for isFlagSet(Flags.Delete_Index).
+   * @return true when the Delete_Index is set.
+   */
+  boolean isDelete() {
+    return isFlagSet(Flags.Delete_Index);
+  }
+
+  /**
+   * Helper function for isFlagSet(Flags.Undelete_Index).
+   * @return true when the Undelete_Index is set.
+   */
+  boolean isUndelete() {
+    return isFlagSet(Flags.Undelete_Index);
+  }
+
+  /**
+   * Helper function to decide if this value is a put value or not.
+   * @return true when it's not a put record.
+   */
+  boolean isPut() {
+    return flags == FLAGS_DEFAULT_VALUE;
+  }
+
   /**
    * @return the expiration time of the index value in ms
    */
@@ -279,6 +322,15 @@ short getContainerId() {
     return containerId;
   }
 
+  /**
+   * True when the life version is not from frontend requests.
+   * @param lifeVersion the given life version.
+   * @return true when it's not from frontend requests.
+   */
+  static boolean hasLifeVersion(short lifeVersion) {
+    return lifeVersion > LIFE_VERSION_FROM_FRONTEND;
+  }
+
   /**
    * @return the lifeVersion of the {@link IndexValue}
    */
@@ -385,12 +437,11 @@ ByteBuffer getBytes() {
 
   @Override
   public String toString() {
-    return "Offset: " + offset + ", Size: " + getSize() + ", Deleted: " + isFlagSet(Flags.Delete_Index)
-        + ", TTL Updated: " + isFlagSet(Flags.Ttl_Update_Index) + ", Undelete: " + isFlagSet(
-        Flags.Undelete_Index) + ", ExpiresAtMs: " + getExpiresAtMs() + ", Original Message Offset: "
+    return "Offset: " + offset + ", Size: " + getSize() + ", Deleted: " + isDelete() + ", TTL Updated: " + isTTLUpdate()
+        + ", Undelete: " + isUndelete() + ", ExpiresAtMs: " + getExpiresAtMs() + ", Original Message Offset: "
         + getOriginalMessageOffset() + (formatVersion != PersistentIndex.VERSION_0 ? (", OperationTimeAtSecs "
-        + getOperationTimeInMs() + ", AccountId " + getAccountId() + ", ContainerId " + getContainerId())
-        : "") + (formatVersion > PersistentIndex.VERSION_2 ? ", Life Version:" + lifeVersion : "");
+        + getOperationTimeInMs() + ", AccountId " + getAccountId() + ", ContainerId " + getContainerId()) : "") + (
+        formatVersion > PersistentIndex.VERSION_2 ? ", Life Version:" + lifeVersion : "");
   }
 
   /**