@Test public void getCluster_ownClusterId_returnsOwnCluster() throws Exception { expect(adminClient.describeCluster(anyObject())).andReturn(describeClusterResult); expect(describeClusterResult.clusterId()).andReturn(KafkaFuture.completedFuture(CLUSTER_ID)); expect(describeClusterResult.controller()).andReturn(KafkaFuture.completedFuture(NODE_1)); expect(describeClusterResult.nodes()).andReturn(KafkaFuture.completedFuture(NODES)); replay(adminClient, describeClusterResult); Cluster cluster = clusterManager.getCluster(CLUSTER_ID).get().get(); Cluster expected = Cluster.create( CLUSTER_ID, Broker.create(CLUSTER_ID, NODE_1.id(), NODE_1.host(), NODE_1.port(), NODE_1.rack()), Arrays.asList( Broker.create(CLUSTER_ID, NODE_1.id(), NODE_1.host(), NODE_1.port(), NODE_1.rack()), Broker.create(CLUSTER_ID, NODE_2.id(), NODE_2.host(), NODE_2.port(), NODE_2.rack()), Broker.create( CLUSTER_ID, NODE_3.id(), NODE_3.host(), NODE_3.port(), NODE_3.rack()))); assertEquals(expected, cluster); }
@Override public CompletableFuture<Optional<Cluster>> getCluster(String clusterId) { requireNonNull(clusterId); return listClusters() .thenApply(clusters -> findEntityByKey(clusters, Cluster::getClusterId, clusterId)); }
ClusterManagerImpl implements ClusterManager { @Override public CompletableFuture<Optional<Cluster>> getCluster(String clusterId) { requireNonNull(clusterId); return listClusters() .thenApply(clusters -> findEntityByKey(clusters, Cluster::getClusterId, clusterId)); } }
ClusterManagerImpl implements ClusterManager { @Override public CompletableFuture<Optional<Cluster>> getCluster(String clusterId) { requireNonNull(clusterId); return listClusters() .thenApply(clusters -> findEntityByKey(clusters, Cluster::getClusterId, clusterId)); } @Inject  ClusterManagerImpl(Admin adminClient); }
ClusterManagerImpl implements ClusterManager { @Override public CompletableFuture<Optional<Cluster>> getCluster(String clusterId) { requireNonNull(clusterId); return listClusters() .thenApply(clusters -> findEntityByKey(clusters, Cluster::getClusterId, clusterId)); } @Inject  ClusterManagerImpl(Admin adminClient); @Override CompletableFuture<List<Cluster>> listClusters(); @Override CompletableFuture<Optional<Cluster>> getCluster(String clusterId); @Override CompletableFuture<Cluster> getLocalCluster(); }
ClusterManagerImpl implements ClusterManager { @Override public CompletableFuture<Optional<Cluster>> getCluster(String clusterId) { requireNonNull(clusterId); return listClusters() .thenApply(clusters -> findEntityByKey(clusters, Cluster::getClusterId, clusterId)); } @Inject  ClusterManagerImpl(Admin adminClient); @Override CompletableFuture<List<Cluster>> listClusters(); @Override CompletableFuture<Optional<Cluster>> getCluster(String clusterId); @Override CompletableFuture<Cluster> getLocalCluster();  }