@Test public void removeBlobStoreTest() 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); StorageManager storageManager = createStorageManager(dataNode, metricRegistry, null); storageManager.start(); for (int i = 3; i < replicas.size(); i++) { ReplicaId replica = replicas.get(i); PartitionId id = replica.getPartitionId(); assertTrue("Disable compaction should succeed", storageManager.controlCompactionForBlobStore(id, false)); assertTrue("Shutdown should succeed on given store", storageManager.shutdownBlobStore(id)); assertTrue("Removing store should succeed", storageManager.removeBlobStore(id)); assertNull("The store should not exist", storageManager.getStore(id, false)); } PartitionId id = replicas.get(2).getPartitionId(); assertTrue("Shutdown should succeed on given store", storageManager.shutdownBlobStore(id)); assertFalse("Removing store should fail because compaction is enabled on this store", storageManager.removeBlobStore(id)); id = replicas.get(1).getPartitionId(); assertFalse("Removing store should fail because store is still started", storageManager.removeBlobStore(id)); id = replicas.get(0).getPartitionId(); storageManager.getDiskManager(id).shutdown(); assertFalse("Removing store should fail because disk manager is not running", storageManager.removeBlobStore(id)); assertFalse("Removing not-found store should return false", storageManager.removeBlobStore(invalidPartition)); shutdownAndAssertStoresInaccessible(storageManager, replicas); VerifiableProperties vProps = new VerifiableProperties(new Properties()); storageManager = new StorageManager(new StoreConfig(vProps), diskManagerConfig, Utils.newScheduler(1, false), metricRegistry, new MockIdFactory(), clusterMap, dataNode, new DummyMessageStoreHardDelete(), null, SystemTime.getInstance(), new DummyMessageStoreRecovery(), new InMemAccountService(false, false)); storageManager.start(); for (ReplicaId replica : replicas) { id = replica.getPartitionId(); assertTrue("Disable compaction should succeed", storageManager.controlCompactionForBlobStore(id, false)); assertTrue("Shutdown should succeed on given store", storageManager.shutdownBlobStore(id)); assertTrue("Removing store should succeed", storageManager.removeBlobStore(id)); assertNull("The store should not exist", storageManager.getStore(id, false)); } shutdownAndAssertStoresInaccessible(storageManager, replicas); }
@Override public boolean removeBlobStore(PartitionId id) { DiskManager diskManager = partitionToDiskManager.get(id); if (diskManager == null) { logger.info("Store {} is not found in storage manager", id); return false; } if (!diskManager.removeBlobStore(id)) { logger.error("Fail to remove store {} from disk manager", id); return false; } partitionToDiskManager.remove(id); logger.info("Store {} is successfully removed from storage manager", id); return true; }
StorageManager implements StoreManager { @Override public boolean removeBlobStore(PartitionId id) { DiskManager diskManager = partitionToDiskManager.get(id); if (diskManager == null) { logger.info("Store {} is not found in storage manager", id); return false; } if (!diskManager.removeBlobStore(id)) { logger.error("Fail to remove store {} from disk manager", id); return false; } partitionToDiskManager.remove(id); logger.info("Store {} is successfully removed from storage manager", id); return true; } }
StorageManager implements StoreManager { @Override public boolean removeBlobStore(PartitionId id) { DiskManager diskManager = partitionToDiskManager.get(id); if (diskManager == null) { logger.info("Store {} is not found in storage manager", id); return false; } if (!diskManager.removeBlobStore(id)) { logger.error("Fail to remove store {} from disk manager", id); return false; } partitionToDiskManager.remove(id); logger.info("Store {} is successfully removed from storage manager", id); return true; } 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 removeBlobStore(PartitionId id) { DiskManager diskManager = partitionToDiskManager.get(id); if (diskManager == null) { logger.info("Store {} is not found in storage manager", id); return false; } if (!diskManager.removeBlobStore(id)) { logger.error("Fail to remove store {} from disk manager", id); return false; } partitionToDiskManager.remove(id); logger.info("Store {} is successfully removed from storage manager", id); return true; } 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 removeBlobStore(PartitionId id) { DiskManager diskManager = partitionToDiskManager.get(id); if (diskManager == null) { logger.info("Store {} is not found in storage manager", id); return false; } if (!diskManager.removeBlobStore(id)) { logger.error("Fail to remove store {} from disk manager", id); return false; } partitionToDiskManager.remove(id); logger.info("Store {} is successfully removed from storage manager", id); return true; } 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);  }