@Test public void rackAwareOverAllocationTest() { int replicaCountPerDataCenter = 4; long replicaCapacityInBytes = 100 * 1024 * 1024 * 1024L; TestHardwareLayout testHardwareLayout = new TestHardwareLayout("Alpha", true); PartitionLayout partitionLayout = new PartitionLayout(testHardwareLayout.getHardwareLayout(), clusterMapConfig); StaticClusterManager clusterMapManager = (new StaticClusterAgentsFactory(getDummyConfig(), partitionLayout)).getClusterMap(); List<PartitionId> allocatedPartitions; allocatedPartitions = clusterMapManager.allocatePartitions(5, MockClusterMap.DEFAULT_PARTITION_CLASS, replicaCountPerDataCenter, replicaCapacityInBytes, false); assertEquals(allocatedPartitions.size(), 5); checkNumReplicasPerDatacenter(allocatedPartitions, 3); checkRackUsage(allocatedPartitions); allocatedPartitions = clusterMapManager.allocatePartitions(5, MockClusterMap.DEFAULT_PARTITION_CLASS, replicaCountPerDataCenter, replicaCapacityInBytes, true); assertEquals(allocatedPartitions.size(), 5); checkNumReplicasPerDatacenter(allocatedPartitions, 4); }
List<PartitionId> allocatePartitions(int numPartitions, String partitionClass, int replicaCountPerDatacenter, long replicaCapacityInBytes, boolean attemptNonRackAwareOnFailure) { ArrayList<PartitionId> partitions = new ArrayList<PartitionId>(numPartitions); int partitionsAllocated = 0; while (checkEnoughUnallocatedRawCapacity(replicaCountPerDatacenter, replicaCapacityInBytes) && partitionsAllocated < numPartitions) { List<Disk> disksToAllocate = new ArrayList<>(); for (Datacenter datacenter : hardwareLayout.getDatacenters()) { List<Disk> disks = allocateDisksForPartition(replicaCountPerDatacenter, replicaCapacityInBytes, datacenter, attemptNonRackAwareOnFailure); disksToAllocate.addAll(disks); } partitions.add(partitionLayout.addNewPartition(disksToAllocate, replicaCapacityInBytes, partitionClass)); partitionsAllocated++; System.out.println("Allocated " + partitionsAllocated + " new partitions so far."); } return partitions; }
StaticClusterManager implements ClusterMap { List<PartitionId> allocatePartitions(int numPartitions, String partitionClass, int replicaCountPerDatacenter, long replicaCapacityInBytes, boolean attemptNonRackAwareOnFailure) { ArrayList<PartitionId> partitions = new ArrayList<PartitionId>(numPartitions); int partitionsAllocated = 0; while (checkEnoughUnallocatedRawCapacity(replicaCountPerDatacenter, replicaCapacityInBytes) && partitionsAllocated < numPartitions) { List<Disk> disksToAllocate = new ArrayList<>(); for (Datacenter datacenter : hardwareLayout.getDatacenters()) { List<Disk> disks = allocateDisksForPartition(replicaCountPerDatacenter, replicaCapacityInBytes, datacenter, attemptNonRackAwareOnFailure); disksToAllocate.addAll(disks); } partitions.add(partitionLayout.addNewPartition(disksToAllocate, replicaCapacityInBytes, partitionClass)); partitionsAllocated++; System.out.println("Allocated " + partitionsAllocated + " new partitions so far."); } return partitions; } }
StaticClusterManager implements ClusterMap { List<PartitionId> allocatePartitions(int numPartitions, String partitionClass, int replicaCountPerDatacenter, long replicaCapacityInBytes, boolean attemptNonRackAwareOnFailure) { ArrayList<PartitionId> partitions = new ArrayList<PartitionId>(numPartitions); int partitionsAllocated = 0; while (checkEnoughUnallocatedRawCapacity(replicaCountPerDatacenter, replicaCapacityInBytes) && partitionsAllocated < numPartitions) { List<Disk> disksToAllocate = new ArrayList<>(); for (Datacenter datacenter : hardwareLayout.getDatacenters()) { List<Disk> disks = allocateDisksForPartition(replicaCountPerDatacenter, replicaCapacityInBytes, datacenter, attemptNonRackAwareOnFailure); disksToAllocate.addAll(disks); } partitions.add(partitionLayout.addNewPartition(disksToAllocate, replicaCapacityInBytes, partitionClass)); partitionsAllocated++; System.out.println("Allocated " + partitionsAllocated + " new partitions so far."); } return partitions; } StaticClusterManager(PartitionLayout partitionLayout, String localDatacenterName, MetricRegistry metricRegistry); }
StaticClusterManager implements ClusterMap { List<PartitionId> allocatePartitions(int numPartitions, String partitionClass, int replicaCountPerDatacenter, long replicaCapacityInBytes, boolean attemptNonRackAwareOnFailure) { ArrayList<PartitionId> partitions = new ArrayList<PartitionId>(numPartitions); int partitionsAllocated = 0; while (checkEnoughUnallocatedRawCapacity(replicaCountPerDatacenter, replicaCapacityInBytes) && partitionsAllocated < numPartitions) { List<Disk> disksToAllocate = new ArrayList<>(); for (Datacenter datacenter : hardwareLayout.getDatacenters()) { List<Disk> disks = allocateDisksForPartition(replicaCountPerDatacenter, replicaCapacityInBytes, datacenter, attemptNonRackAwareOnFailure); disksToAllocate.addAll(disks); } partitions.add(partitionLayout.addNewPartition(disksToAllocate, replicaCapacityInBytes, partitionClass)); partitionsAllocated++; System.out.println("Allocated " + partitionsAllocated + " new partitions so far."); } return partitions; } StaticClusterManager(PartitionLayout partitionLayout, String localDatacenterName, MetricRegistry metricRegistry); @Override List<PartitionId> getWritablePartitionIds(String partitionClass); @Override PartitionId getRandomWritablePartition(String partitionClass, List<PartitionId> partitionsToExclude); @Override List<PartitionId> getAllPartitionIds(String partitionClass); @Override PartitionId getPartitionIdFromStream(InputStream stream); @Override boolean hasDatacenter(String datacenterName); @Override byte getLocalDatacenterId(); @Override String getDatacenterName(byte id); @Override DataNodeId getDataNodeId(String hostname, int port); @Override List<ReplicaId> getReplicaIds(DataNodeId dataNodeId); @Override List<DataNodeId> getDataNodeIds(); @Override MetricRegistry getMetricRegistry(); @Override void registerClusterMapListener(ClusterMapChangeListener clusterMapChangeListener); @Override boolean equals(Object o); @Override void onReplicaEvent(ReplicaId replicaId, ReplicaEventType event); @Override JSONObject getSnapshot(); @Override ReplicaId getBootstrapReplica(String partitionIdStr, DataNodeId dataNodeId); @Override void close(); }
StaticClusterManager implements ClusterMap { List<PartitionId> allocatePartitions(int numPartitions, String partitionClass, int replicaCountPerDatacenter, long replicaCapacityInBytes, boolean attemptNonRackAwareOnFailure) { ArrayList<PartitionId> partitions = new ArrayList<PartitionId>(numPartitions); int partitionsAllocated = 0; while (checkEnoughUnallocatedRawCapacity(replicaCountPerDatacenter, replicaCapacityInBytes) && partitionsAllocated < numPartitions) { List<Disk> disksToAllocate = new ArrayList<>(); for (Datacenter datacenter : hardwareLayout.getDatacenters()) { List<Disk> disks = allocateDisksForPartition(replicaCountPerDatacenter, replicaCapacityInBytes, datacenter, attemptNonRackAwareOnFailure); disksToAllocate.addAll(disks); } partitions.add(partitionLayout.addNewPartition(disksToAllocate, replicaCapacityInBytes, partitionClass)); partitionsAllocated++; System.out.println("Allocated " + partitionsAllocated + " new partitions so far."); } return partitions; } StaticClusterManager(PartitionLayout partitionLayout, String localDatacenterName, MetricRegistry metricRegistry); @Override List<PartitionId> getWritablePartitionIds(String partitionClass); @Override PartitionId getRandomWritablePartition(String partitionClass, List<PartitionId> partitionsToExclude); @Override List<PartitionId> getAllPartitionIds(String partitionClass); @Override PartitionId getPartitionIdFromStream(InputStream stream); @Override boolean hasDatacenter(String datacenterName); @Override byte getLocalDatacenterId(); @Override String getDatacenterName(byte id); @Override DataNodeId getDataNodeId(String hostname, int port); @Override List<ReplicaId> getReplicaIds(DataNodeId dataNodeId); @Override List<DataNodeId> getDataNodeIds(); @Override MetricRegistry getMetricRegistry(); @Override void registerClusterMapListener(ClusterMapChangeListener clusterMapChangeListener); @Override boolean equals(Object o); @Override void onReplicaEvent(ReplicaId replicaId, ReplicaEventType event); @Override JSONObject getSnapshot(); @Override ReplicaId getBootstrapReplica(String partitionIdStr, DataNodeId dataNodeId); @Override void close();  }