@Test public void testGetPartitionInfos() throws IOException, StorageException { File dir = Files.createTempDirectory("StorageManagerTest-").toFile(); StorageManager manager = new StorageManager(dir.getPath(), WaltzStorageConfig.DEFAULT_SEGMENT_SIZE_THRESHOLD, NUM_PARTITION, clusterKey, WaltzStorageConfig.DEFAULT_STORAGE_SEGMENT_CACHE_CAPACITY); try { manager.open(clusterKey, NUM_PARTITION); List<PartitionInfoSnapshot> partitionInfos = manager.getPartitionInfos() .stream() .sorted(Comparator.comparingInt(o -> o.partitionId)) .collect(Collectors.toList()); assertEquals(NUM_PARTITION, partitionInfos.size()); for (int i = 0; i < partitionInfos.size(); i++) { PartitionInfoSnapshot expectedPartitionInfoSnapshot = new PartitionInfoSnapshot(i, -1L, -1L, -1, PartitionInfo.Flags.PARTITION_IS_AVAILABLE); assertEquals(expectedPartitionInfoSnapshot, partitionInfos.get(i)); } } finally { manager.close(); } }
public Set<PartitionInfoSnapshot> getPartitionInfos() throws StorageException, IOException { synchronized (this) { if (running) { if (controlFile == null) { throw new StorageException("Control file is not opened, cannot get partition info."); } Set<PartitionInfoSnapshot> infos = new HashSet<>(); for (int id : controlFile.getPartitionIds()) { infos.add(controlFile.getPartitionInfo(id).getSnapshot()); } return infos; } else { throw new StorageException("Control file is already closed."); } } }
StorageManager { public Set<PartitionInfoSnapshot> getPartitionInfos() throws StorageException, IOException { synchronized (this) { if (running) { if (controlFile == null) { throw new StorageException("Control file is not opened, cannot get partition info."); } Set<PartitionInfoSnapshot> infos = new HashSet<>(); for (int id : controlFile.getPartitionIds()) { infos.add(controlFile.getPartitionInfo(id).getSnapshot()); } return infos; } else { throw new StorageException("Control file is already closed."); } } } }
StorageManager { public Set<PartitionInfoSnapshot> getPartitionInfos() throws StorageException, IOException { synchronized (this) { if (running) { if (controlFile == null) { throw new StorageException("Control file is not opened, cannot get partition info."); } Set<PartitionInfoSnapshot> infos = new HashSet<>(); for (int id : controlFile.getPartitionIds()) { infos.add(controlFile.getPartitionInfo(id).getSnapshot()); } return infos; } else { throw new StorageException("Control file is already closed."); } } } StorageManager(String directory, long segmentSizeThreshold, int numPartitions, UUID key, int segmentCacheCapacity); }
StorageManager { public Set<PartitionInfoSnapshot> getPartitionInfos() throws StorageException, IOException { synchronized (this) { if (running) { if (controlFile == null) { throw new StorageException("Control file is not opened, cannot get partition info."); } Set<PartitionInfoSnapshot> infos = new HashSet<>(); for (int id : controlFile.getPartitionIds()) { infos.add(controlFile.getPartitionInfo(id).getSnapshot()); } return infos; } else { throw new StorageException("Control file is already closed."); } } } StorageManager(String directory, long segmentSizeThreshold, int numPartitions, UUID key, int segmentCacheCapacity); void open(UUID key, int numPartitions); void setPartitionAssignment(int partitionId, boolean isAssigned, boolean deleteStorageFiles); void setPartitionAvailable(int partitionId, boolean isAvailable); void close(); int checksum(int partitionId); Set<PartitionInfoSnapshot> getPartitionInfos(); Set<Integer> getAssignedPartitionIds(); }
StorageManager { public Set<PartitionInfoSnapshot> getPartitionInfos() throws StorageException, IOException { synchronized (this) { if (running) { if (controlFile == null) { throw new StorageException("Control file is not opened, cannot get partition info."); } Set<PartitionInfoSnapshot> infos = new HashSet<>(); for (int id : controlFile.getPartitionIds()) { infos.add(controlFile.getPartitionInfo(id).getSnapshot()); } return infos; } else { throw new StorageException("Control file is already closed."); } } } StorageManager(String directory, long segmentSizeThreshold, int numPartitions, UUID key, int segmentCacheCapacity); void open(UUID key, int numPartitions); void setPartitionAssignment(int partitionId, boolean isAssigned, boolean deleteStorageFiles); void setPartitionAvailable(int partitionId, boolean isAvailable); void close(); int checksum(int partitionId); Set<PartitionInfoSnapshot> getPartitionInfos(); Set<Integer> getAssignedPartitionIds();  }