@Test public void updateInstanceConfigSuccessTest() throws Exception { generateConfigs(true, true); MockDataNodeId localNode = clusterMap.getDataNodes().get(0); List<ReplicaId> localReplicas = clusterMap.getReplicaIds(localNode); MockClusterParticipant mockHelixParticipant = new MockClusterParticipant(); StorageManager storageManager = createStorageManager(localNode, metricRegistry, Collections.singletonList(mockHelixParticipant)); storageManager.start(); PartitionId newPartition = clusterMap.createNewPartition(Collections.singletonList(localNode)); ReplicaId newReplica = newPartition.getReplicaIds().get(0); String instanceName = ClusterMapUtils.getInstanceName(clusterMapConfig.clusterMapHostName, clusterMapConfig.clusterMapPort); InstanceConfig instanceConfig = new InstanceConfig(instanceName); instanceConfig.setHostName(localNode.getHostname()); instanceConfig.setPort(Integer.toString(localNode.getPort())); Map<String, Map<String, String>> diskInfos = new HashMap<>(); instanceConfig.getRecord().setMapFields(diskInfos); HelixAdmin helixAdmin = mockHelixParticipant.getHelixAdmin(); helixAdmin.addCluster(CLUSTER_NAME); helixAdmin.addInstance(CLUSTER_NAME, instanceConfig); mockHelixParticipant.onPartitionBecomeBootstrapFromOffline(newPartition.toPathString()); instanceConfig = helixAdmin.getInstanceConfig(CLUSTER_NAME, instanceName); Map<String, Map<String, String>> mountPathToDiskInfos = instanceConfig.getRecord().getMapFields(); Map<String, String> diskInfo = mountPathToDiskInfos.get(newReplica.getMountPath()); String replicasStr = diskInfo.get("Replicas"); Set<String> partitionStrs = new HashSet<>(); for (String replicaInfo : replicasStr.split(",")) { String[] infos = replicaInfo.split(":"); partitionStrs.add(infos[0]); } assertTrue("New replica info is not found in InstanceConfig", partitionStrs.contains(newPartition.toPathString())); shutdownAndAssertStoresInaccessible(storageManager, localReplicas); }
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);  }