@Test public void testRetainLocalPartitions() { List<PartitionInfo> partitions = Arrays.asList( new PartitionInfo("topix", 0, new Node(0, "0", 9092, "us-east-1a"), null, null), new PartitionInfo("topix", 1, new Node(0, "0", 9092, "us-east-1c"), null, null), new PartitionInfo("topix", 3, new Node(0, "0", 9092, "us-east-1d"), null, null), new PartitionInfo("topix", 2, new Node(0, "0", 9092, "us-east-1e"), null, null)); LocalityAwareRandomPartitioner partitioner = new LocalityAwareRandomPartitioner("us-east-1e", 10000); List<PartitionInfo> retainLocalPartition = partitioner.retainLocalPartitions(partitions); assertEquals(1, retainLocalPartition.size()); assertEquals("us-east-1e", retainLocalPartition.get(0).leader().rack()); partitioner = new LocalityAwareRandomPartitioner("us-east-1c", 10000); retainLocalPartition = partitioner.retainLocalPartitions(partitions); assertEquals(1, retainLocalPartition.size()); assertEquals("us-east-1c", retainLocalPartition.get(0).leader().rack()); partitions = Arrays.asList( new PartitionInfo("topix", 0, new Node(0, "0", 9092, "us-east-1a"), null, null), new PartitionInfo("topix", 1, new Node(0, "0", 9092, "us-east-1c"), null, null), new PartitionInfo("topix", 2, new Node(0, "0", 9092, "us-east-1d"), null, null), new PartitionInfo("topix", 4, new Node(0, "0", 9092, "us-east-1d"), null, null), new PartitionInfo("topix", 3, new Node(0, "0", 9092, "us-east-1e"), null, null)); partitioner = new LocalityAwareRandomPartitioner("us-east-1d", 10000); retainLocalPartition = partitioner.retainLocalPartitions(partitions); assertEquals(2, retainLocalPartition.size()); assertEquals("us-east-1d", retainLocalPartition.get(0).leader().rack()); assertEquals(2, retainLocalPartition.get(0).partition()); assertEquals(4, retainLocalPartition.get(1).partition()); }
@Override public int partition(Object messageKey, List<PartitionInfo> partitions) { if (localPartitions == null || isTimeToRefresh()) { checkAndAssignLocalPartitions(partitions); updateNextRefreshTime(); } return localPartitions.get(Math.abs(random.nextInt() % localPartitions.size())).partition(); }
LocalityAwareRandomPartitioner extends LocalityAwarePartitioner { @Override public int partition(Object messageKey, List<PartitionInfo> partitions) { if (localPartitions == null || isTimeToRefresh()) { checkAndAssignLocalPartitions(partitions); updateNextRefreshTime(); } return localPartitions.get(Math.abs(random.nextInt() % localPartitions.size())).partition(); } }
LocalityAwareRandomPartitioner extends LocalityAwarePartitioner { @Override public int partition(Object messageKey, List<PartitionInfo> partitions) { if (localPartitions == null || isTimeToRefresh()) { checkAndAssignLocalPartitions(partitions); updateNextRefreshTime(); } return localPartitions.get(Math.abs(random.nextInt() % localPartitions.size())).partition(); } LocalityAwareRandomPartitioner(); protected  LocalityAwareRandomPartitioner(String rack, long refreshIntervalMs); }
LocalityAwareRandomPartitioner extends LocalityAwarePartitioner { @Override public int partition(Object messageKey, List<PartitionInfo> partitions) { if (localPartitions == null || isTimeToRefresh()) { checkAndAssignLocalPartitions(partitions); updateNextRefreshTime(); } return localPartitions.get(Math.abs(random.nextInt() % localPartitions.size())).partition(); } LocalityAwareRandomPartitioner(); protected  LocalityAwareRandomPartitioner(String rack, long refreshIntervalMs); @Override int partition(Object messageKey, List<PartitionInfo> partitions); }
LocalityAwareRandomPartitioner extends LocalityAwarePartitioner { @Override public int partition(Object messageKey, List<PartitionInfo> partitions) { if (localPartitions == null || isTimeToRefresh()) { checkAndAssignLocalPartitions(partitions); updateNextRefreshTime(); } return localPartitions.get(Math.abs(random.nextInt() % localPartitions.size())).partition(); } LocalityAwareRandomPartitioner(); protected  LocalityAwareRandomPartitioner(String rack, long refreshIntervalMs); @Override int partition(Object messageKey, List<PartitionInfo> partitions); static final long REFRESH_INTERVAL_MS; }