A representation of a subset of the nodes, topics, and partitions in the Kafka cluster.
| 21 | * A representation of a subset of the nodes, topics, and partitions in the Kafka cluster. |
| 22 | */ |
| 23 | public final class Cluster { |
| 24 | |
| 25 | private final List<Node> nodes; |
| 26 | private final Map<TopicPartition, PartitionInfo> partitionsByTopicPartition; |
| 27 | private final Map<String, List<PartitionInfo>> partitionsByTopic; |
| 28 | private final Map<Integer, List<PartitionInfo>> partitionsByNode; |
| 29 | |
| 30 | /** |
| 31 | * Create a new cluster with the given nodes and partitions |
| 32 | * @param nodes The nodes in the cluster |
| 33 | * @param partitions Information about a subset of the topic-partitions this cluster hosts |
| 34 | */ |
| 35 | public Cluster(Collection<Node> nodes, Collection<PartitionInfo> partitions) { |
| 36 | // make a randomized, unmodifiable copy of the nodes |
| 37 | List<Node> copy = new ArrayList<Node>(nodes); |
| 38 | Collections.shuffle(copy); |
| 39 | this.nodes = Collections.unmodifiableList(copy); |
| 40 | |
| 41 | // index the partitions by topic/partition for quick lookup |
| 42 | this.partitionsByTopicPartition = new HashMap<TopicPartition, PartitionInfo>(partitions.size()); |
| 43 | for (PartitionInfo p : partitions) |
| 44 | this.partitionsByTopicPartition.put(new TopicPartition(p.topic(), p.partition()), p); |
| 45 | |
| 46 | // index the partitions by topic and node respectively, and make the lists |
| 47 | // unmodifiable so we can hand them out in user-facing apis without risk |
| 48 | // of the client modifying the contents |
| 49 | HashMap<String, List<PartitionInfo>> partsForTopic = new HashMap<String, List<PartitionInfo>>(); |
| 50 | HashMap<Integer, List<PartitionInfo>> partsForNode = new HashMap<Integer, List<PartitionInfo>>(); |
| 51 | for (Node n : this.nodes) { |
| 52 | partsForNode.put(n.id(), new ArrayList<PartitionInfo>()); |
| 53 | } |
| 54 | for (PartitionInfo p : partitions) { |
| 55 | if (!partsForTopic.containsKey(p.topic())) |
| 56 | partsForTopic.put(p.topic(), new ArrayList<PartitionInfo>()); |
| 57 | List<PartitionInfo> psTopic = partsForTopic.get(p.topic()); |
| 58 | psTopic.add(p); |
| 59 | |
| 60 | if (p.leader() != null) { |
| 61 | List<PartitionInfo> psNode = Utils.notNull(partsForNode.get(p.leader().id())); |
| 62 | psNode.add(p); |
| 63 | } |
| 64 | } |
| 65 | this.partitionsByTopic = new HashMap<String, List<PartitionInfo>>(partsForTopic.size()); |
| 66 | for (Map.Entry<String, List<PartitionInfo>> entry : partsForTopic.entrySet()) |
| 67 | this.partitionsByTopic.put(entry.getKey(), Collections.unmodifiableList(entry.getValue())); |
| 68 | this.partitionsByNode = new HashMap<Integer, List<PartitionInfo>>(partsForNode.size()); |
| 69 | for (Map.Entry<Integer, List<PartitionInfo>> entry : partsForNode.entrySet()) |
| 70 | this.partitionsByNode.put(entry.getKey(), Collections.unmodifiableList(entry.getValue())); |
| 71 | |
| 72 | } |
| 73 | |
| 74 | /** |
| 75 | * Create an empty cluster instance with no nodes and no topic-partitions. |
| 76 | */ |
| 77 | public static Cluster empty() { |
| 78 | return new Cluster(new ArrayList<Node>(0), new ArrayList<PartitionInfo>(0)); |
| 79 | } |
| 80 |
nothing calls this directly
no outgoing calls
no test coverage detected