@Test public void shutdownBlobStoreTest() throws Exception { MockDataNodeId dataNode = clusterMap.getDataNodes().get(0); List<ReplicaId> replicas = clusterMap.getReplicaIds(dataNode); List<MockDataNodeId> dataNodes = new ArrayList<>(); dataNodes.add(dataNode); MockPartitionId invalidPartition = new MockPartitionId(Long.MAX_VALUE, MockClusterMap.DEFAULT_PARTITION_CLASS, dataNodes, 0); List<? extends ReplicaId> invalidPartitionReplicas = invalidPartition.getReplicaIds(); StorageManager storageManager = createStorageManager(dataNode, metricRegistry, null); storageManager.start(); assertEquals("There should be 1 unexpected partition reported", 1, getNumUnrecognizedPartitionsReported()); for (int i = 1; i < replicas.size() - 1; i++) { ReplicaId replica = replicas.get(i); PartitionId id = replica.getPartitionId(); assertTrue("Shutdown should succeed on given store", storageManager.shutdownBlobStore(id)); } ReplicaId replica = replicas.get(replicas.size() - 1); PartitionId id = replica.getPartitionId(); Store store = storageManager.getStore(id, false); store.shutdown(); assertTrue("Shutdown should succeed on the store which is not started", storageManager.shutdownBlobStore(id)); replica = replicas.get(0); id = replica.getPartitionId(); storageManager.getDiskManager(id).shutdown(); assertFalse("Shutdown should fail on given store whose DiskManager is not running", storageManager.shutdownBlobStore(id)); replica = invalidPartitionReplicas.get(0); id = replica.getPartitionId(); assertFalse("Shutdown should fail on given invalid replica", storageManager.shutdownBlobStore(id)); shutdownAndAssertStoresInaccessible(storageManager, replicas); }
@Override public boolean shutdownBlobStore(PartitionId id) { DiskManager diskManager = partitionToDiskManager.get(id); return diskManager != null && diskManager.shutdownBlobStore(id); }
StorageManager implements StoreManager { @Override public boolean shutdownBlobStore(PartitionId id) { DiskManager diskManager = partitionToDiskManager.get(id); return diskManager != null && diskManager.shutdownBlobStore(id); } }
StorageManager implements StoreManager { @Override public boolean shutdownBlobStore(PartitionId id) { DiskManager diskManager = partitionToDiskManager.get(id); return diskManager != null && diskManager.shutdownBlobStore(id); } StorageManager(StoreConfig storeConfig, DiskManagerConfig diskManagerConfig,
      ScheduledExecutorService scheduler, MetricRegistry registry, StoreKeyFactory keyFactory, ClusterMap clusterMap,
      DataNodeId dataNodeId, MessageStoreHardDelete hardDelete, List<ClusterParticipant> clusterParticipants, Time time,
      MessageStoreRecovery recovery, AccountService accountService); }
StorageManager implements StoreManager { @Override public boolean shutdownBlobStore(PartitionId id) { DiskManager diskManager = partitionToDiskManager.get(id); return diskManager != null && diskManager.shutdownBlobStore(id); } StorageManager(StoreConfig storeConfig, DiskManagerConfig diskManagerConfig,
      ScheduledExecutorService scheduler, MetricRegistry registry, StoreKeyFactory keyFactory, ClusterMap clusterMap,
      DataNodeId dataNodeId, MessageStoreHardDelete hardDelete, List<ClusterParticipant> clusterParticipants, Time time,
      MessageStoreRecovery recovery, AccountService accountService); void start(); @Override Store getStore(PartitionId id); Store getStore(PartitionId id, boolean skipStateCheck); @Override ReplicaId getReplica(String partitionName); @Override Collection<PartitionId> getLocalPartitions(); @Override ServerErrorCode checkLocalPartitionStatus(PartitionId partition, ReplicaId localReplica); @Override boolean scheduleNextForCompaction(PartitionId id); @Override boolean controlCompactionForBlobStore(PartitionId id, boolean enabled); void shutdown(); @Override boolean addBlobStore(ReplicaId replica); @Override boolean startBlobStore(PartitionId id); @Override boolean shutdownBlobStore(PartitionId id); @Override boolean removeBlobStore(PartitionId id); @Override List<PartitionId> setBlobStoreStoppedState(List<PartitionId> partitionIds, boolean markStop); }
StorageManager implements StoreManager { @Override public boolean shutdownBlobStore(PartitionId id) { DiskManager diskManager = partitionToDiskManager.get(id); return diskManager != null && diskManager.shutdownBlobStore(id); } StorageManager(StoreConfig storeConfig, DiskManagerConfig diskManagerConfig,
      ScheduledExecutorService scheduler, MetricRegistry registry, StoreKeyFactory keyFactory, ClusterMap clusterMap,
      DataNodeId dataNodeId, MessageStoreHardDelete hardDelete, List<ClusterParticipant> clusterParticipants, Time time,
      MessageStoreRecovery recovery, AccountService accountService); void start(); @Override Store getStore(PartitionId id); Store getStore(PartitionId id, boolean skipStateCheck); @Override ReplicaId getReplica(String partitionName); @Override Collection<PartitionId> getLocalPartitions(); @Override ServerErrorCode checkLocalPartitionStatus(PartitionId partition, ReplicaId localReplica); @Override boolean scheduleNextForCompaction(PartitionId id); @Override boolean controlCompactionForBlobStore(PartitionId id, boolean enabled); void shutdown(); @Override boolean addBlobStore(ReplicaId replica); @Override boolean startBlobStore(PartitionId id); @Override boolean shutdownBlobStore(PartitionId id); @Override boolean removeBlobStore(PartitionId id); @Override List<PartitionId> setBlobStoreStoppedState(List<PartitionId> partitionIds, boolean markStop);  }