@Test public void updateInstanceConfigFailureTest() 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)); mockHelixParticipant.updateNodeInfoReturnVal = false; try { mockHelixParticipant.onPartitionBecomeBootstrapFromOffline(newPartition.toPathString()); fail("should fail because updating InstanceConfig didn't succeed during Offline-To-Bootstrap"); } catch (StateTransitionException e) { assertEquals("Error code doesn't match", StateTransitionException.TransitionErrorCode.HelixUpdateFailure, e.getErrorCode()); } try { mockHelixParticipant.onPartitionBecomeOfflineFromInactive(localReplicas.get(0).getPartitionId().toPathString()); fail("should fail because updating InstanceConfig didn't succeed during Inactive-To-Offline"); } catch (StateTransitionException e) { assertEquals("Error code doesn't match", StateTransitionException.TransitionErrorCode.HelixUpdateFailure, e.getErrorCode()); } mockHelixParticipant.updateNodeInfoReturnVal = null; newPartition = clusterMap.createNewPartition(Collections.singletonList(localNode)); try { mockHelixParticipant.onPartitionBecomeBootstrapFromOffline(newPartition.toPathString()); fail("should fail because InstanceConfig is not found during Offline-To-Bootstrap"); } catch (StateTransitionException e) { assertEquals("Error code doesn't match", StateTransitionException.TransitionErrorCode.HelixUpdateFailure, e.getErrorCode()); } try { mockHelixParticipant.onPartitionBecomeOfflineFromInactive(localReplicas.get(1).getPartitionId().toPathString()); fail("should fail because InstanceConfig is not found during Inactive-To-Offline"); } catch (StateTransitionException e) { assertEquals("Error code doesn't match", StateTransitionException.TransitionErrorCode.HelixUpdateFailure, e.getErrorCode()); } 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);  }