@Test public void unrecognizedDirsOnDiskTest() throws Exception { MockDataNodeId dataNode = clusterMap.getDataNodes().get(0); List<ReplicaId> replicas = clusterMap.getReplicaIds(dataNode); int extraDirsCount = 0; Set<String> createdMountPaths = new HashSet<>(); for (ReplicaId replicaId : replicas) { if (createdMountPaths.add(replicaId.getMountPath())) { int count = TestUtils.RANDOM.nextInt(6) + 5; createFilesAndDirsAtPath(new File(replicaId.getDiskId().getMountPath()), count - 1, count); extraDirsCount += count; } } StorageManager storageManager = createStorageManager(dataNode, metricRegistry, null); storageManager.start(); assertEquals("There should be some unexpected partitions reported", extraDirsCount, getNumUnrecognizedPartitionsReported()); checkStoreAccessibility(replicas, null, storageManager); shutdownAndAssertStoresInaccessible(storageManager, replicas); }
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);  }