@Test public void concurrentGetTest() throws Exception { int extraBlobCount = 4000 / PUT_RECORD_SIZE + 1; put(extraBlobCount, PUT_RECORD_SIZE, Utils.Infinite_Time); List<Getter> getters = new ArrayList<>(allKeys.size()); for (MockId id : allKeys.keySet()) { getters.add(new Getter(id, EnumSet.noneOf(StoreGetOptions.class))); } ExecutorService executorService = Executors.newFixedThreadPool(getters.size()); List<Future<CallableResult>> futures = executorService.invokeAll(getters); verifyGetFutures(getters, futures); }
@Override public void put(MessageWriteSet messageSetToWrite) throws StoreException { checkStarted(); if (messageSetToWrite.getMessageSetInfo().isEmpty()) { throw new IllegalArgumentException("Message write set cannot be empty"); } checkDuplicates(messageSetToWrite.getMessageSetInfo()); final Timer.Context context = metrics.putResponse.time(); try { Offset indexEndOffsetBeforeCheck = index.getCurrentEndOffset(); MessageWriteSetStateInStore state = checkWriteSetStateInStore(messageSetToWrite, new FileSpan(index.getStartOffset(), indexEndOffsetBeforeCheck)); if (state == MessageWriteSetStateInStore.ALL_ABSENT) { maybeCallBeforeSynchronization(); synchronized (storeWriteLock) { Offset currentIndexEndOffset = index.getCurrentEndOffset(); if (!currentIndexEndOffset.equals(indexEndOffsetBeforeCheck)) { FileSpan fileSpan = new FileSpan(indexEndOffsetBeforeCheck, currentIndexEndOffset); state = checkWriteSetStateInStore(messageSetToWrite, fileSpan); } if (state == MessageWriteSetStateInStore.ALL_ABSENT) { Offset endOffsetOfLastMessage = log.getEndOffset(); messageSetToWrite.writeTo(log); logger.trace("Store : {} message set written to log", dataDir); List<MessageInfo> messageInfo = messageSetToWrite.getMessageSetInfo(); ArrayList<IndexEntry> indexEntries = new ArrayList<>(messageInfo.size()); for (MessageInfo info : messageInfo) { FileSpan fileSpan = log.getFileSpanForMessage(endOffsetOfLastMessage, info.getSize()); short lifeVersion = IndexValue.hasLifeVersion(info.getLifeVersion()) ? info.getLifeVersion() : (short) 0; IndexValue value = new IndexValue(info.getSize(), fileSpan.getStartOffset(), IndexValue.FLAGS_DEFAULT_VALUE, info.getExpirationTimeInMs(), info.getOperationTimeMs(), info.getAccountId(), info.getContainerId(), lifeVersion); IndexEntry entry = new IndexEntry(info.getStoreKey(), value, info.getCrc()); indexEntries.add(entry); endOffsetOfLastMessage = fileSpan.getEndOffset(); } FileSpan fileSpan = new FileSpan(indexEntries.get(0).getValue().getOffset(), endOffsetOfLastMessage); index.addToIndex(indexEntries, fileSpan); for (IndexEntry newEntry : indexEntries) { blobStoreStats.handleNewPutEntry(newEntry.getValue()); } logger.trace("Store : {} message set written to index ", dataDir); checkCapacityAndUpdateReplicaStatusDelegate(); } } } switch (state) { case COLLIDING: throw new StoreException( "For at least one message in the write set, another blob with same key exists in store", StoreErrorCodes.Already_Exist); case SOME_NOT_ALL_DUPLICATE: throw new StoreException( "At least one message but not all in the write set is identical to an existing entry", StoreErrorCodes.Already_Exist); case ALL_DUPLICATE: logger.trace("All entries to put already exist in the store, marking operation as successful"); break; case ALL_ABSENT: logger.trace("All entries were absent, and were written to the store successfully"); break; } onSuccess(); } catch (StoreException e) { if (e.getErrorCode() == StoreErrorCodes.IOError) { onError(); } throw e; } catch (Exception e) { throw new StoreException("Unknown error while trying to put blobs to store " + dataDir, e, StoreErrorCodes.Unknown_Error); } finally { context.stop(); } }
BlobStore implements Store { @Override public void put(MessageWriteSet messageSetToWrite) throws StoreException { checkStarted(); if (messageSetToWrite.getMessageSetInfo().isEmpty()) { throw new IllegalArgumentException("Message write set cannot be empty"); } checkDuplicates(messageSetToWrite.getMessageSetInfo()); final Timer.Context context = metrics.putResponse.time(); try { Offset indexEndOffsetBeforeCheck = index.getCurrentEndOffset(); MessageWriteSetStateInStore state = checkWriteSetStateInStore(messageSetToWrite, new FileSpan(index.getStartOffset(), indexEndOffsetBeforeCheck)); if (state == MessageWriteSetStateInStore.ALL_ABSENT) { maybeCallBeforeSynchronization(); synchronized (storeWriteLock) { Offset currentIndexEndOffset = index.getCurrentEndOffset(); if (!currentIndexEndOffset.equals(indexEndOffsetBeforeCheck)) { FileSpan fileSpan = new FileSpan(indexEndOffsetBeforeCheck, currentIndexEndOffset); state = checkWriteSetStateInStore(messageSetToWrite, fileSpan); } if (state == MessageWriteSetStateInStore.ALL_ABSENT) { Offset endOffsetOfLastMessage = log.getEndOffset(); messageSetToWrite.writeTo(log); logger.trace("Store : {} message set written to log", dataDir); List<MessageInfo> messageInfo = messageSetToWrite.getMessageSetInfo(); ArrayList<IndexEntry> indexEntries = new ArrayList<>(messageInfo.size()); for (MessageInfo info : messageInfo) { FileSpan fileSpan = log.getFileSpanForMessage(endOffsetOfLastMessage, info.getSize()); short lifeVersion = IndexValue.hasLifeVersion(info.getLifeVersion()) ? info.getLifeVersion() : (short) 0; IndexValue value = new IndexValue(info.getSize(), fileSpan.getStartOffset(), IndexValue.FLAGS_DEFAULT_VALUE, info.getExpirationTimeInMs(), info.getOperationTimeMs(), info.getAccountId(), info.getContainerId(), lifeVersion); IndexEntry entry = new IndexEntry(info.getStoreKey(), value, info.getCrc()); indexEntries.add(entry); endOffsetOfLastMessage = fileSpan.getEndOffset(); } FileSpan fileSpan = new FileSpan(indexEntries.get(0).getValue().getOffset(), endOffsetOfLastMessage); index.addToIndex(indexEntries, fileSpan); for (IndexEntry newEntry : indexEntries) { blobStoreStats.handleNewPutEntry(newEntry.getValue()); } logger.trace("Store : {} message set written to index ", dataDir); checkCapacityAndUpdateReplicaStatusDelegate(); } } } switch (state) { case COLLIDING: throw new StoreException( "For at least one message in the write set, another blob with same key exists in store", StoreErrorCodes.Already_Exist); case SOME_NOT_ALL_DUPLICATE: throw new StoreException( "At least one message but not all in the write set is identical to an existing entry", StoreErrorCodes.Already_Exist); case ALL_DUPLICATE: logger.trace("All entries to put already exist in the store, marking operation as successful"); break; case ALL_ABSENT: logger.trace("All entries were absent, and were written to the store successfully"); break; } onSuccess(); } catch (StoreException e) { if (e.getErrorCode() == StoreErrorCodes.IOError) { onError(); } throw e; } catch (Exception e) { throw new StoreException("Unknown error while trying to put blobs to store " + dataDir, e, StoreErrorCodes.Unknown_Error); } finally { context.stop(); } } }
BlobStore implements Store { @Override public void put(MessageWriteSet messageSetToWrite) throws StoreException { checkStarted(); if (messageSetToWrite.getMessageSetInfo().isEmpty()) { throw new IllegalArgumentException("Message write set cannot be empty"); } checkDuplicates(messageSetToWrite.getMessageSetInfo()); final Timer.Context context = metrics.putResponse.time(); try { Offset indexEndOffsetBeforeCheck = index.getCurrentEndOffset(); MessageWriteSetStateInStore state = checkWriteSetStateInStore(messageSetToWrite, new FileSpan(index.getStartOffset(), indexEndOffsetBeforeCheck)); if (state == MessageWriteSetStateInStore.ALL_ABSENT) { maybeCallBeforeSynchronization(); synchronized (storeWriteLock) { Offset currentIndexEndOffset = index.getCurrentEndOffset(); if (!currentIndexEndOffset.equals(indexEndOffsetBeforeCheck)) { FileSpan fileSpan = new FileSpan(indexEndOffsetBeforeCheck, currentIndexEndOffset); state = checkWriteSetStateInStore(messageSetToWrite, fileSpan); } if (state == MessageWriteSetStateInStore.ALL_ABSENT) { Offset endOffsetOfLastMessage = log.getEndOffset(); messageSetToWrite.writeTo(log); logger.trace("Store : {} message set written to log", dataDir); List<MessageInfo> messageInfo = messageSetToWrite.getMessageSetInfo(); ArrayList<IndexEntry> indexEntries = new ArrayList<>(messageInfo.size()); for (MessageInfo info : messageInfo) { FileSpan fileSpan = log.getFileSpanForMessage(endOffsetOfLastMessage, info.getSize()); short lifeVersion = IndexValue.hasLifeVersion(info.getLifeVersion()) ? info.getLifeVersion() : (short) 0; IndexValue value = new IndexValue(info.getSize(), fileSpan.getStartOffset(), IndexValue.FLAGS_DEFAULT_VALUE, info.getExpirationTimeInMs(), info.getOperationTimeMs(), info.getAccountId(), info.getContainerId(), lifeVersion); IndexEntry entry = new IndexEntry(info.getStoreKey(), value, info.getCrc()); indexEntries.add(entry); endOffsetOfLastMessage = fileSpan.getEndOffset(); } FileSpan fileSpan = new FileSpan(indexEntries.get(0).getValue().getOffset(), endOffsetOfLastMessage); index.addToIndex(indexEntries, fileSpan); for (IndexEntry newEntry : indexEntries) { blobStoreStats.handleNewPutEntry(newEntry.getValue()); } logger.trace("Store : {} message set written to index ", dataDir); checkCapacityAndUpdateReplicaStatusDelegate(); } } } switch (state) { case COLLIDING: throw new StoreException( "For at least one message in the write set, another blob with same key exists in store", StoreErrorCodes.Already_Exist); case SOME_NOT_ALL_DUPLICATE: throw new StoreException( "At least one message but not all in the write set is identical to an existing entry", StoreErrorCodes.Already_Exist); case ALL_DUPLICATE: logger.trace("All entries to put already exist in the store, marking operation as successful"); break; case ALL_ABSENT: logger.trace("All entries were absent, and were written to the store successfully"); break; } onSuccess(); } catch (StoreException e) { if (e.getErrorCode() == StoreErrorCodes.IOError) { onError(); } throw e; } catch (Exception e) { throw new StoreException("Unknown error while trying to put blobs to 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 put(MessageWriteSet messageSetToWrite) throws StoreException { checkStarted(); if (messageSetToWrite.getMessageSetInfo().isEmpty()) { throw new IllegalArgumentException("Message write set cannot be empty"); } checkDuplicates(messageSetToWrite.getMessageSetInfo()); final Timer.Context context = metrics.putResponse.time(); try { Offset indexEndOffsetBeforeCheck = index.getCurrentEndOffset(); MessageWriteSetStateInStore state = checkWriteSetStateInStore(messageSetToWrite, new FileSpan(index.getStartOffset(), indexEndOffsetBeforeCheck)); if (state == MessageWriteSetStateInStore.ALL_ABSENT) { maybeCallBeforeSynchronization(); synchronized (storeWriteLock) { Offset currentIndexEndOffset = index.getCurrentEndOffset(); if (!currentIndexEndOffset.equals(indexEndOffsetBeforeCheck)) { FileSpan fileSpan = new FileSpan(indexEndOffsetBeforeCheck, currentIndexEndOffset); state = checkWriteSetStateInStore(messageSetToWrite, fileSpan); } if (state == MessageWriteSetStateInStore.ALL_ABSENT) { Offset endOffsetOfLastMessage = log.getEndOffset(); messageSetToWrite.writeTo(log); logger.trace("Store : {} message set written to log", dataDir); List<MessageInfo> messageInfo = messageSetToWrite.getMessageSetInfo(); ArrayList<IndexEntry> indexEntries = new ArrayList<>(messageInfo.size()); for (MessageInfo info : messageInfo) { FileSpan fileSpan = log.getFileSpanForMessage(endOffsetOfLastMessage, info.getSize()); short lifeVersion = IndexValue.hasLifeVersion(info.getLifeVersion()) ? info.getLifeVersion() : (short) 0; IndexValue value = new IndexValue(info.getSize(), fileSpan.getStartOffset(), IndexValue.FLAGS_DEFAULT_VALUE, info.getExpirationTimeInMs(), info.getOperationTimeMs(), info.getAccountId(), info.getContainerId(), lifeVersion); IndexEntry entry = new IndexEntry(info.getStoreKey(), value, info.getCrc()); indexEntries.add(entry); endOffsetOfLastMessage = fileSpan.getEndOffset(); } FileSpan fileSpan = new FileSpan(indexEntries.get(0).getValue().getOffset(), endOffsetOfLastMessage); index.addToIndex(indexEntries, fileSpan); for (IndexEntry newEntry : indexEntries) { blobStoreStats.handleNewPutEntry(newEntry.getValue()); } logger.trace("Store : {} message set written to index ", dataDir); checkCapacityAndUpdateReplicaStatusDelegate(); } } } switch (state) { case COLLIDING: throw new StoreException( "For at least one message in the write set, another blob with same key exists in store", StoreErrorCodes.Already_Exist); case SOME_NOT_ALL_DUPLICATE: throw new StoreException( "At least one message but not all in the write set is identical to an existing entry", StoreErrorCodes.Already_Exist); case ALL_DUPLICATE: logger.trace("All entries to put already exist in the store, marking operation as successful"); break; case ALL_ABSENT: logger.trace("All entries were absent, and were written to the store successfully"); break; } onSuccess(); } catch (StoreException e) { if (e.getErrorCode() == StoreErrorCodes.IOError) { onError(); } throw e; } catch (Exception e) { throw new StoreException("Unknown error while trying to put blobs to 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 put(MessageWriteSet messageSetToWrite) throws StoreException { checkStarted(); if (messageSetToWrite.getMessageSetInfo().isEmpty()) { throw new IllegalArgumentException("Message write set cannot be empty"); } checkDuplicates(messageSetToWrite.getMessageSetInfo()); final Timer.Context context = metrics.putResponse.time(); try { Offset indexEndOffsetBeforeCheck = index.getCurrentEndOffset(); MessageWriteSetStateInStore state = checkWriteSetStateInStore(messageSetToWrite, new FileSpan(index.getStartOffset(), indexEndOffsetBeforeCheck)); if (state == MessageWriteSetStateInStore.ALL_ABSENT) { maybeCallBeforeSynchronization(); synchronized (storeWriteLock) { Offset currentIndexEndOffset = index.getCurrentEndOffset(); if (!currentIndexEndOffset.equals(indexEndOffsetBeforeCheck)) { FileSpan fileSpan = new FileSpan(indexEndOffsetBeforeCheck, currentIndexEndOffset); state = checkWriteSetStateInStore(messageSetToWrite, fileSpan); } if (state == MessageWriteSetStateInStore.ALL_ABSENT) { Offset endOffsetOfLastMessage = log.getEndOffset(); messageSetToWrite.writeTo(log); logger.trace("Store : {} message set written to log", dataDir); List<MessageInfo> messageInfo = messageSetToWrite.getMessageSetInfo(); ArrayList<IndexEntry> indexEntries = new ArrayList<>(messageInfo.size()); for (MessageInfo info : messageInfo) { FileSpan fileSpan = log.getFileSpanForMessage(endOffsetOfLastMessage, info.getSize()); short lifeVersion = IndexValue.hasLifeVersion(info.getLifeVersion()) ? info.getLifeVersion() : (short) 0; IndexValue value = new IndexValue(info.getSize(), fileSpan.getStartOffset(), IndexValue.FLAGS_DEFAULT_VALUE, info.getExpirationTimeInMs(), info.getOperationTimeMs(), info.getAccountId(), info.getContainerId(), lifeVersion); IndexEntry entry = new IndexEntry(info.getStoreKey(), value, info.getCrc()); indexEntries.add(entry); endOffsetOfLastMessage = fileSpan.getEndOffset(); } FileSpan fileSpan = new FileSpan(indexEntries.get(0).getValue().getOffset(), endOffsetOfLastMessage); index.addToIndex(indexEntries, fileSpan); for (IndexEntry newEntry : indexEntries) { blobStoreStats.handleNewPutEntry(newEntry.getValue()); } logger.trace("Store : {} message set written to index ", dataDir); checkCapacityAndUpdateReplicaStatusDelegate(); } } } switch (state) { case COLLIDING: throw new StoreException( "For at least one message in the write set, another blob with same key exists in store", StoreErrorCodes.Already_Exist); case SOME_NOT_ALL_DUPLICATE: throw new StoreException( "At least one message but not all in the write set is identical to an existing entry", StoreErrorCodes.Already_Exist); case ALL_DUPLICATE: logger.trace("All entries to put already exist in the store, marking operation as successful"); break; case ALL_ABSENT: logger.trace("All entries were absent, and were written to the store successfully"); break; } onSuccess(); } catch (StoreException e) { if (e.getErrorCode() == StoreErrorCodes.IOError) { onError(); } throw e; } catch (Exception e) { throw new StoreException("Unknown error while trying to put blobs to 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();  }