Revert "Server returns Blob_Deleted for deleted blob out of retention (…

…linkedin#2821)" (linkedin#2835)

This reverts commit 1d989b0.

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

@@ -25,12 +25,8 @@ public enum StoreGetOptions {
   Store_Include_Expired,
   /**
    * This option indicates that the store needs to return the message even if it has been
-   * marked for deletion as long as the message is still in delete retention window
+   * marked for deletion as long as the message has not been physically deleted from the
+   * store.
    */
-  Store_Include_Deleted,
-  /**
-   * This option indicates that the store needs to return the message as long as the message
-   * has not been physically deleted from the store.
-   */
-  Store_Include_Compaction_Ready
+  Store_Include_Deleted
 }

ambry-server/src/main/java/com/github/ambry/server/AmbryRequests.java

@@ -1527,7 +1527,7 @@ protected EnumSet<StoreGetOptions> getStoreGetOptions(GetRequest getRequest) {
       storeGetOptions = EnumSet.of(StoreGetOptions.Store_Include_Deleted);
     }
     if (getRequest.getGetOption() == GetOption.Include_All) {
-      storeGetOptions = EnumSet.allOf(StoreGetOptions.class);
+      storeGetOptions = EnumSet.of(StoreGetOptions.Store_Include_Deleted, StoreGetOptions.Store_Include_Expired);
     }
     return storeGetOptions;
   }

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

@@ -343,7 +343,7 @@ public Long getBlobContentCRC(MessageInfo msg) throws StoreException, IOExceptio
         StoreGetOptions.Store_Include_Expired);
     MessageReadSet rdset = null;
     try {
-      StoreInfo stinfo = get(Collections.singletonList(msg.getStoreKey()), storeGetOptions);
+      StoreInfo stinfo = this.get(Collections.singletonList(msg.getStoreKey()), storeGetOptions);
       rdset = stinfo.getMessageReadSet();
       MessageInfo minfo = stinfo.getMessageReadSetInfo().get(0);
       rdset.doPrefetch(0, minfo.getSize() - MessageFormatRecord.Crc_Size,

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

@@ -632,8 +632,7 @@ private void persistCleanupToken() throws IOException, StoreException {
    */
   private void performHardDeletes(List<MessageInfo> messageInfoList) throws StoreException {
     try {
-      EnumSet<StoreGetOptions> getOptions =
-          EnumSet.of(StoreGetOptions.Store_Include_Deleted, StoreGetOptions.Store_Include_Compaction_Ready);
+      EnumSet<StoreGetOptions> getOptions = EnumSet.of(StoreGetOptions.Store_Include_Deleted);
       List<BlobReadOptions> readOptionsList = new ArrayList<BlobReadOptions>(messageInfoList.size());
 
       /* First create the readOptionsList */

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

@@ -1370,18 +1370,10 @@ BlobReadOptions getBlobReadInfo(StoreKey id, EnumSet<StoreGetOptions> getOptions
       if (value == null) {
         throw new StoreException("Id " + id + " not present in index " + dataDir, StoreErrorCodes.ID_Not_Found);
       } else if (value.isDelete()) {
-        // Blob is deleted, as long as
-        // 1. get option includes Compaction_Ready blob
-        // 2. or get option includes Deleted blob and the blob is deleted no more than retention window time.
-        // we return the blob back to frontend
-        long deleteOperationTime = value.getOperationTimeInMs();
-        if (getOptions.contains(StoreGetOptions.Store_Include_Compaction_Ready) || (
-            getOptions.contains(StoreGetOptions.Store_Include_Deleted)
-                && deleteOperationTime + TimeUnit.MINUTES.toMillis(config.storeDeletedMessageRetentionMinutes)
-                >= time.milliseconds())) {
-          readOptions = getDeletedBlobReadOptions(value, id, indexSegments);
-        } else {
+        if (!getOptions.contains(StoreGetOptions.Store_Include_Deleted)) {
           throw new StoreException("Id " + id + " has been deleted in index " + dataDir, StoreErrorCodes.ID_Deleted);
+        } else {
+          readOptions = getDeletedBlobReadOptions(value, id, indexSegments);
         }
       } else if (isExpired(value) && !getOptions.contains(StoreGetOptions.Store_Include_Expired)) {
         throw new StoreException("Id " + id + " has expired ttl in index " + dataDir, StoreErrorCodes.TTL_Expired);

ambry-store/src/test/java/com/github/ambry/store/BlobStoreTest.java

@@ -276,7 +276,7 @@ public CallableResult call() throws Exception {
   // A list of keys grouped by the log segment that they belong to
   private final List<Set<MockId>> idsByLogSegment = new ArrayList<>();
   // Set of all deleted keys
-  private final Map<MockId, Long> deletedKeys = new ConcurrentHashMap<MockId, Long>();
+  private final Set<MockId> deletedKeys = Collections.newSetFromMap(new ConcurrentHashMap<MockId, Boolean>());
   // Set of all expired keys
   private final Set<MockId> expiredKeys = Collections.newSetFromMap(new ConcurrentHashMap<MockId, Boolean>());
   // Set of all keys that are not deleted/expired
@@ -657,28 +657,14 @@ public void basicTest() throws InterruptedException, IOException, StoreException
     checkStoreInfo(storeInfo, liveKeys);
 
     MockMessageStoreHardDelete hd = (MockMessageStoreHardDelete) hardDelete;
-    for (MockId id : deletedKeys.keySet()) {
+    for (MockId id : deletedKeys) {
       // cannot get without StoreGetOptions
       verifyGetFailure(id, StoreErrorCodes.ID_Deleted);
 
       // with StoreGetOptions.Store_Include_Deleted
-      storeInfo = store.get(Collections.singletonList(id), EnumSet.of(StoreGetOptions.Store_Include_Compaction_Ready));
+      storeInfo = store.get(Collections.singletonList(id), EnumSet.of(StoreGetOptions.Store_Include_Deleted));
       checkStoreInfo(storeInfo, Collections.singleton(id));
 
-      long operationTimeMs = deletedKeys.get(id);
-      if (operationTimeMs + TimeUnit.MINUTES.toMillis(CuratedLogIndexState.deleteRetentionHour * 60)
-          >= time.milliseconds()) {
-        storeInfo = store.get(Collections.singletonList(id), EnumSet.of(StoreGetOptions.Store_Include_Deleted));
-        checkStoreInfo(storeInfo, Collections.singleton(id));
-      } else {
-        try {
-          store.get(Collections.singletonList(id), EnumSet.of(StoreGetOptions.Store_Include_Deleted));
-          fail("Should not be able to GET " + id);
-        } catch (StoreException e) {
-          assertEquals("Unexpected StoreErrorCode", StoreErrorCodes.ID_Deleted, e.getErrorCode());
-        }
-      }
-
       // with all StoreGetOptions
       storeInfo = store.get(Collections.singletonList(id), EnumSet.allOf(StoreGetOptions.class));
       checkStoreInfo(storeInfo, Collections.singleton(id));
@@ -1242,7 +1228,7 @@ public void deleteLifeVersionTest() throws StoreException {
     info = new MessageInfo(addedId, DELETE_RECORD_SIZE, true, false, false, Utils.Infinite_Time, null,
         addedId.getAccountId(), addedId.getContainerId(), time.milliseconds(), lifeVersion);
     store.delete(Collections.singletonList(info));
-    deletedKeys.put(addedId, time.milliseconds());
+    deletedKeys.add(addedId);
     undeletedKeys.remove(addedId);
     liveKeys.remove(addedId);
     StoreInfo storeInfo = store.get(Arrays.asList(addedId), EnumSet.of(StoreGetOptions.Store_Include_Deleted));
@@ -1287,7 +1273,7 @@ public void deleteLifeVersionTest() throws StoreException {
     info = new MessageInfo(addedId, DELETE_RECORD_SIZE, true, false, false, Utils.Infinite_Time, null,
         addedId.getAccountId(), addedId.getContainerId(), time.milliseconds(), MessageInfo.LIFE_VERSION_FROM_FRONTEND);
     store.delete(Collections.singletonList(info));
-    deletedKeys.put(addedId, time.milliseconds());
+    deletedKeys.add(addedId);
     undeletedKeys.remove(addedId);
     liveKeys.remove(addedId);
     storeInfo = store.get(Arrays.asList(addedId), EnumSet.of(StoreGetOptions.Store_Include_Deleted));
@@ -1437,7 +1423,7 @@ public void putErrorCasesTest() throws StoreException {
   @Test
   public void deleteErrorCasesTest() throws StoreException {
     // ID that is already deleted
-    verifyDeleteFailure(deletedKeys.keySet().iterator().next(), StoreErrorCodes.ID_Deleted);
+    verifyDeleteFailure(deletedKeys.iterator().next(), StoreErrorCodes.ID_Deleted);
     // ID that does not exist
     verifyDeleteFailure(getUniqueId(), StoreErrorCodes.ID_Not_Found);
     MockId id = getUniqueId();
@@ -1569,12 +1555,12 @@ public void ttlUpdateErrorCasesTest() throws Exception {
     inNoTtlUpdatePeriodTest();
     // ID that is already updated
     for (MockId ttlUpdated : ttlUpdatedKeys) {
-      if (!deletedKeys.containsKey(ttlUpdated)) {
+      if (!deletedKeys.contains(ttlUpdated)) {
         verifyTtlUpdateFailure(ttlUpdated, Utils.Infinite_Time, StoreErrorCodes.Already_Updated);
       }
     }
     // ID that is already deleted
-    for (MockId deleted : deletedKeys.keySet()) {
+    for (MockId deleted : deletedKeys) {
       verifyTtlUpdateFailure(deleted, Utils.Infinite_Time, StoreErrorCodes.ID_Deleted);
     }
     // Attempt to set expiry time to anything other than infinity
@@ -1802,7 +1788,7 @@ public void findMissingKeysTest() throws StoreException {
   @Test
   public void isKeyDeletedTest() throws StoreException {
     for (MockId id : allKeys.keySet()) {
-      assertEquals("Returned state is not as expected", deletedKeys.containsKey(id), store.isKeyDeleted(id));
+      assertEquals("Returned state is not as expected", deletedKeys.contains(id), store.isKeyDeleted(id));
     }
     for (MockId id : deletedAndShouldBeCompactedKeys) {
       assertTrue("Returned state is not as expected", store.isKeyDeleted(id));
@@ -3021,7 +3007,7 @@ private MessageInfo delete(MockId idToDelete, long operationTimeMs, boolean forc
       store.forceDelete(Collections.singletonList(info));
     }
 
-    deletedKeys.put(idToDelete, operationTimeMs);
+    deletedKeys.add(idToDelete);
     undeletedKeys.remove(idToDelete);
     liveKeys.remove(idToDelete);
     return info;
@@ -3095,7 +3081,7 @@ private void checkStoreInfo(StoreInfo storeInfo, Set<MockId> expectedKeys, short
       assertEquals("ContainerId mismatch", expectedInfo.getContainerId(), messageInfo.getContainerId());
       assertEquals("OperationTime mismatch", expectedInfo.getOperationTimeMs(), messageInfo.getOperationTimeMs());
       assertEquals("isTTLUpdated not as expected", ttlUpdatedKeys.contains(id), messageInfo.isTtlUpdated());
-      assertEquals("isDeleted not as expected", deletedKeys.containsKey(id), messageInfo.isDeleted());
+      assertEquals("isDeleted not as expected", deletedKeys.contains(id), messageInfo.isDeleted());
       assertEquals("isUndeleted not as expected", undeletedKeys.contains(id), messageInfo.isUndeleted());
       if (IndexValue.hasLifeVersion(lifeVersion)) {
         assertEquals("lifeVersion not as expected", lifeVersion, messageInfo.getLifeVersion());
@@ -3195,6 +3181,8 @@ private void setupTestState(boolean addTtlUpdates, boolean addUndelete) throws I
       deletes++;
       idsByLogSegment.get(2).add(idToDelete);
       // 1 DELETE for the PUT in the same segment
+      deletedKeys.add(addedId);
+      liveKeys.remove(addedId);
       delete(addedId);
       deletes++;
 
@@ -3377,6 +3365,7 @@ private void addCuratedData(long sizeToWrite, boolean addTtlUpdates) throws Stor
       // 1 DELETE for the expired PUT
       delete(id);
       deletedKeyCount++;
+      deletedKeys.add(id);
       expiredKeys.remove(id);
       idsGroupedByIndexSegment.add(idsInIndexSegment);
       idsInLogSegment.addAll(idsInIndexSegment);
@@ -3512,6 +3501,8 @@ private MockId getIdToDelete(Set<MockId> ids, boolean ttlUpdated) {
     if (deleteCandidate == null) {
       throw new IllegalStateException("Could not find a key to delete in set: " + ids);
     }
+    deletedKeys.add(deleteCandidate);
+    liveKeys.remove(deleteCandidate);
     return deleteCandidate;
   }
 
@@ -3635,7 +3626,7 @@ private void verifyGetFutures(List<Getter> getters, List<Future<CallableResult>>
         } catch (ExecutionException e) {
           StoreException storeException = (StoreException) e.getCause();
           StoreErrorCodes expectedCode = StoreErrorCodes.ID_Not_Found;
-          if (deletedKeys.containsKey(id)) {
+          if (deletedKeys.contains(id)) {
             expectedCode = StoreErrorCodes.ID_Deleted;
           } else if (expiredKeys.contains(id)) {
             expectedCode = StoreErrorCodes.TTL_Expired;