@Test public void deleteErrorCasesTest() throws StoreException { verifyDeleteFailure(deletedKeys.iterator().next(), StoreErrorCodes.ID_Deleted); verifyDeleteFailure(getUniqueId(), StoreErrorCodes.ID_Not_Found); MockId id = getUniqueId(); MessageInfo info = new MessageInfo(id, DELETE_RECORD_SIZE, id.getAccountId(), id.getContainerId(), time.milliseconds()); try { store.delete(Arrays.asList(info, info)); fail("Store DELETE should have failed"); } catch (IllegalArgumentException e) { } }
@Override public void delete(List<MessageInfo> infosToDelete) throws StoreException { checkStarted(); checkDuplicates(infosToDelete); final Timer.Context context = metrics.deleteResponse.time(); try { List<IndexValue> indexValuesPriorToDelete = new ArrayList<>(); List<IndexValue> originalPuts = new ArrayList<>(); List<Short> lifeVersions = new ArrayList<>(); Offset indexEndOffsetBeforeCheck = index.getCurrentEndOffset(); maybeCallInDeleteBetweenGetEndOffsetAndFindKey(); for (MessageInfo info : infosToDelete) { IndexValue value = index.findKey(info.getStoreKey(), new FileSpan(index.getStartOffset(), indexEndOffsetBeforeCheck)); if (value == null) { throw new StoreException("Cannot delete id " + info.getStoreKey() + " because it is not present in the index", StoreErrorCodes.ID_Not_Found); } if (!info.getStoreKey().isAccountContainerMatch(value.getAccountId(), value.getContainerId())) { if (config.storeValidateAuthorization) { throw new StoreException("DELETE authorization failure. Key: " + info.getStoreKey() + "Actually accountId: " + value.getAccountId() + "Actually containerId: " + value.getContainerId(), StoreErrorCodes.Authorization_Failure); } else { logger.warn("DELETE authorization failure. Key: {} Actually accountId: {} Actually containerId: {}", info.getStoreKey(), value.getAccountId(), value.getContainerId()); metrics.deleteAuthorizationFailureCount.inc(); } } short revisedLifeVersion = info.getLifeVersion(); if (info.getLifeVersion() == MessageInfo.LIFE_VERSION_FROM_FRONTEND) { if (value.isDelete()) { throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since it is already deleted in the index.", StoreErrorCodes.ID_Deleted); } revisedLifeVersion = value.getLifeVersion(); } else { if (value.isDelete() && value.getLifeVersion() == info.getLifeVersion()) { throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since it is already deleted in the index with lifeVersion " + value.getLifeVersion() + ".", StoreErrorCodes.ID_Deleted); } if (value.getLifeVersion() > info.getLifeVersion()) { throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since it has a higher lifeVersion than the message info: " + value.getLifeVersion() + ">" + info.getLifeVersion(), StoreErrorCodes.Life_Version_Conflict); } } indexValuesPriorToDelete.add(value); lifeVersions.add(revisedLifeVersion); if (!value.isDelete() && !value.isUndelete()) { originalPuts.add(value); } else { originalPuts.add(index.findKey(info.getStoreKey(), new FileSpan(index.getStartOffset(), value.getOffset()), EnumSet.of(PersistentIndex.IndexEntryType.PUT))); } } maybeCallBeforeSynchronization(); synchronized (storeWriteLock) { Offset currentIndexEndOffset = index.getCurrentEndOffset(); if (!currentIndexEndOffset.equals(indexEndOffsetBeforeCheck)) { FileSpan fileSpan = new FileSpan(indexEndOffsetBeforeCheck, currentIndexEndOffset); int i = 0; for (MessageInfo info : infosToDelete) { IndexValue value = index.findKey(info.getStoreKey(), fileSpan, EnumSet.allOf(PersistentIndex.IndexEntryType.class)); if (value != null) { if (value.getLifeVersion() == lifeVersions.get(i)) { if (value.isDelete()) { throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since it is already deleted in the index.", StoreErrorCodes.ID_Deleted); } } else { logger.warn("Concurrent operation for id " + info.getStoreKey() + " in store " + dataDir + ". Newly added value " + value); throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since there are concurrent operation while delete", StoreErrorCodes.Life_Version_Conflict); } indexValuesPriorToDelete.set(i, value); } i++; } } List<InputStream> inputStreams = new ArrayList<>(infosToDelete.size()); List<MessageInfo> updatedInfos = new ArrayList<>(infosToDelete.size()); int i = 0; for (MessageInfo info : infosToDelete) { MessageFormatInputStream stream = new DeleteMessageFormatInputStream(info.getStoreKey(), info.getAccountId(), info.getContainerId(), info.getOperationTimeMs(), lifeVersions.get(i)); updatedInfos.add( new MessageInfo(info.getStoreKey(), stream.getSize(), info.getAccountId(), info.getContainerId(), info.getOperationTimeMs(), info.getLifeVersion())); inputStreams.add(stream); i++; } Offset endOffsetOfLastMessage = log.getEndOffset(); MessageFormatWriteSet writeSet = new MessageFormatWriteSet(new SequenceInputStream(Collections.enumeration(inputStreams)), updatedInfos, false); writeSet.writeTo(log); logger.trace("Store : {} delete mark written to log", dataDir); int correspondingPutIndex = 0; for (MessageInfo info : updatedInfos) { FileSpan fileSpan = log.getFileSpanForMessage(endOffsetOfLastMessage, info.getSize()); IndexValue deleteIndexValue = index.markAsDeleted(info.getStoreKey(), fileSpan, null, info.getOperationTimeMs(), info.getLifeVersion()); endOffsetOfLastMessage = fileSpan.getEndOffset(); blobStoreStats.handleNewDeleteEntry(info.getStoreKey(), deleteIndexValue, originalPuts.get(correspondingPutIndex), indexValuesPriorToDelete.get(correspondingPutIndex)); correspondingPutIndex++; } logger.trace("Store : {} delete has been marked in the index ", dataDir); } onSuccess(); } catch (StoreException e) { if (e.getErrorCode() == StoreErrorCodes.IOError) { onError(); } throw e; } catch (Exception e) { throw new StoreException("Unknown error while trying to delete blobs from store " + dataDir, e, StoreErrorCodes.Unknown_Error); } finally { context.stop(); } }
BlobStore implements Store { @Override public void delete(List<MessageInfo> infosToDelete) throws StoreException { checkStarted(); checkDuplicates(infosToDelete); final Timer.Context context = metrics.deleteResponse.time(); try { List<IndexValue> indexValuesPriorToDelete = new ArrayList<>(); List<IndexValue> originalPuts = new ArrayList<>(); List<Short> lifeVersions = new ArrayList<>(); Offset indexEndOffsetBeforeCheck = index.getCurrentEndOffset(); maybeCallInDeleteBetweenGetEndOffsetAndFindKey(); for (MessageInfo info : infosToDelete) { IndexValue value = index.findKey(info.getStoreKey(), new FileSpan(index.getStartOffset(), indexEndOffsetBeforeCheck)); if (value == null) { throw new StoreException("Cannot delete id " + info.getStoreKey() + " because it is not present in the index", StoreErrorCodes.ID_Not_Found); } if (!info.getStoreKey().isAccountContainerMatch(value.getAccountId(), value.getContainerId())) { if (config.storeValidateAuthorization) { throw new StoreException("DELETE authorization failure. Key: " + info.getStoreKey() + "Actually accountId: " + value.getAccountId() + "Actually containerId: " + value.getContainerId(), StoreErrorCodes.Authorization_Failure); } else { logger.warn("DELETE authorization failure. Key: {} Actually accountId: {} Actually containerId: {}", info.getStoreKey(), value.getAccountId(), value.getContainerId()); metrics.deleteAuthorizationFailureCount.inc(); } } short revisedLifeVersion = info.getLifeVersion(); if (info.getLifeVersion() == MessageInfo.LIFE_VERSION_FROM_FRONTEND) { if (value.isDelete()) { throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since it is already deleted in the index.", StoreErrorCodes.ID_Deleted); } revisedLifeVersion = value.getLifeVersion(); } else { if (value.isDelete() && value.getLifeVersion() == info.getLifeVersion()) { throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since it is already deleted in the index with lifeVersion " + value.getLifeVersion() + ".", StoreErrorCodes.ID_Deleted); } if (value.getLifeVersion() > info.getLifeVersion()) { throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since it has a higher lifeVersion than the message info: " + value.getLifeVersion() + ">" + info.getLifeVersion(), StoreErrorCodes.Life_Version_Conflict); } } indexValuesPriorToDelete.add(value); lifeVersions.add(revisedLifeVersion); if (!value.isDelete() && !value.isUndelete()) { originalPuts.add(value); } else { originalPuts.add(index.findKey(info.getStoreKey(), new FileSpan(index.getStartOffset(), value.getOffset()), EnumSet.of(PersistentIndex.IndexEntryType.PUT))); } } maybeCallBeforeSynchronization(); synchronized (storeWriteLock) { Offset currentIndexEndOffset = index.getCurrentEndOffset(); if (!currentIndexEndOffset.equals(indexEndOffsetBeforeCheck)) { FileSpan fileSpan = new FileSpan(indexEndOffsetBeforeCheck, currentIndexEndOffset); int i = 0; for (MessageInfo info : infosToDelete) { IndexValue value = index.findKey(info.getStoreKey(), fileSpan, EnumSet.allOf(PersistentIndex.IndexEntryType.class)); if (value != null) { if (value.getLifeVersion() == lifeVersions.get(i)) { if (value.isDelete()) { throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since it is already deleted in the index.", StoreErrorCodes.ID_Deleted); } } else { logger.warn("Concurrent operation for id " + info.getStoreKey() + " in store " + dataDir + ". Newly added value " + value); throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since there are concurrent operation while delete", StoreErrorCodes.Life_Version_Conflict); } indexValuesPriorToDelete.set(i, value); } i++; } } List<InputStream> inputStreams = new ArrayList<>(infosToDelete.size()); List<MessageInfo> updatedInfos = new ArrayList<>(infosToDelete.size()); int i = 0; for (MessageInfo info : infosToDelete) { MessageFormatInputStream stream = new DeleteMessageFormatInputStream(info.getStoreKey(), info.getAccountId(), info.getContainerId(), info.getOperationTimeMs(), lifeVersions.get(i)); updatedInfos.add( new MessageInfo(info.getStoreKey(), stream.getSize(), info.getAccountId(), info.getContainerId(), info.getOperationTimeMs(), info.getLifeVersion())); inputStreams.add(stream); i++; } Offset endOffsetOfLastMessage = log.getEndOffset(); MessageFormatWriteSet writeSet = new MessageFormatWriteSet(new SequenceInputStream(Collections.enumeration(inputStreams)), updatedInfos, false); writeSet.writeTo(log); logger.trace("Store : {} delete mark written to log", dataDir); int correspondingPutIndex = 0; for (MessageInfo info : updatedInfos) { FileSpan fileSpan = log.getFileSpanForMessage(endOffsetOfLastMessage, info.getSize()); IndexValue deleteIndexValue = index.markAsDeleted(info.getStoreKey(), fileSpan, null, info.getOperationTimeMs(), info.getLifeVersion()); endOffsetOfLastMessage = fileSpan.getEndOffset(); blobStoreStats.handleNewDeleteEntry(info.getStoreKey(), deleteIndexValue, originalPuts.get(correspondingPutIndex), indexValuesPriorToDelete.get(correspondingPutIndex)); correspondingPutIndex++; } logger.trace("Store : {} delete has been marked in the index ", dataDir); } onSuccess(); } catch (StoreException e) { if (e.getErrorCode() == StoreErrorCodes.IOError) { onError(); } throw e; } catch (Exception e) { throw new StoreException("Unknown error while trying to delete blobs from store " + dataDir, e, StoreErrorCodes.Unknown_Error); } finally { context.stop(); } } }
BlobStore implements Store { @Override public void delete(List<MessageInfo> infosToDelete) throws StoreException { checkStarted(); checkDuplicates(infosToDelete); final Timer.Context context = metrics.deleteResponse.time(); try { List<IndexValue> indexValuesPriorToDelete = new ArrayList<>(); List<IndexValue> originalPuts = new ArrayList<>(); List<Short> lifeVersions = new ArrayList<>(); Offset indexEndOffsetBeforeCheck = index.getCurrentEndOffset(); maybeCallInDeleteBetweenGetEndOffsetAndFindKey(); for (MessageInfo info : infosToDelete) { IndexValue value = index.findKey(info.getStoreKey(), new FileSpan(index.getStartOffset(), indexEndOffsetBeforeCheck)); if (value == null) { throw new StoreException("Cannot delete id " + info.getStoreKey() + " because it is not present in the index", StoreErrorCodes.ID_Not_Found); } if (!info.getStoreKey().isAccountContainerMatch(value.getAccountId(), value.getContainerId())) { if (config.storeValidateAuthorization) { throw new StoreException("DELETE authorization failure. Key: " + info.getStoreKey() + "Actually accountId: " + value.getAccountId() + "Actually containerId: " + value.getContainerId(), StoreErrorCodes.Authorization_Failure); } else { logger.warn("DELETE authorization failure. Key: {} Actually accountId: {} Actually containerId: {}", info.getStoreKey(), value.getAccountId(), value.getContainerId()); metrics.deleteAuthorizationFailureCount.inc(); } } short revisedLifeVersion = info.getLifeVersion(); if (info.getLifeVersion() == MessageInfo.LIFE_VERSION_FROM_FRONTEND) { if (value.isDelete()) { throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since it is already deleted in the index.", StoreErrorCodes.ID_Deleted); } revisedLifeVersion = value.getLifeVersion(); } else { if (value.isDelete() && value.getLifeVersion() == info.getLifeVersion()) { throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since it is already deleted in the index with lifeVersion " + value.getLifeVersion() + ".", StoreErrorCodes.ID_Deleted); } if (value.getLifeVersion() > info.getLifeVersion()) { throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since it has a higher lifeVersion than the message info: " + value.getLifeVersion() + ">" + info.getLifeVersion(), StoreErrorCodes.Life_Version_Conflict); } } indexValuesPriorToDelete.add(value); lifeVersions.add(revisedLifeVersion); if (!value.isDelete() && !value.isUndelete()) { originalPuts.add(value); } else { originalPuts.add(index.findKey(info.getStoreKey(), new FileSpan(index.getStartOffset(), value.getOffset()), EnumSet.of(PersistentIndex.IndexEntryType.PUT))); } } maybeCallBeforeSynchronization(); synchronized (storeWriteLock) { Offset currentIndexEndOffset = index.getCurrentEndOffset(); if (!currentIndexEndOffset.equals(indexEndOffsetBeforeCheck)) { FileSpan fileSpan = new FileSpan(indexEndOffsetBeforeCheck, currentIndexEndOffset); int i = 0; for (MessageInfo info : infosToDelete) { IndexValue value = index.findKey(info.getStoreKey(), fileSpan, EnumSet.allOf(PersistentIndex.IndexEntryType.class)); if (value != null) { if (value.getLifeVersion() == lifeVersions.get(i)) { if (value.isDelete()) { throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since it is already deleted in the index.", StoreErrorCodes.ID_Deleted); } } else { logger.warn("Concurrent operation for id " + info.getStoreKey() + " in store " + dataDir + ". Newly added value " + value); throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since there are concurrent operation while delete", StoreErrorCodes.Life_Version_Conflict); } indexValuesPriorToDelete.set(i, value); } i++; } } List<InputStream> inputStreams = new ArrayList<>(infosToDelete.size()); List<MessageInfo> updatedInfos = new ArrayList<>(infosToDelete.size()); int i = 0; for (MessageInfo info : infosToDelete) { MessageFormatInputStream stream = new DeleteMessageFormatInputStream(info.getStoreKey(), info.getAccountId(), info.getContainerId(), info.getOperationTimeMs(), lifeVersions.get(i)); updatedInfos.add( new MessageInfo(info.getStoreKey(), stream.getSize(), info.getAccountId(), info.getContainerId(), info.getOperationTimeMs(), info.getLifeVersion())); inputStreams.add(stream); i++; } Offset endOffsetOfLastMessage = log.getEndOffset(); MessageFormatWriteSet writeSet = new MessageFormatWriteSet(new SequenceInputStream(Collections.enumeration(inputStreams)), updatedInfos, false); writeSet.writeTo(log); logger.trace("Store : {} delete mark written to log", dataDir); int correspondingPutIndex = 0; for (MessageInfo info : updatedInfos) { FileSpan fileSpan = log.getFileSpanForMessage(endOffsetOfLastMessage, info.getSize()); IndexValue deleteIndexValue = index.markAsDeleted(info.getStoreKey(), fileSpan, null, info.getOperationTimeMs(), info.getLifeVersion()); endOffsetOfLastMessage = fileSpan.getEndOffset(); blobStoreStats.handleNewDeleteEntry(info.getStoreKey(), deleteIndexValue, originalPuts.get(correspondingPutIndex), indexValuesPriorToDelete.get(correspondingPutIndex)); correspondingPutIndex++; } logger.trace("Store : {} delete has been marked in the index ", dataDir); } onSuccess(); } catch (StoreException e) { if (e.getErrorCode() == StoreErrorCodes.IOError) { onError(); } throw e; } catch (Exception e) { throw new StoreException("Unknown error while trying to delete blobs from store " + dataDir, e, StoreErrorCodes.Unknown_Error); } finally { context.stop(); } } BlobStore(ReplicaId replicaId, StoreConfig config, ScheduledExecutorService taskScheduler,
      ScheduledExecutorService longLivedTaskScheduler, DiskIOScheduler diskIOScheduler,
      DiskSpaceAllocator diskSpaceAllocator, StoreMetrics metrics, StoreMetrics storeUnderCompactionMetrics,
      StoreKeyFactory factory, MessageStoreRecovery recovery, MessageStoreHardDelete hardDelete,
      List<ReplicaStatusDelegate> replicaStatusDelegates, Time time, AccountService accountService);   BlobStore(String storeId, StoreConfig config, ScheduledExecutorService taskScheduler,
      ScheduledExecutorService longLivedTaskScheduler, DiskIOScheduler diskIOScheduler,
      DiskSpaceAllocator diskSpaceAllocator, StoreMetrics metrics, StoreMetrics storeUnderCompactionMetrics,
      String dataDir, long capacityInBytes, StoreKeyFactory factory, MessageStoreRecovery recovery,
      MessageStoreHardDelete hardDelete, Time time);   BlobStore(ReplicaId replicaId, String storeId, StoreConfig config, ScheduledExecutorService taskScheduler,
      ScheduledExecutorService longLivedTaskScheduler, DiskIOScheduler diskIOScheduler,
      DiskSpaceAllocator diskSpaceAllocator, StoreMetrics metrics, StoreMetrics storeUnderCompactionMetrics,
      String dataDir, long capacityInBytes, StoreKeyFactory factory, MessageStoreRecovery recovery,
      MessageStoreHardDelete hardDelete, List<ReplicaStatusDelegate> replicaStatusDelegates, Time time,
      AccountService accountService, BlobStoreStats blobStoreStats); }
BlobStore implements Store { @Override public void delete(List<MessageInfo> infosToDelete) throws StoreException { checkStarted(); checkDuplicates(infosToDelete); final Timer.Context context = metrics.deleteResponse.time(); try { List<IndexValue> indexValuesPriorToDelete = new ArrayList<>(); List<IndexValue> originalPuts = new ArrayList<>(); List<Short> lifeVersions = new ArrayList<>(); Offset indexEndOffsetBeforeCheck = index.getCurrentEndOffset(); maybeCallInDeleteBetweenGetEndOffsetAndFindKey(); for (MessageInfo info : infosToDelete) { IndexValue value = index.findKey(info.getStoreKey(), new FileSpan(index.getStartOffset(), indexEndOffsetBeforeCheck)); if (value == null) { throw new StoreException("Cannot delete id " + info.getStoreKey() + " because it is not present in the index", StoreErrorCodes.ID_Not_Found); } if (!info.getStoreKey().isAccountContainerMatch(value.getAccountId(), value.getContainerId())) { if (config.storeValidateAuthorization) { throw new StoreException("DELETE authorization failure. Key: " + info.getStoreKey() + "Actually accountId: " + value.getAccountId() + "Actually containerId: " + value.getContainerId(), StoreErrorCodes.Authorization_Failure); } else { logger.warn("DELETE authorization failure. Key: {} Actually accountId: {} Actually containerId: {}", info.getStoreKey(), value.getAccountId(), value.getContainerId()); metrics.deleteAuthorizationFailureCount.inc(); } } short revisedLifeVersion = info.getLifeVersion(); if (info.getLifeVersion() == MessageInfo.LIFE_VERSION_FROM_FRONTEND) { if (value.isDelete()) { throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since it is already deleted in the index.", StoreErrorCodes.ID_Deleted); } revisedLifeVersion = value.getLifeVersion(); } else { if (value.isDelete() && value.getLifeVersion() == info.getLifeVersion()) { throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since it is already deleted in the index with lifeVersion " + value.getLifeVersion() + ".", StoreErrorCodes.ID_Deleted); } if (value.getLifeVersion() > info.getLifeVersion()) { throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since it has a higher lifeVersion than the message info: " + value.getLifeVersion() + ">" + info.getLifeVersion(), StoreErrorCodes.Life_Version_Conflict); } } indexValuesPriorToDelete.add(value); lifeVersions.add(revisedLifeVersion); if (!value.isDelete() && !value.isUndelete()) { originalPuts.add(value); } else { originalPuts.add(index.findKey(info.getStoreKey(), new FileSpan(index.getStartOffset(), value.getOffset()), EnumSet.of(PersistentIndex.IndexEntryType.PUT))); } } maybeCallBeforeSynchronization(); synchronized (storeWriteLock) { Offset currentIndexEndOffset = index.getCurrentEndOffset(); if (!currentIndexEndOffset.equals(indexEndOffsetBeforeCheck)) { FileSpan fileSpan = new FileSpan(indexEndOffsetBeforeCheck, currentIndexEndOffset); int i = 0; for (MessageInfo info : infosToDelete) { IndexValue value = index.findKey(info.getStoreKey(), fileSpan, EnumSet.allOf(PersistentIndex.IndexEntryType.class)); if (value != null) { if (value.getLifeVersion() == lifeVersions.get(i)) { if (value.isDelete()) { throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since it is already deleted in the index.", StoreErrorCodes.ID_Deleted); } } else { logger.warn("Concurrent operation for id " + info.getStoreKey() + " in store " + dataDir + ". Newly added value " + value); throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since there are concurrent operation while delete", StoreErrorCodes.Life_Version_Conflict); } indexValuesPriorToDelete.set(i, value); } i++; } } List<InputStream> inputStreams = new ArrayList<>(infosToDelete.size()); List<MessageInfo> updatedInfos = new ArrayList<>(infosToDelete.size()); int i = 0; for (MessageInfo info : infosToDelete) { MessageFormatInputStream stream = new DeleteMessageFormatInputStream(info.getStoreKey(), info.getAccountId(), info.getContainerId(), info.getOperationTimeMs(), lifeVersions.get(i)); updatedInfos.add( new MessageInfo(info.getStoreKey(), stream.getSize(), info.getAccountId(), info.getContainerId(), info.getOperationTimeMs(), info.getLifeVersion())); inputStreams.add(stream); i++; } Offset endOffsetOfLastMessage = log.getEndOffset(); MessageFormatWriteSet writeSet = new MessageFormatWriteSet(new SequenceInputStream(Collections.enumeration(inputStreams)), updatedInfos, false); writeSet.writeTo(log); logger.trace("Store : {} delete mark written to log", dataDir); int correspondingPutIndex = 0; for (MessageInfo info : updatedInfos) { FileSpan fileSpan = log.getFileSpanForMessage(endOffsetOfLastMessage, info.getSize()); IndexValue deleteIndexValue = index.markAsDeleted(info.getStoreKey(), fileSpan, null, info.getOperationTimeMs(), info.getLifeVersion()); endOffsetOfLastMessage = fileSpan.getEndOffset(); blobStoreStats.handleNewDeleteEntry(info.getStoreKey(), deleteIndexValue, originalPuts.get(correspondingPutIndex), indexValuesPriorToDelete.get(correspondingPutIndex)); correspondingPutIndex++; } logger.trace("Store : {} delete has been marked in the index ", dataDir); } onSuccess(); } catch (StoreException e) { if (e.getErrorCode() == StoreErrorCodes.IOError) { onError(); } throw e; } catch (Exception e) { throw new StoreException("Unknown error while trying to delete blobs from store " + dataDir, e, StoreErrorCodes.Unknown_Error); } finally { context.stop(); } } BlobStore(ReplicaId replicaId, StoreConfig config, ScheduledExecutorService taskScheduler,
      ScheduledExecutorService longLivedTaskScheduler, DiskIOScheduler diskIOScheduler,
      DiskSpaceAllocator diskSpaceAllocator, StoreMetrics metrics, StoreMetrics storeUnderCompactionMetrics,
      StoreKeyFactory factory, MessageStoreRecovery recovery, MessageStoreHardDelete hardDelete,
      List<ReplicaStatusDelegate> replicaStatusDelegates, Time time, AccountService accountService);   BlobStore(String storeId, StoreConfig config, ScheduledExecutorService taskScheduler,
      ScheduledExecutorService longLivedTaskScheduler, DiskIOScheduler diskIOScheduler,
      DiskSpaceAllocator diskSpaceAllocator, StoreMetrics metrics, StoreMetrics storeUnderCompactionMetrics,
      String dataDir, long capacityInBytes, StoreKeyFactory factory, MessageStoreRecovery recovery,
      MessageStoreHardDelete hardDelete, Time time);   BlobStore(ReplicaId replicaId, String storeId, StoreConfig config, ScheduledExecutorService taskScheduler,
      ScheduledExecutorService longLivedTaskScheduler, DiskIOScheduler diskIOScheduler,
      DiskSpaceAllocator diskSpaceAllocator, StoreMetrics metrics, StoreMetrics storeUnderCompactionMetrics,
      String dataDir, long capacityInBytes, StoreKeyFactory factory, MessageStoreRecovery recovery,
      MessageStoreHardDelete hardDelete, List<ReplicaStatusDelegate> replicaStatusDelegates, Time time,
      AccountService accountService, BlobStoreStats blobStoreStats); @Override void start(); @Override StoreInfo get(List<? extends StoreKey> ids, EnumSet<StoreGetOptions> storeGetOptions); @Override void put(MessageWriteSet messageSetToWrite); @Override void delete(List<MessageInfo> infosToDelete); @Override void updateTtl(List<MessageInfo> infosToUpdate); @Override short undelete(MessageInfo info); @Override FindInfo findEntriesSince(FindToken token, long maxTotalSizeOfEntries); @Override Set<StoreKey> findMissingKeys(List<StoreKey> keys); @Override MessageInfo findKey(StoreKey key); @Override StoreStats getStoreStats(); @Override boolean isKeyDeleted(StoreKey key); @Override long getSizeInBytes(); long getLogEndOffsetInBytes(); @Override boolean isEmpty(); @Override boolean isBootstrapInProgress(); @Override boolean isDecommissionInProgress(); @Override void completeBootstrap(); @Override void setCurrentState(ReplicaState state); @Override ReplicaState getCurrentState(); @Override boolean recoverFromDecommission(); @Override boolean disabledOnError(); void deleteStoreFiles(); @Override long getEndPositionOfLastPut(); List<ReplicaStatusDelegate> getReplicaStatusDelegates(); @Override void shutdown(); @Override boolean isStarted(); @Override String toString(); }
BlobStore implements Store { @Override public void delete(List<MessageInfo> infosToDelete) throws StoreException { checkStarted(); checkDuplicates(infosToDelete); final Timer.Context context = metrics.deleteResponse.time(); try { List<IndexValue> indexValuesPriorToDelete = new ArrayList<>(); List<IndexValue> originalPuts = new ArrayList<>(); List<Short> lifeVersions = new ArrayList<>(); Offset indexEndOffsetBeforeCheck = index.getCurrentEndOffset(); maybeCallInDeleteBetweenGetEndOffsetAndFindKey(); for (MessageInfo info : infosToDelete) { IndexValue value = index.findKey(info.getStoreKey(), new FileSpan(index.getStartOffset(), indexEndOffsetBeforeCheck)); if (value == null) { throw new StoreException("Cannot delete id " + info.getStoreKey() + " because it is not present in the index", StoreErrorCodes.ID_Not_Found); } if (!info.getStoreKey().isAccountContainerMatch(value.getAccountId(), value.getContainerId())) { if (config.storeValidateAuthorization) { throw new StoreException("DELETE authorization failure. Key: " + info.getStoreKey() + "Actually accountId: " + value.getAccountId() + "Actually containerId: " + value.getContainerId(), StoreErrorCodes.Authorization_Failure); } else { logger.warn("DELETE authorization failure. Key: {} Actually accountId: {} Actually containerId: {}", info.getStoreKey(), value.getAccountId(), value.getContainerId()); metrics.deleteAuthorizationFailureCount.inc(); } } short revisedLifeVersion = info.getLifeVersion(); if (info.getLifeVersion() == MessageInfo.LIFE_VERSION_FROM_FRONTEND) { if (value.isDelete()) { throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since it is already deleted in the index.", StoreErrorCodes.ID_Deleted); } revisedLifeVersion = value.getLifeVersion(); } else { if (value.isDelete() && value.getLifeVersion() == info.getLifeVersion()) { throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since it is already deleted in the index with lifeVersion " + value.getLifeVersion() + ".", StoreErrorCodes.ID_Deleted); } if (value.getLifeVersion() > info.getLifeVersion()) { throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since it has a higher lifeVersion than the message info: " + value.getLifeVersion() + ">" + info.getLifeVersion(), StoreErrorCodes.Life_Version_Conflict); } } indexValuesPriorToDelete.add(value); lifeVersions.add(revisedLifeVersion); if (!value.isDelete() && !value.isUndelete()) { originalPuts.add(value); } else { originalPuts.add(index.findKey(info.getStoreKey(), new FileSpan(index.getStartOffset(), value.getOffset()), EnumSet.of(PersistentIndex.IndexEntryType.PUT))); } } maybeCallBeforeSynchronization(); synchronized (storeWriteLock) { Offset currentIndexEndOffset = index.getCurrentEndOffset(); if (!currentIndexEndOffset.equals(indexEndOffsetBeforeCheck)) { FileSpan fileSpan = new FileSpan(indexEndOffsetBeforeCheck, currentIndexEndOffset); int i = 0; for (MessageInfo info : infosToDelete) { IndexValue value = index.findKey(info.getStoreKey(), fileSpan, EnumSet.allOf(PersistentIndex.IndexEntryType.class)); if (value != null) { if (value.getLifeVersion() == lifeVersions.get(i)) { if (value.isDelete()) { throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since it is already deleted in the index.", StoreErrorCodes.ID_Deleted); } } else { logger.warn("Concurrent operation for id " + info.getStoreKey() + " in store " + dataDir + ". Newly added value " + value); throw new StoreException( "Cannot delete id " + info.getStoreKey() + " since there are concurrent operation while delete", StoreErrorCodes.Life_Version_Conflict); } indexValuesPriorToDelete.set(i, value); } i++; } } List<InputStream> inputStreams = new ArrayList<>(infosToDelete.size()); List<MessageInfo> updatedInfos = new ArrayList<>(infosToDelete.size()); int i = 0; for (MessageInfo info : infosToDelete) { MessageFormatInputStream stream = new DeleteMessageFormatInputStream(info.getStoreKey(), info.getAccountId(), info.getContainerId(), info.getOperationTimeMs(), lifeVersions.get(i)); updatedInfos.add( new MessageInfo(info.getStoreKey(), stream.getSize(), info.getAccountId(), info.getContainerId(), info.getOperationTimeMs(), info.getLifeVersion())); inputStreams.add(stream); i++; } Offset endOffsetOfLastMessage = log.getEndOffset(); MessageFormatWriteSet writeSet = new MessageFormatWriteSet(new SequenceInputStream(Collections.enumeration(inputStreams)), updatedInfos, false); writeSet.writeTo(log); logger.trace("Store : {} delete mark written to log", dataDir); int correspondingPutIndex = 0; for (MessageInfo info : updatedInfos) { FileSpan fileSpan = log.getFileSpanForMessage(endOffsetOfLastMessage, info.getSize()); IndexValue deleteIndexValue = index.markAsDeleted(info.getStoreKey(), fileSpan, null, info.getOperationTimeMs(), info.getLifeVersion()); endOffsetOfLastMessage = fileSpan.getEndOffset(); blobStoreStats.handleNewDeleteEntry(info.getStoreKey(), deleteIndexValue, originalPuts.get(correspondingPutIndex), indexValuesPriorToDelete.get(correspondingPutIndex)); correspondingPutIndex++; } logger.trace("Store : {} delete has been marked in the index ", dataDir); } onSuccess(); } catch (StoreException e) { if (e.getErrorCode() == StoreErrorCodes.IOError) { onError(); } throw e; } catch (Exception e) { throw new StoreException("Unknown error while trying to delete blobs from store " + dataDir, e, StoreErrorCodes.Unknown_Error); } finally { context.stop(); } } BlobStore(ReplicaId replicaId, StoreConfig config, ScheduledExecutorService taskScheduler,
      ScheduledExecutorService longLivedTaskScheduler, DiskIOScheduler diskIOScheduler,
      DiskSpaceAllocator diskSpaceAllocator, StoreMetrics metrics, StoreMetrics storeUnderCompactionMetrics,
      StoreKeyFactory factory, MessageStoreRecovery recovery, MessageStoreHardDelete hardDelete,
      List<ReplicaStatusDelegate> replicaStatusDelegates, Time time, AccountService accountService);   BlobStore(String storeId, StoreConfig config, ScheduledExecutorService taskScheduler,
      ScheduledExecutorService longLivedTaskScheduler, DiskIOScheduler diskIOScheduler,
      DiskSpaceAllocator diskSpaceAllocator, StoreMetrics metrics, StoreMetrics storeUnderCompactionMetrics,
      String dataDir, long capacityInBytes, StoreKeyFactory factory, MessageStoreRecovery recovery,
      MessageStoreHardDelete hardDelete, Time time);   BlobStore(ReplicaId replicaId, String storeId, StoreConfig config, ScheduledExecutorService taskScheduler,
      ScheduledExecutorService longLivedTaskScheduler, DiskIOScheduler diskIOScheduler,
      DiskSpaceAllocator diskSpaceAllocator, StoreMetrics metrics, StoreMetrics storeUnderCompactionMetrics,
      String dataDir, long capacityInBytes, StoreKeyFactory factory, MessageStoreRecovery recovery,
      MessageStoreHardDelete hardDelete, List<ReplicaStatusDelegate> replicaStatusDelegates, Time time,
      AccountService accountService, BlobStoreStats blobStoreStats); @Override void start(); @Override StoreInfo get(List<? extends StoreKey> ids, EnumSet<StoreGetOptions> storeGetOptions); @Override void put(MessageWriteSet messageSetToWrite); @Override void delete(List<MessageInfo> infosToDelete); @Override void updateTtl(List<MessageInfo> infosToUpdate); @Override short undelete(MessageInfo info); @Override FindInfo findEntriesSince(FindToken token, long maxTotalSizeOfEntries); @Override Set<StoreKey> findMissingKeys(List<StoreKey> keys); @Override MessageInfo findKey(StoreKey key); @Override StoreStats getStoreStats(); @Override boolean isKeyDeleted(StoreKey key); @Override long getSizeInBytes(); long getLogEndOffsetInBytes(); @Override boolean isEmpty(); @Override boolean isBootstrapInProgress(); @Override boolean isDecommissionInProgress(); @Override void completeBootstrap(); @Override void setCurrentState(ReplicaState state); @Override ReplicaState getCurrentState(); @Override boolean recoverFromDecommission(); @Override boolean disabledOnError(); void deleteStoreFiles(); @Override long getEndPositionOfLastPut(); List<ReplicaStatusDelegate> getReplicaStatusDelegates(); @Override void shutdown(); @Override boolean isStarted(); @Override String toString();  }