@Test public void ttlUpdateErrorCasesTest() throws Exception { verifyTtlUpdateFailure(getUniqueId(), Utils.Infinite_Time, StoreErrorCodes.ID_Not_Found); MockId expiredId = null; for (MockId expired : expiredKeys) { verifyTtlUpdateFailure(expired, Utils.Infinite_Time, StoreErrorCodes.Update_Not_Allowed); expiredId = expired; } assertNotNull(expiredId); MessageInfo info = new MessageInfo(expiredId, TTL_UPDATE_RECORD_SIZE, false, true, false, Utils.Infinite_Time, null, expiredId.getAccountId(), expiredId.getContainerId(), time.milliseconds(), (short) 1); store.updateTtl(Arrays.asList(info)); inNoTtlUpdatePeriodTest(); for (MockId ttlUpdated : ttlUpdatedKeys) { if (!deletedKeys.contains(ttlUpdated)) { verifyTtlUpdateFailure(ttlUpdated, Utils.Infinite_Time, StoreErrorCodes.Already_Updated); } } for (MockId deleted : deletedKeys) { verifyTtlUpdateFailure(deleted, Utils.Infinite_Time, StoreErrorCodes.ID_Deleted); } MockId id = getIdToTtlUpdate(liveKeys); verifyTtlUpdateFailure(id, time.milliseconds() + 5, StoreErrorCodes.Update_Not_Allowed); ttlUpdateAuthorizationFailureTest(); id = getUniqueId(); info = new MessageInfo(id, TTL_UPDATE_RECORD_SIZE, false, true, Utils.Infinite_Time, id.getAccountId(), id.getContainerId(), time.milliseconds()); try { store.updateTtl(Arrays.asList(info, info)); fail("Store TTL UPDATE should have failed"); } catch (IllegalArgumentException e) { } }
@Override public void updateTtl(List<MessageInfo> infosToUpdate) throws StoreException { checkStarted(); checkDuplicates(infosToUpdate); final Timer.Context context = metrics.ttlUpdateResponse.time(); try { List<IndexValue> indexValuesToUpdate = new ArrayList<>(); List<Short> lifeVersions = new ArrayList<>(); Offset indexEndOffsetBeforeCheck = index.getCurrentEndOffset(); for (MessageInfo info : infosToUpdate) { if (info.getExpirationTimeInMs() != Utils.Infinite_Time) { throw new StoreException("BlobStore only supports removing the expiration time", StoreErrorCodes.Update_Not_Allowed); } IndexValue value = index.findKey(info.getStoreKey(), new FileSpan(index.getStartOffset(), indexEndOffsetBeforeCheck)); if (value == null) { throw new StoreException("Cannot update TTL of " + info.getStoreKey() + " since it's not in the index", StoreErrorCodes.ID_Not_Found); } else if (!info.getStoreKey().isAccountContainerMatch(value.getAccountId(), value.getContainerId())) { if (config.storeValidateAuthorization) { throw new StoreException( "UPDATE authorization failure. Key: " + info.getStoreKey() + " AccountId in store: " + value.getAccountId() + " ContainerId in store: " + value.getContainerId(), StoreErrorCodes.Authorization_Failure); } else { logger.warn("UPDATE authorization failure. Key: {} AccountId in store: {} ContainerId in store: {}", info.getStoreKey(), value.getAccountId(), value.getContainerId()); metrics.ttlUpdateAuthorizationFailureCount.inc(); } } else if (value.isDelete()) { throw new StoreException( "Cannot update TTL of " + info.getStoreKey() + " since it is already deleted in the index.", StoreErrorCodes.ID_Deleted); } else if (value.isTtlUpdate()) { throw new StoreException("TTL of " + info.getStoreKey() + " is already updated in the index.", StoreErrorCodes.Already_Updated); } else if (!IndexValue.hasLifeVersion(info.getLifeVersion()) && value.getExpiresAtMs() != Utils.Infinite_Time && value.getExpiresAtMs() < info.getOperationTimeMs() + ttlUpdateBufferTimeMs) { throw new StoreException( "TTL of " + info.getStoreKey() + " cannot be updated because it is too close to expiry. Op time (ms): " + info.getOperationTimeMs() + ". ExpiresAtMs: " + value.getExpiresAtMs(), StoreErrorCodes.Update_Not_Allowed); } indexValuesToUpdate.add(value); lifeVersions.add(value.getLifeVersion()); } maybeCallBeforeSynchronization(); synchronized (storeWriteLock) { Offset currentIndexEndOffset = index.getCurrentEndOffset(); if (!currentIndexEndOffset.equals(indexEndOffsetBeforeCheck)) { FileSpan fileSpan = new FileSpan(indexEndOffsetBeforeCheck, currentIndexEndOffset); for (MessageInfo info : infosToUpdate) { IndexValue value = index.findKey(info.getStoreKey(), fileSpan, EnumSet.allOf(PersistentIndex.IndexEntryType.class)); if (value != null) { if (value.isDelete()) { throw new StoreException( "Cannot update TTL of " + info.getStoreKey() + " since it is already deleted in the index.", StoreErrorCodes.ID_Deleted); } else if (value.isTtlUpdate()) { throw new StoreException("TTL of " + info.getStoreKey() + " is already updated in the index.", StoreErrorCodes.Already_Updated); } } } } List<InputStream> inputStreams = new ArrayList<>(infosToUpdate.size()); List<MessageInfo> updatedInfos = new ArrayList<>(infosToUpdate.size()); int i = 0; for (MessageInfo info : infosToUpdate) { MessageFormatInputStream stream = new TtlUpdateMessageFormatInputStream(info.getStoreKey(), info.getAccountId(), info.getContainerId(), info.getExpirationTimeInMs(), 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 : {} ttl update mark written to log", dataDir); int correspondingPutIndex = 0; for (MessageInfo info : updatedInfos) { FileSpan fileSpan = log.getFileSpanForMessage(endOffsetOfLastMessage, info.getSize()); IndexValue ttlUpdateValue = index.markAsPermanent(info.getStoreKey(), fileSpan, null, info.getOperationTimeMs(), MessageInfo.LIFE_VERSION_FROM_FRONTEND); endOffsetOfLastMessage = fileSpan.getEndOffset(); blobStoreStats.handleNewTtlUpdateEntry(ttlUpdateValue, indexValuesToUpdate.get(correspondingPutIndex++)); } logger.trace("Store : {} ttl update 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 update ttl of blobs from store " + dataDir, e, StoreErrorCodes.Unknown_Error); } finally { context.stop(); } }
BlobStore implements Store { @Override public void updateTtl(List<MessageInfo> infosToUpdate) throws StoreException { checkStarted(); checkDuplicates(infosToUpdate); final Timer.Context context = metrics.ttlUpdateResponse.time(); try { List<IndexValue> indexValuesToUpdate = new ArrayList<>(); List<Short> lifeVersions = new ArrayList<>(); Offset indexEndOffsetBeforeCheck = index.getCurrentEndOffset(); for (MessageInfo info : infosToUpdate) { if (info.getExpirationTimeInMs() != Utils.Infinite_Time) { throw new StoreException("BlobStore only supports removing the expiration time", StoreErrorCodes.Update_Not_Allowed); } IndexValue value = index.findKey(info.getStoreKey(), new FileSpan(index.getStartOffset(), indexEndOffsetBeforeCheck)); if (value == null) { throw new StoreException("Cannot update TTL of " + info.getStoreKey() + " since it's not in the index", StoreErrorCodes.ID_Not_Found); } else if (!info.getStoreKey().isAccountContainerMatch(value.getAccountId(), value.getContainerId())) { if (config.storeValidateAuthorization) { throw new StoreException( "UPDATE authorization failure. Key: " + info.getStoreKey() + " AccountId in store: " + value.getAccountId() + " ContainerId in store: " + value.getContainerId(), StoreErrorCodes.Authorization_Failure); } else { logger.warn("UPDATE authorization failure. Key: {} AccountId in store: {} ContainerId in store: {}", info.getStoreKey(), value.getAccountId(), value.getContainerId()); metrics.ttlUpdateAuthorizationFailureCount.inc(); } } else if (value.isDelete()) { throw new StoreException( "Cannot update TTL of " + info.getStoreKey() + " since it is already deleted in the index.", StoreErrorCodes.ID_Deleted); } else if (value.isTtlUpdate()) { throw new StoreException("TTL of " + info.getStoreKey() + " is already updated in the index.", StoreErrorCodes.Already_Updated); } else if (!IndexValue.hasLifeVersion(info.getLifeVersion()) && value.getExpiresAtMs() != Utils.Infinite_Time && value.getExpiresAtMs() < info.getOperationTimeMs() + ttlUpdateBufferTimeMs) { throw new StoreException( "TTL of " + info.getStoreKey() + " cannot be updated because it is too close to expiry. Op time (ms): " + info.getOperationTimeMs() + ". ExpiresAtMs: " + value.getExpiresAtMs(), StoreErrorCodes.Update_Not_Allowed); } indexValuesToUpdate.add(value); lifeVersions.add(value.getLifeVersion()); } maybeCallBeforeSynchronization(); synchronized (storeWriteLock) { Offset currentIndexEndOffset = index.getCurrentEndOffset(); if (!currentIndexEndOffset.equals(indexEndOffsetBeforeCheck)) { FileSpan fileSpan = new FileSpan(indexEndOffsetBeforeCheck, currentIndexEndOffset); for (MessageInfo info : infosToUpdate) { IndexValue value = index.findKey(info.getStoreKey(), fileSpan, EnumSet.allOf(PersistentIndex.IndexEntryType.class)); if (value != null) { if (value.isDelete()) { throw new StoreException( "Cannot update TTL of " + info.getStoreKey() + " since it is already deleted in the index.", StoreErrorCodes.ID_Deleted); } else if (value.isTtlUpdate()) { throw new StoreException("TTL of " + info.getStoreKey() + " is already updated in the index.", StoreErrorCodes.Already_Updated); } } } } List<InputStream> inputStreams = new ArrayList<>(infosToUpdate.size()); List<MessageInfo> updatedInfos = new ArrayList<>(infosToUpdate.size()); int i = 0; for (MessageInfo info : infosToUpdate) { MessageFormatInputStream stream = new TtlUpdateMessageFormatInputStream(info.getStoreKey(), info.getAccountId(), info.getContainerId(), info.getExpirationTimeInMs(), 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 : {} ttl update mark written to log", dataDir); int correspondingPutIndex = 0; for (MessageInfo info : updatedInfos) { FileSpan fileSpan = log.getFileSpanForMessage(endOffsetOfLastMessage, info.getSize()); IndexValue ttlUpdateValue = index.markAsPermanent(info.getStoreKey(), fileSpan, null, info.getOperationTimeMs(), MessageInfo.LIFE_VERSION_FROM_FRONTEND); endOffsetOfLastMessage = fileSpan.getEndOffset(); blobStoreStats.handleNewTtlUpdateEntry(ttlUpdateValue, indexValuesToUpdate.get(correspondingPutIndex++)); } logger.trace("Store : {} ttl update 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 update ttl of blobs from store " + dataDir, e, StoreErrorCodes.Unknown_Error); } finally { context.stop(); } } }
BlobStore implements Store { @Override public void updateTtl(List<MessageInfo> infosToUpdate) throws StoreException { checkStarted(); checkDuplicates(infosToUpdate); final Timer.Context context = metrics.ttlUpdateResponse.time(); try { List<IndexValue> indexValuesToUpdate = new ArrayList<>(); List<Short> lifeVersions = new ArrayList<>(); Offset indexEndOffsetBeforeCheck = index.getCurrentEndOffset(); for (MessageInfo info : infosToUpdate) { if (info.getExpirationTimeInMs() != Utils.Infinite_Time) { throw new StoreException("BlobStore only supports removing the expiration time", StoreErrorCodes.Update_Not_Allowed); } IndexValue value = index.findKey(info.getStoreKey(), new FileSpan(index.getStartOffset(), indexEndOffsetBeforeCheck)); if (value == null) { throw new StoreException("Cannot update TTL of " + info.getStoreKey() + " since it's not in the index", StoreErrorCodes.ID_Not_Found); } else if (!info.getStoreKey().isAccountContainerMatch(value.getAccountId(), value.getContainerId())) { if (config.storeValidateAuthorization) { throw new StoreException( "UPDATE authorization failure. Key: " + info.getStoreKey() + " AccountId in store: " + value.getAccountId() + " ContainerId in store: " + value.getContainerId(), StoreErrorCodes.Authorization_Failure); } else { logger.warn("UPDATE authorization failure. Key: {} AccountId in store: {} ContainerId in store: {}", info.getStoreKey(), value.getAccountId(), value.getContainerId()); metrics.ttlUpdateAuthorizationFailureCount.inc(); } } else if (value.isDelete()) { throw new StoreException( "Cannot update TTL of " + info.getStoreKey() + " since it is already deleted in the index.", StoreErrorCodes.ID_Deleted); } else if (value.isTtlUpdate()) { throw new StoreException("TTL of " + info.getStoreKey() + " is already updated in the index.", StoreErrorCodes.Already_Updated); } else if (!IndexValue.hasLifeVersion(info.getLifeVersion()) && value.getExpiresAtMs() != Utils.Infinite_Time && value.getExpiresAtMs() < info.getOperationTimeMs() + ttlUpdateBufferTimeMs) { throw new StoreException( "TTL of " + info.getStoreKey() + " cannot be updated because it is too close to expiry. Op time (ms): " + info.getOperationTimeMs() + ". ExpiresAtMs: " + value.getExpiresAtMs(), StoreErrorCodes.Update_Not_Allowed); } indexValuesToUpdate.add(value); lifeVersions.add(value.getLifeVersion()); } maybeCallBeforeSynchronization(); synchronized (storeWriteLock) { Offset currentIndexEndOffset = index.getCurrentEndOffset(); if (!currentIndexEndOffset.equals(indexEndOffsetBeforeCheck)) { FileSpan fileSpan = new FileSpan(indexEndOffsetBeforeCheck, currentIndexEndOffset); for (MessageInfo info : infosToUpdate) { IndexValue value = index.findKey(info.getStoreKey(), fileSpan, EnumSet.allOf(PersistentIndex.IndexEntryType.class)); if (value != null) { if (value.isDelete()) { throw new StoreException( "Cannot update TTL of " + info.getStoreKey() + " since it is already deleted in the index.", StoreErrorCodes.ID_Deleted); } else if (value.isTtlUpdate()) { throw new StoreException("TTL of " + info.getStoreKey() + " is already updated in the index.", StoreErrorCodes.Already_Updated); } } } } List<InputStream> inputStreams = new ArrayList<>(infosToUpdate.size()); List<MessageInfo> updatedInfos = new ArrayList<>(infosToUpdate.size()); int i = 0; for (MessageInfo info : infosToUpdate) { MessageFormatInputStream stream = new TtlUpdateMessageFormatInputStream(info.getStoreKey(), info.getAccountId(), info.getContainerId(), info.getExpirationTimeInMs(), 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 : {} ttl update mark written to log", dataDir); int correspondingPutIndex = 0; for (MessageInfo info : updatedInfos) { FileSpan fileSpan = log.getFileSpanForMessage(endOffsetOfLastMessage, info.getSize()); IndexValue ttlUpdateValue = index.markAsPermanent(info.getStoreKey(), fileSpan, null, info.getOperationTimeMs(), MessageInfo.LIFE_VERSION_FROM_FRONTEND); endOffsetOfLastMessage = fileSpan.getEndOffset(); blobStoreStats.handleNewTtlUpdateEntry(ttlUpdateValue, indexValuesToUpdate.get(correspondingPutIndex++)); } logger.trace("Store : {} ttl update 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 update ttl of 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 updateTtl(List<MessageInfo> infosToUpdate) throws StoreException { checkStarted(); checkDuplicates(infosToUpdate); final Timer.Context context = metrics.ttlUpdateResponse.time(); try { List<IndexValue> indexValuesToUpdate = new ArrayList<>(); List<Short> lifeVersions = new ArrayList<>(); Offset indexEndOffsetBeforeCheck = index.getCurrentEndOffset(); for (MessageInfo info : infosToUpdate) { if (info.getExpirationTimeInMs() != Utils.Infinite_Time) { throw new StoreException("BlobStore only supports removing the expiration time", StoreErrorCodes.Update_Not_Allowed); } IndexValue value = index.findKey(info.getStoreKey(), new FileSpan(index.getStartOffset(), indexEndOffsetBeforeCheck)); if (value == null) { throw new StoreException("Cannot update TTL of " + info.getStoreKey() + " since it's not in the index", StoreErrorCodes.ID_Not_Found); } else if (!info.getStoreKey().isAccountContainerMatch(value.getAccountId(), value.getContainerId())) { if (config.storeValidateAuthorization) { throw new StoreException( "UPDATE authorization failure. Key: " + info.getStoreKey() + " AccountId in store: " + value.getAccountId() + " ContainerId in store: " + value.getContainerId(), StoreErrorCodes.Authorization_Failure); } else { logger.warn("UPDATE authorization failure. Key: {} AccountId in store: {} ContainerId in store: {}", info.getStoreKey(), value.getAccountId(), value.getContainerId()); metrics.ttlUpdateAuthorizationFailureCount.inc(); } } else if (value.isDelete()) { throw new StoreException( "Cannot update TTL of " + info.getStoreKey() + " since it is already deleted in the index.", StoreErrorCodes.ID_Deleted); } else if (value.isTtlUpdate()) { throw new StoreException("TTL of " + info.getStoreKey() + " is already updated in the index.", StoreErrorCodes.Already_Updated); } else if (!IndexValue.hasLifeVersion(info.getLifeVersion()) && value.getExpiresAtMs() != Utils.Infinite_Time && value.getExpiresAtMs() < info.getOperationTimeMs() + ttlUpdateBufferTimeMs) { throw new StoreException( "TTL of " + info.getStoreKey() + " cannot be updated because it is too close to expiry. Op time (ms): " + info.getOperationTimeMs() + ". ExpiresAtMs: " + value.getExpiresAtMs(), StoreErrorCodes.Update_Not_Allowed); } indexValuesToUpdate.add(value); lifeVersions.add(value.getLifeVersion()); } maybeCallBeforeSynchronization(); synchronized (storeWriteLock) { Offset currentIndexEndOffset = index.getCurrentEndOffset(); if (!currentIndexEndOffset.equals(indexEndOffsetBeforeCheck)) { FileSpan fileSpan = new FileSpan(indexEndOffsetBeforeCheck, currentIndexEndOffset); for (MessageInfo info : infosToUpdate) { IndexValue value = index.findKey(info.getStoreKey(), fileSpan, EnumSet.allOf(PersistentIndex.IndexEntryType.class)); if (value != null) { if (value.isDelete()) { throw new StoreException( "Cannot update TTL of " + info.getStoreKey() + " since it is already deleted in the index.", StoreErrorCodes.ID_Deleted); } else if (value.isTtlUpdate()) { throw new StoreException("TTL of " + info.getStoreKey() + " is already updated in the index.", StoreErrorCodes.Already_Updated); } } } } List<InputStream> inputStreams = new ArrayList<>(infosToUpdate.size()); List<MessageInfo> updatedInfos = new ArrayList<>(infosToUpdate.size()); int i = 0; for (MessageInfo info : infosToUpdate) { MessageFormatInputStream stream = new TtlUpdateMessageFormatInputStream(info.getStoreKey(), info.getAccountId(), info.getContainerId(), info.getExpirationTimeInMs(), 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 : {} ttl update mark written to log", dataDir); int correspondingPutIndex = 0; for (MessageInfo info : updatedInfos) { FileSpan fileSpan = log.getFileSpanForMessage(endOffsetOfLastMessage, info.getSize()); IndexValue ttlUpdateValue = index.markAsPermanent(info.getStoreKey(), fileSpan, null, info.getOperationTimeMs(), MessageInfo.LIFE_VERSION_FROM_FRONTEND); endOffsetOfLastMessage = fileSpan.getEndOffset(); blobStoreStats.handleNewTtlUpdateEntry(ttlUpdateValue, indexValuesToUpdate.get(correspondingPutIndex++)); } logger.trace("Store : {} ttl update 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 update ttl of 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 updateTtl(List<MessageInfo> infosToUpdate) throws StoreException { checkStarted(); checkDuplicates(infosToUpdate); final Timer.Context context = metrics.ttlUpdateResponse.time(); try { List<IndexValue> indexValuesToUpdate = new ArrayList<>(); List<Short> lifeVersions = new ArrayList<>(); Offset indexEndOffsetBeforeCheck = index.getCurrentEndOffset(); for (MessageInfo info : infosToUpdate) { if (info.getExpirationTimeInMs() != Utils.Infinite_Time) { throw new StoreException("BlobStore only supports removing the expiration time", StoreErrorCodes.Update_Not_Allowed); } IndexValue value = index.findKey(info.getStoreKey(), new FileSpan(index.getStartOffset(), indexEndOffsetBeforeCheck)); if (value == null) { throw new StoreException("Cannot update TTL of " + info.getStoreKey() + " since it's not in the index", StoreErrorCodes.ID_Not_Found); } else if (!info.getStoreKey().isAccountContainerMatch(value.getAccountId(), value.getContainerId())) { if (config.storeValidateAuthorization) { throw new StoreException( "UPDATE authorization failure. Key: " + info.getStoreKey() + " AccountId in store: " + value.getAccountId() + " ContainerId in store: " + value.getContainerId(), StoreErrorCodes.Authorization_Failure); } else { logger.warn("UPDATE authorization failure. Key: {} AccountId in store: {} ContainerId in store: {}", info.getStoreKey(), value.getAccountId(), value.getContainerId()); metrics.ttlUpdateAuthorizationFailureCount.inc(); } } else if (value.isDelete()) { throw new StoreException( "Cannot update TTL of " + info.getStoreKey() + " since it is already deleted in the index.", StoreErrorCodes.ID_Deleted); } else if (value.isTtlUpdate()) { throw new StoreException("TTL of " + info.getStoreKey() + " is already updated in the index.", StoreErrorCodes.Already_Updated); } else if (!IndexValue.hasLifeVersion(info.getLifeVersion()) && value.getExpiresAtMs() != Utils.Infinite_Time && value.getExpiresAtMs() < info.getOperationTimeMs() + ttlUpdateBufferTimeMs) { throw new StoreException( "TTL of " + info.getStoreKey() + " cannot be updated because it is too close to expiry. Op time (ms): " + info.getOperationTimeMs() + ". ExpiresAtMs: " + value.getExpiresAtMs(), StoreErrorCodes.Update_Not_Allowed); } indexValuesToUpdate.add(value); lifeVersions.add(value.getLifeVersion()); } maybeCallBeforeSynchronization(); synchronized (storeWriteLock) { Offset currentIndexEndOffset = index.getCurrentEndOffset(); if (!currentIndexEndOffset.equals(indexEndOffsetBeforeCheck)) { FileSpan fileSpan = new FileSpan(indexEndOffsetBeforeCheck, currentIndexEndOffset); for (MessageInfo info : infosToUpdate) { IndexValue value = index.findKey(info.getStoreKey(), fileSpan, EnumSet.allOf(PersistentIndex.IndexEntryType.class)); if (value != null) { if (value.isDelete()) { throw new StoreException( "Cannot update TTL of " + info.getStoreKey() + " since it is already deleted in the index.", StoreErrorCodes.ID_Deleted); } else if (value.isTtlUpdate()) { throw new StoreException("TTL of " + info.getStoreKey() + " is already updated in the index.", StoreErrorCodes.Already_Updated); } } } } List<InputStream> inputStreams = new ArrayList<>(infosToUpdate.size()); List<MessageInfo> updatedInfos = new ArrayList<>(infosToUpdate.size()); int i = 0; for (MessageInfo info : infosToUpdate) { MessageFormatInputStream stream = new TtlUpdateMessageFormatInputStream(info.getStoreKey(), info.getAccountId(), info.getContainerId(), info.getExpirationTimeInMs(), 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 : {} ttl update mark written to log", dataDir); int correspondingPutIndex = 0; for (MessageInfo info : updatedInfos) { FileSpan fileSpan = log.getFileSpanForMessage(endOffsetOfLastMessage, info.getSize()); IndexValue ttlUpdateValue = index.markAsPermanent(info.getStoreKey(), fileSpan, null, info.getOperationTimeMs(), MessageInfo.LIFE_VERSION_FROM_FRONTEND); endOffsetOfLastMessage = fileSpan.getEndOffset(); blobStoreStats.handleNewTtlUpdateEntry(ttlUpdateValue, indexValuesToUpdate.get(correspondingPutIndex++)); } logger.trace("Store : {} ttl update 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 update ttl of 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();  }