@Test public void addBlobStoreTest() throws Exception { generateConfigs(true, false); MockDataNodeId localNode = clusterMap.getDataNodes().get(0); List<ReplicaId> localReplicas = clusterMap.getReplicaIds(localNode); int newMountPathIndex = 3; File f = File.createTempFile("ambry", ".tmp"); File mountFile = new File(f.getParent(), "mountpathfile" + MockClusterMap.PLAIN_TEXT_PORT_START_NUMBER + newMountPathIndex); MockClusterMap.deleteFileOrDirectory(mountFile); assertTrue("Couldn't create mount path directory", mountFile.mkdir()); localNode.addMountPaths(Collections.singletonList(mountFile.getAbsolutePath())); PartitionId newPartition1 = new MockPartitionId(10L, MockClusterMap.DEFAULT_PARTITION_CLASS, clusterMap.getDataNodes(), newMountPathIndex); StorageManager storageManager = createStorageManager(localNode, metricRegistry, null); storageManager.start(); assertFalse("Add store which is already existing should fail", storageManager.addBlobStore(localReplicas.get(0))); assertTrue("Add new store should succeed", storageManager.addBlobStore(newPartition1.getReplicaIds().get(0))); assertNotNull("The store shouldn't be null because new store is successfully added", storageManager.getStore(newPartition1, false)); PartitionId newPartition2 = new MockPartitionId(11L, MockClusterMap.DEFAULT_PARTITION_CLASS, clusterMap.getDataNodes(), 0); storageManager.getDiskManager(localReplicas.get(0).getPartitionId()).shutdown(); assertFalse("Add store onto the DiskManager which is not running should fail", storageManager.addBlobStore(newPartition2.getReplicaIds().get(0))); storageManager.getDiskManager(localReplicas.get(0).getPartitionId()).start(); PartitionId newPartition3 = new MockPartitionId(12L, MockClusterMap.DEFAULT_PARTITION_CLASS, clusterMap.getDataNodes(), 0); ReplicaId replicaToAdd = newPartition3.getReplicaIds().get(0); File previousDir = new File(replicaToAdd.getReplicaPath()); File testFile = new File(previousDir, "testFile"); MockClusterMap.deleteFileOrDirectory(previousDir); assertTrue("Cannot create dir for " + replicaToAdd.getReplicaPath(), previousDir.mkdir()); assertTrue("Cannot create test file within previous dir", testFile.createNewFile()); assertTrue("Adding new store should succeed", storageManager.addBlobStore(replicaToAdd)); assertFalse("Test file should not exist", testFile.exists()); assertNotNull("Store associated new added replica should not be null", storageManager.getStore(newPartition3, false)); shutdownAndAssertStoresInaccessible(storageManager, localReplicas); List<String> mountPaths = localNode.getMountPaths(); String diskToFail = mountPaths.get(0); File reservePoolDir = new File(diskToFail, diskManagerConfig.diskManagerReserveFileDirName); File storeReserveDir = new File(reservePoolDir, DiskSpaceAllocator.STORE_DIR_PREFIX + newPartition2.toString()); StorageManager storageManager2 = createStorageManager(localNode, new MetricRegistry(), null); storageManager2.start(); Utils.deleteFileOrDirectory(storeReserveDir); assertTrue("File creation should succeed", storeReserveDir.createNewFile()); assertFalse("Add store should fail if store couldn't start due to initializePool failure", storageManager2.addBlobStore(newPartition2.getReplicaIds().get(0))); assertNull("New store shouldn't be in in-memory data structure", storageManager2.getStore(newPartition2, false)); shutdownAndAssertStoresInaccessible(storageManager2, localReplicas); }
@Override public boolean addBlobStore(ReplicaId replica) { if (partitionToDiskManager.containsKey(replica.getPartitionId())) { logger.info("{} already exists in storage manager, rejecting adding store request", replica.getPartitionId()); return false; } DiskManager diskManager = diskToDiskManager.computeIfAbsent(replica.getDiskId(), disk -> { DiskManager newDiskManager = new DiskManager(disk, Collections.emptyList(), storeConfig, diskManagerConfig, scheduler, metrics, storeMainMetrics, storeUnderCompactionMetrics, keyFactory, recovery, hardDelete, replicaStatusDelegates, stoppedReplicas, time, accountService); logger.info("Creating new DiskManager on {} for new added store", replica.getDiskId().getMountPath()); try { newDiskManager.start(); } catch (Exception e) { logger.error("Error while starting the new DiskManager for {}", disk.getMountPath(), e); return null; } return newDiskManager; }); if (diskManager == null || !diskManager.addBlobStore(replica)) { logger.error("Failed to add new store into DiskManager"); return false; } partitionToDiskManager.put(replica.getPartitionId(), diskManager); partitionNameToReplicaId.put(replica.getPartitionId().toPathString(), replica); logger.info("New store is successfully added into StorageManager"); return true; }
StorageManager implements StoreManager { @Override public boolean addBlobStore(ReplicaId replica) { if (partitionToDiskManager.containsKey(replica.getPartitionId())) { logger.info("{} already exists in storage manager, rejecting adding store request", replica.getPartitionId()); return false; } DiskManager diskManager = diskToDiskManager.computeIfAbsent(replica.getDiskId(), disk -> { DiskManager newDiskManager = new DiskManager(disk, Collections.emptyList(), storeConfig, diskManagerConfig, scheduler, metrics, storeMainMetrics, storeUnderCompactionMetrics, keyFactory, recovery, hardDelete, replicaStatusDelegates, stoppedReplicas, time, accountService); logger.info("Creating new DiskManager on {} for new added store", replica.getDiskId().getMountPath()); try { newDiskManager.start(); } catch (Exception e) { logger.error("Error while starting the new DiskManager for {}", disk.getMountPath(), e); return null; } return newDiskManager; }); if (diskManager == null || !diskManager.addBlobStore(replica)) { logger.error("Failed to add new store into DiskManager"); return false; } partitionToDiskManager.put(replica.getPartitionId(), diskManager); partitionNameToReplicaId.put(replica.getPartitionId().toPathString(), replica); logger.info("New store is successfully added into StorageManager"); return true; } }
StorageManager implements StoreManager { @Override public boolean addBlobStore(ReplicaId replica) { if (partitionToDiskManager.containsKey(replica.getPartitionId())) { logger.info("{} already exists in storage manager, rejecting adding store request", replica.getPartitionId()); return false; } DiskManager diskManager = diskToDiskManager.computeIfAbsent(replica.getDiskId(), disk -> { DiskManager newDiskManager = new DiskManager(disk, Collections.emptyList(), storeConfig, diskManagerConfig, scheduler, metrics, storeMainMetrics, storeUnderCompactionMetrics, keyFactory, recovery, hardDelete, replicaStatusDelegates, stoppedReplicas, time, accountService); logger.info("Creating new DiskManager on {} for new added store", replica.getDiskId().getMountPath()); try { newDiskManager.start(); } catch (Exception e) { logger.error("Error while starting the new DiskManager for {}", disk.getMountPath(), e); return null; } return newDiskManager; }); if (diskManager == null || !diskManager.addBlobStore(replica)) { logger.error("Failed to add new store into DiskManager"); return false; } partitionToDiskManager.put(replica.getPartitionId(), diskManager); partitionNameToReplicaId.put(replica.getPartitionId().toPathString(), replica); logger.info("New store is successfully added into StorageManager"); 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 addBlobStore(ReplicaId replica) { if (partitionToDiskManager.containsKey(replica.getPartitionId())) { logger.info("{} already exists in storage manager, rejecting adding store request", replica.getPartitionId()); return false; } DiskManager diskManager = diskToDiskManager.computeIfAbsent(replica.getDiskId(), disk -> { DiskManager newDiskManager = new DiskManager(disk, Collections.emptyList(), storeConfig, diskManagerConfig, scheduler, metrics, storeMainMetrics, storeUnderCompactionMetrics, keyFactory, recovery, hardDelete, replicaStatusDelegates, stoppedReplicas, time, accountService); logger.info("Creating new DiskManager on {} for new added store", replica.getDiskId().getMountPath()); try { newDiskManager.start(); } catch (Exception e) { logger.error("Error while starting the new DiskManager for {}", disk.getMountPath(), e); return null; } return newDiskManager; }); if (diskManager == null || !diskManager.addBlobStore(replica)) { logger.error("Failed to add new store into DiskManager"); return false; } partitionToDiskManager.put(replica.getPartitionId(), diskManager); partitionNameToReplicaId.put(replica.getPartitionId().toPathString(), replica); logger.info("New store is successfully added into StorageManager"); 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 addBlobStore(ReplicaId replica) { if (partitionToDiskManager.containsKey(replica.getPartitionId())) { logger.info("{} already exists in storage manager, rejecting adding store request", replica.getPartitionId()); return false; } DiskManager diskManager = diskToDiskManager.computeIfAbsent(replica.getDiskId(), disk -> { DiskManager newDiskManager = new DiskManager(disk, Collections.emptyList(), storeConfig, diskManagerConfig, scheduler, metrics, storeMainMetrics, storeUnderCompactionMetrics, keyFactory, recovery, hardDelete, replicaStatusDelegates, stoppedReplicas, time, accountService); logger.info("Creating new DiskManager on {} for new added store", replica.getDiskId().getMountPath()); try { newDiskManager.start(); } catch (Exception e) { logger.error("Error while starting the new DiskManager for {}", disk.getMountPath(), e); return null; } return newDiskManager; }); if (diskManager == null || !diskManager.addBlobStore(replica)) { logger.error("Failed to add new store into DiskManager"); return false; } partitionToDiskManager.put(replica.getPartitionId(), diskManager); partitionNameToReplicaId.put(replica.getPartitionId().toPathString(), replica); logger.info("New store is successfully added into StorageManager"); 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);  }