@Test public void diskSpaceAllocatorTest() throws Exception { generateConfigs(true, false); MockDataNodeId dataNode = clusterMap.getDataNodes().get(0); List<ReplicaId> replicas = clusterMap.getReplicaIds(dataNode); List<String> mountPaths = dataNode.getMountPaths(); Map<String, List<ReplicaId>> replicasByMountPath = new HashMap<>(); for (ReplicaId replica : replicas) { replicasByMountPath.computeIfAbsent(replica.getMountPath(), key -> new ArrayList<>()).add(replica); } for (int i = 0; i < 2; i++) { metricRegistry = new MetricRegistry(); StorageManager storageManager = createStorageManager(dataNode, metricRegistry, null); storageManager.start(); assertEquals("There should be no unexpected partitions reported", 0, getNumUnrecognizedPartitionsReported()); checkStoreAccessibility(replicas, null, storageManager); Map<String, Counter> counters = metricRegistry.getCounters(); assertEquals(0, getCounterValue(counters, DiskSpaceAllocator.class.getName(), "DiskSpaceAllocatorInitFailureCount")); assertEquals(0, getCounterValue(counters, DiskManager.class.getName(), "TotalStoreStartFailures")); assertEquals(0, getCounterValue(counters, DiskManager.class.getName(), "DiskMountPathFailures")); for (String mountPath : dataNode.getMountPaths()) { List<ReplicaId> replicasOnDisk = replicasByMountPath.get(mountPath); DiskSpaceAllocatorTest.ExpectedState expectedState = new DiskSpaceAllocatorTest.ExpectedState(); expectedState.addSwapSeg(storeConfig.storeSegmentSizeInBytes, 1); for (ReplicaId replica : replicasOnDisk) { expectedState.addStoreSeg(replica.getPartitionId().toPathString(), storeConfig.storeSegmentSizeInBytes, 1); } DiskSpaceAllocatorTest.verifyPoolState(new File(mountPath, diskManagerConfig.diskManagerReserveFileDirName), expectedState); } shutdownAndAssertStoresInaccessible(storageManager, replicas); assertEquals(0, getCounterValue(counters, DiskManager.class.getName(), "TotalStoreShutdownFailures")); } metricRegistry = new MetricRegistry(); String diskToFail = mountPaths.get(RANDOM.nextInt(mountPaths.size())); File reservePoolDir = new File(diskToFail, diskManagerConfig.diskManagerReserveFileDirName); File storeReserveDir = new File(reservePoolDir, DiskSpaceAllocator.STORE_DIR_PREFIX + replicasByMountPath.get(diskToFail) .get(0) .getPartitionId() .toPathString()); File fileSizeDir = new File(storeReserveDir, DiskSpaceAllocator.generateFileSizeDirName(storeConfig.storeSegmentSizeInBytes)); Utils.deleteFileOrDirectory(fileSizeDir); StorageManager storageManager = createStorageManager(dataNode, metricRegistry, null); assertTrue("File creation should have succeeded", fileSizeDir.createNewFile()); storageManager.start(); assertEquals("There should be no unexpected partitions reported", 0, getNumUnrecognizedPartitionsReported()); checkStoreAccessibility(replicas, diskToFail, storageManager); Map<String, Counter> counters = metricRegistry.getCounters(); shutdownAndAssertStoresInaccessible(storageManager, replicas); assertEquals(0, getCounterValue(counters, DiskManager.class.getName(), "TotalStoreShutdownFailures")); }
public void start() throws InterruptedException { long startTimeMs = time.milliseconds(); try { logger.info("Starting storage manager"); List<Thread> startupThreads = new ArrayList<>(); for (final DiskManager diskManager : diskToDiskManager.values()) { Thread thread = Utils.newThread("disk-manager-startup-" + diskManager.getDisk(), () -> { try { diskManager.start(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); logger.error("Disk manager startup thread interrupted for disk {}", diskManager.getDisk(), e); } }, false); thread.start(); startupThreads.add(thread); } for (Thread startupThread : startupThreads) { startupThread.join(); } metrics.initializeCompactionThreadsTracker(this, diskToDiskManager.size()); if (clusterParticipants != null) { clusterParticipants.forEach(participant -> { participant.registerPartitionStateChangeListener(StateModelListenerType.StorageManagerListener, new PartitionStateChangeListenerImpl()); participant.setInitialLocalPartitions(partitionNameToReplicaId.keySet()); }); } diskToDiskManager.values().forEach(diskManager -> unexpectedDirs.addAll(diskManager.getUnexpectedDirs())); logger.info("Starting storage manager complete"); } finally { metrics.storageManagerStartTimeMs.update(time.milliseconds() - startTimeMs); } }
StorageManager implements StoreManager { public void start() throws InterruptedException { long startTimeMs = time.milliseconds(); try { logger.info("Starting storage manager"); List<Thread> startupThreads = new ArrayList<>(); for (final DiskManager diskManager : diskToDiskManager.values()) { Thread thread = Utils.newThread("disk-manager-startup-" + diskManager.getDisk(), () -> { try { diskManager.start(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); logger.error("Disk manager startup thread interrupted for disk {}", diskManager.getDisk(), e); } }, false); thread.start(); startupThreads.add(thread); } for (Thread startupThread : startupThreads) { startupThread.join(); } metrics.initializeCompactionThreadsTracker(this, diskToDiskManager.size()); if (clusterParticipants != null) { clusterParticipants.forEach(participant -> { participant.registerPartitionStateChangeListener(StateModelListenerType.StorageManagerListener, new PartitionStateChangeListenerImpl()); participant.setInitialLocalPartitions(partitionNameToReplicaId.keySet()); }); } diskToDiskManager.values().forEach(diskManager -> unexpectedDirs.addAll(diskManager.getUnexpectedDirs())); logger.info("Starting storage manager complete"); } finally { metrics.storageManagerStartTimeMs.update(time.milliseconds() - startTimeMs); } } }
StorageManager implements StoreManager { public void start() throws InterruptedException { long startTimeMs = time.milliseconds(); try { logger.info("Starting storage manager"); List<Thread> startupThreads = new ArrayList<>(); for (final DiskManager diskManager : diskToDiskManager.values()) { Thread thread = Utils.newThread("disk-manager-startup-" + diskManager.getDisk(), () -> { try { diskManager.start(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); logger.error("Disk manager startup thread interrupted for disk {}", diskManager.getDisk(), e); } }, false); thread.start(); startupThreads.add(thread); } for (Thread startupThread : startupThreads) { startupThread.join(); } metrics.initializeCompactionThreadsTracker(this, diskToDiskManager.size()); if (clusterParticipants != null) { clusterParticipants.forEach(participant -> { participant.registerPartitionStateChangeListener(StateModelListenerType.StorageManagerListener, new PartitionStateChangeListenerImpl()); participant.setInitialLocalPartitions(partitionNameToReplicaId.keySet()); }); } diskToDiskManager.values().forEach(diskManager -> unexpectedDirs.addAll(diskManager.getUnexpectedDirs())); logger.info("Starting storage manager complete"); } finally { metrics.storageManagerStartTimeMs.update(time.milliseconds() - startTimeMs); } } 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 { public void start() throws InterruptedException { long startTimeMs = time.milliseconds(); try { logger.info("Starting storage manager"); List<Thread> startupThreads = new ArrayList<>(); for (final DiskManager diskManager : diskToDiskManager.values()) { Thread thread = Utils.newThread("disk-manager-startup-" + diskManager.getDisk(), () -> { try { diskManager.start(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); logger.error("Disk manager startup thread interrupted for disk {}", diskManager.getDisk(), e); } }, false); thread.start(); startupThreads.add(thread); } for (Thread startupThread : startupThreads) { startupThread.join(); } metrics.initializeCompactionThreadsTracker(this, diskToDiskManager.size()); if (clusterParticipants != null) { clusterParticipants.forEach(participant -> { participant.registerPartitionStateChangeListener(StateModelListenerType.StorageManagerListener, new PartitionStateChangeListenerImpl()); participant.setInitialLocalPartitions(partitionNameToReplicaId.keySet()); }); } diskToDiskManager.values().forEach(diskManager -> unexpectedDirs.addAll(diskManager.getUnexpectedDirs())); logger.info("Starting storage manager complete"); } finally { metrics.storageManagerStartTimeMs.update(time.milliseconds() - startTimeMs); } } 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 { public void start() throws InterruptedException { long startTimeMs = time.milliseconds(); try { logger.info("Starting storage manager"); List<Thread> startupThreads = new ArrayList<>(); for (final DiskManager diskManager : diskToDiskManager.values()) { Thread thread = Utils.newThread("disk-manager-startup-" + diskManager.getDisk(), () -> { try { diskManager.start(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); logger.error("Disk manager startup thread interrupted for disk {}", diskManager.getDisk(), e); } }, false); thread.start(); startupThreads.add(thread); } for (Thread startupThread : startupThreads) { startupThread.join(); } metrics.initializeCompactionThreadsTracker(this, diskToDiskManager.size()); if (clusterParticipants != null) { clusterParticipants.forEach(participant -> { participant.registerPartitionStateChangeListener(StateModelListenerType.StorageManagerListener, new PartitionStateChangeListenerImpl()); participant.setInitialLocalPartitions(partitionNameToReplicaId.keySet()); }); } diskToDiskManager.values().forEach(diskManager -> unexpectedDirs.addAll(diskManager.getUnexpectedDirs())); logger.info("Starting storage manager complete"); } finally { metrics.storageManagerStartTimeMs.update(time.milliseconds() - startTimeMs); } } 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);  }