@Test public void testPartitionAssignment() throws Exception { MockClusterManager clusterManager = new MockClusterManager(3); PartitionStateChecker partitionStateChecker = new PartitionStateChecker(); WaltzServerRunner waltzServerRunner = getWaltzServerRunner(null, clusterManager); waltzServerRunner.startAsync(); WaltzServer server = waltzServerRunner.awaitStart(); assertEquals(1, clusterManager.managedServers().size()); ManagedServer managedServer = clusterManager.managedServers().iterator().next(); setPartitions(managedServer, partition1); assertEquals(Utils.set(1), server.partitions().keySet()); partitionStateChecker.check(server.partitions()); setPartitions(managedServer, partition1, partition3); assertEquals(Utils.set(1, 3), server.partitions().keySet()); partitionStateChecker.check(server.partitions()); setPartitions(managedServer, partition3); assertEquals(Utils.set(3), server.partitions().keySet()); partitionStateChecker.check(server.partitions()); setPartitions(managedServer, partition2, partition3); assertEquals(Utils.set(2, 3), server.partitions().keySet()); partitionStateChecker.check(server.partitions()); waltzServerRunner.stop(); assertEquals(0, clusterManager.managedServers().size()); }
Map<Integer, Partition> partitions() { return Collections.unmodifiableMap(partitions); }
WaltzServer { Map<Integer, Partition> partitions() { return Collections.unmodifiableMap(partitions); } }
WaltzServer { Map<Integer, Partition> partitions() { return Collections.unmodifiableMap(partitions); } WaltzServer(final int port, SslContext sslCtx, Store store, ClusterManager clusterManager,
                       WaltzServerConfig config); }
WaltzServer { Map<Integer, Partition> partitions() { return Collections.unmodifiableMap(partitions); } WaltzServer(final int port, SslContext sslCtx, Store store, ClusterManager clusterManager,
                       WaltzServerConfig config); void close(); void setJettyServer(ZooKeeperClient zkClient); Map<Integer, Boolean> getPartitionHealth(); boolean isClosed(); List<Integer> getPartitionIds(); void closeNetworkServer(); static void main(String[] args); }
WaltzServer { Map<Integer, Partition> partitions() { return Collections.unmodifiableMap(partitions); } WaltzServer(final int port, SslContext sslCtx, Store store, ClusterManager clusterManager,
                       WaltzServerConfig config); void close(); void setJettyServer(ZooKeeperClient zkClient); Map<Integer, Boolean> getPartitionHealth(); boolean isClosed(); List<Integer> getPartitionIds(); void closeNetworkServer(); static void main(String[] args); final int port; }