@Test public void getLocalCluster_noClusterId_throwsUnsupportedProtocol() throws Exception { expect(adminClient.describeCluster(anyObject())).andReturn(describeClusterResult); expect(describeClusterResult.clusterId()).andReturn(KafkaFuture.completedFuture(null)); expect(describeClusterResult.controller()).andReturn(KafkaFuture.completedFuture(NODE_1)); expect(describeClusterResult.nodes()).andReturn(KafkaFuture.completedFuture(NODES)); replay(adminClient, describeClusterResult); try { clusterManager.getLocalCluster().get(); fail(); } catch (ExecutionException e) { assertEquals(UnsupportedProtocolException.class, e.getCause().getClass()); } }
@Override public CompletableFuture<Cluster> getLocalCluster() { DescribeClusterResult describeClusterResult = adminClient.describeCluster( new DescribeClusterOptions().includeAuthorizedOperations(false)); return CompletableFuture.completedFuture(Cluster.builder()) .thenCombine( KafkaFutures.toCompletableFuture(describeClusterResult.clusterId()), (clusterBuilder, clusterId) -> { if (clusterId == null) { throw new UnsupportedProtocolException( "Metadata Response protocol version >= 2 required."); } return clusterBuilder.setClusterId(clusterId); }) .thenCombine( KafkaFutures.toCompletableFuture(describeClusterResult.controller()), (clusterBuilder, controller) -> { if (controller == null || controller.isEmpty()) { return clusterBuilder; } return clusterBuilder.setController( Broker.fromNode(clusterBuilder.build().getClusterId(), controller)); }) .thenCombine( KafkaFutures.toCompletableFuture(describeClusterResult.nodes()), (clusterBuilder, nodes) -> { if (nodes == null) { return clusterBuilder; } return clusterBuilder.addAllBrokers( nodes.stream() .filter(node -> node != null && !node.isEmpty()) .map(node -> Broker.fromNode(clusterBuilder.build().getClusterId(), node)) .collect(Collectors.toList())); }) .thenApply(Cluster.Builder::build); }
ClusterManagerImpl implements ClusterManager { @Override public CompletableFuture<Cluster> getLocalCluster() { DescribeClusterResult describeClusterResult = adminClient.describeCluster( new DescribeClusterOptions().includeAuthorizedOperations(false)); return CompletableFuture.completedFuture(Cluster.builder()) .thenCombine( KafkaFutures.toCompletableFuture(describeClusterResult.clusterId()), (clusterBuilder, clusterId) -> { if (clusterId == null) { throw new UnsupportedProtocolException( "Metadata Response protocol version >= 2 required."); } return clusterBuilder.setClusterId(clusterId); }) .thenCombine( KafkaFutures.toCompletableFuture(describeClusterResult.controller()), (clusterBuilder, controller) -> { if (controller == null || controller.isEmpty()) { return clusterBuilder; } return clusterBuilder.setController( Broker.fromNode(clusterBuilder.build().getClusterId(), controller)); }) .thenCombine( KafkaFutures.toCompletableFuture(describeClusterResult.nodes()), (clusterBuilder, nodes) -> { if (nodes == null) { return clusterBuilder; } return clusterBuilder.addAllBrokers( nodes.stream() .filter(node -> node != null && !node.isEmpty()) .map(node -> Broker.fromNode(clusterBuilder.build().getClusterId(), node)) .collect(Collectors.toList())); }) .thenApply(Cluster.Builder::build); } }
ClusterManagerImpl implements ClusterManager { @Override public CompletableFuture<Cluster> getLocalCluster() { DescribeClusterResult describeClusterResult = adminClient.describeCluster( new DescribeClusterOptions().includeAuthorizedOperations(false)); return CompletableFuture.completedFuture(Cluster.builder()) .thenCombine( KafkaFutures.toCompletableFuture(describeClusterResult.clusterId()), (clusterBuilder, clusterId) -> { if (clusterId == null) { throw new UnsupportedProtocolException( "Metadata Response protocol version >= 2 required."); } return clusterBuilder.setClusterId(clusterId); }) .thenCombine( KafkaFutures.toCompletableFuture(describeClusterResult.controller()), (clusterBuilder, controller) -> { if (controller == null || controller.isEmpty()) { return clusterBuilder; } return clusterBuilder.setController( Broker.fromNode(clusterBuilder.build().getClusterId(), controller)); }) .thenCombine( KafkaFutures.toCompletableFuture(describeClusterResult.nodes()), (clusterBuilder, nodes) -> { if (nodes == null) { return clusterBuilder; } return clusterBuilder.addAllBrokers( nodes.stream() .filter(node -> node != null && !node.isEmpty()) .map(node -> Broker.fromNode(clusterBuilder.build().getClusterId(), node)) .collect(Collectors.toList())); }) .thenApply(Cluster.Builder::build); } @Inject  ClusterManagerImpl(Admin adminClient); }
ClusterManagerImpl implements ClusterManager { @Override public CompletableFuture<Cluster> getLocalCluster() { DescribeClusterResult describeClusterResult = adminClient.describeCluster( new DescribeClusterOptions().includeAuthorizedOperations(false)); return CompletableFuture.completedFuture(Cluster.builder()) .thenCombine( KafkaFutures.toCompletableFuture(describeClusterResult.clusterId()), (clusterBuilder, clusterId) -> { if (clusterId == null) { throw new UnsupportedProtocolException( "Metadata Response protocol version >= 2 required."); } return clusterBuilder.setClusterId(clusterId); }) .thenCombine( KafkaFutures.toCompletableFuture(describeClusterResult.controller()), (clusterBuilder, controller) -> { if (controller == null || controller.isEmpty()) { return clusterBuilder; } return clusterBuilder.setController( Broker.fromNode(clusterBuilder.build().getClusterId(), controller)); }) .thenCombine( KafkaFutures.toCompletableFuture(describeClusterResult.nodes()), (clusterBuilder, nodes) -> { if (nodes == null) { return clusterBuilder; } return clusterBuilder.addAllBrokers( nodes.stream() .filter(node -> node != null && !node.isEmpty()) .map(node -> Broker.fromNode(clusterBuilder.build().getClusterId(), node)) .collect(Collectors.toList())); }) .thenApply(Cluster.Builder::build); } @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<Cluster> getLocalCluster() { DescribeClusterResult describeClusterResult = adminClient.describeCluster( new DescribeClusterOptions().includeAuthorizedOperations(false)); return CompletableFuture.completedFuture(Cluster.builder()) .thenCombine( KafkaFutures.toCompletableFuture(describeClusterResult.clusterId()), (clusterBuilder, clusterId) -> { if (clusterId == null) { throw new UnsupportedProtocolException( "Metadata Response protocol version >= 2 required."); } return clusterBuilder.setClusterId(clusterId); }) .thenCombine( KafkaFutures.toCompletableFuture(describeClusterResult.controller()), (clusterBuilder, controller) -> { if (controller == null || controller.isEmpty()) { return clusterBuilder; } return clusterBuilder.setController( Broker.fromNode(clusterBuilder.build().getClusterId(), controller)); }) .thenCombine( KafkaFutures.toCompletableFuture(describeClusterResult.nodes()), (clusterBuilder, nodes) -> { if (nodes == null) { return clusterBuilder; } return clusterBuilder.addAllBrokers( nodes.stream() .filter(node -> node != null && !node.isEmpty()) .map(node -> Broker.fromNode(clusterBuilder.build().getClusterId(), node)) .collect(Collectors.toList())); }) .thenApply(Cluster.Builder::build); } @Inject  ClusterManagerImpl(Admin adminClient); @Override CompletableFuture<List<Cluster>> listClusters(); @Override CompletableFuture<Optional<Cluster>> getCluster(String clusterId); @Override CompletableFuture<Cluster> getLocalCluster();  }