@Test public void searchAcls_nonExistingCluster_throwsNotFound() throws Exception { expect(clusterManager.getCluster(CLUSTER_ID)).andReturn(completedFuture(Optional.empty())); replay(clusterManager); try { aclManager.searchAcls( CLUSTER_ID, Acl.ResourceType.ANY, null, Acl.PatternType.ANY, null, null, Acl.Operation.ANY, Acl.Permission.ANY).get(); fail(); } catch (ExecutionException e) { assertEquals(NotFoundException.class, e.getCause().getClass()); } }
@Override public CompletableFuture<List<Acl>> searchAcls( String clusterId, Acl.ResourceType resourceType, @Nullable String resourceName, Acl.PatternType patternType, @Nullable String principal, @Nullable String host, Acl.Operation operation, Acl.Permission permission ) { AclBindingFilter aclBindingFilter = new AclBindingFilter( new ResourcePatternFilter( resourceType.toAdminResourceType(), resourceName, patternType.toAdminPatternType()), new AccessControlEntryFilter( principal, host, operation.toAclOperation(), permission.toAclPermissionType())); return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenApply(cluster -> adminClient.describeAcls(aclBindingFilter)) .thenCompose( describeAclsResult -> KafkaFutures.toCompletableFuture(describeAclsResult.values())) .thenApply( aclBindings -> aclBindings.stream() .map(aclBinding -> toAcl(clusterId, aclBinding)) .collect(Collectors.toList())); }
AclManagerImpl implements AclManager { @Override public CompletableFuture<List<Acl>> searchAcls( String clusterId, Acl.ResourceType resourceType, @Nullable String resourceName, Acl.PatternType patternType, @Nullable String principal, @Nullable String host, Acl.Operation operation, Acl.Permission permission ) { AclBindingFilter aclBindingFilter = new AclBindingFilter( new ResourcePatternFilter( resourceType.toAdminResourceType(), resourceName, patternType.toAdminPatternType()), new AccessControlEntryFilter( principal, host, operation.toAclOperation(), permission.toAclPermissionType())); return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenApply(cluster -> adminClient.describeAcls(aclBindingFilter)) .thenCompose( describeAclsResult -> KafkaFutures.toCompletableFuture(describeAclsResult.values())) .thenApply( aclBindings -> aclBindings.stream() .map(aclBinding -> toAcl(clusterId, aclBinding)) .collect(Collectors.toList())); } }
AclManagerImpl implements AclManager { @Override public CompletableFuture<List<Acl>> searchAcls( String clusterId, Acl.ResourceType resourceType, @Nullable String resourceName, Acl.PatternType patternType, @Nullable String principal, @Nullable String host, Acl.Operation operation, Acl.Permission permission ) { AclBindingFilter aclBindingFilter = new AclBindingFilter( new ResourcePatternFilter( resourceType.toAdminResourceType(), resourceName, patternType.toAdminPatternType()), new AccessControlEntryFilter( principal, host, operation.toAclOperation(), permission.toAclPermissionType())); return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenApply(cluster -> adminClient.describeAcls(aclBindingFilter)) .thenCompose( describeAclsResult -> KafkaFutures.toCompletableFuture(describeAclsResult.values())) .thenApply( aclBindings -> aclBindings.stream() .map(aclBinding -> toAcl(clusterId, aclBinding)) .collect(Collectors.toList())); } @Inject  AclManagerImpl(Admin adminClient, ClusterManager clusterManager); }
AclManagerImpl implements AclManager { @Override public CompletableFuture<List<Acl>> searchAcls( String clusterId, Acl.ResourceType resourceType, @Nullable String resourceName, Acl.PatternType patternType, @Nullable String principal, @Nullable String host, Acl.Operation operation, Acl.Permission permission ) { AclBindingFilter aclBindingFilter = new AclBindingFilter( new ResourcePatternFilter( resourceType.toAdminResourceType(), resourceName, patternType.toAdminPatternType()), new AccessControlEntryFilter( principal, host, operation.toAclOperation(), permission.toAclPermissionType())); return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenApply(cluster -> adminClient.describeAcls(aclBindingFilter)) .thenCompose( describeAclsResult -> KafkaFutures.toCompletableFuture(describeAclsResult.values())) .thenApply( aclBindings -> aclBindings.stream() .map(aclBinding -> toAcl(clusterId, aclBinding)) .collect(Collectors.toList())); } @Inject  AclManagerImpl(Admin adminClient, ClusterManager clusterManager); @Override CompletableFuture<List<Acl>> searchAcls(
      String clusterId,
      Acl.ResourceType resourceType,
      @Nullable String resourceName,
      Acl.PatternType patternType,
      @Nullable String principal,
      @Nullable String host,
      Acl.Operation operation,
      Acl.Permission permission
  ); @Override CompletableFuture<Void> createAcl(
      String clusterId,
      ResourceType resourceType,
      String resourceName,
      PatternType patternType,
      String principal,
      String host,
      Operation operation,
      Permission permission
  ); @Override CompletableFuture<List<Acl>> deleteAcls(
      String clusterId,
      ResourceType resourceType,
      String resourceName,
      PatternType patternType,
      String principal,
      String host,
      Operation operation,
      Permission permission
  ); }
AclManagerImpl implements AclManager { @Override public CompletableFuture<List<Acl>> searchAcls( String clusterId, Acl.ResourceType resourceType, @Nullable String resourceName, Acl.PatternType patternType, @Nullable String principal, @Nullable String host, Acl.Operation operation, Acl.Permission permission ) { AclBindingFilter aclBindingFilter = new AclBindingFilter( new ResourcePatternFilter( resourceType.toAdminResourceType(), resourceName, patternType.toAdminPatternType()), new AccessControlEntryFilter( principal, host, operation.toAclOperation(), permission.toAclPermissionType())); return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenApply(cluster -> adminClient.describeAcls(aclBindingFilter)) .thenCompose( describeAclsResult -> KafkaFutures.toCompletableFuture(describeAclsResult.values())) .thenApply( aclBindings -> aclBindings.stream() .map(aclBinding -> toAcl(clusterId, aclBinding)) .collect(Collectors.toList())); } @Inject  AclManagerImpl(Admin adminClient, ClusterManager clusterManager); @Override CompletableFuture<List<Acl>> searchAcls(
      String clusterId,
      Acl.ResourceType resourceType,
      @Nullable String resourceName,
      Acl.PatternType patternType,
      @Nullable String principal,
      @Nullable String host,
      Acl.Operation operation,
      Acl.Permission permission
  ); @Override CompletableFuture<Void> createAcl(
      String clusterId,
      ResourceType resourceType,
      String resourceName,
      PatternType patternType,
      String principal,
      String host,
      Operation operation,
      Permission permission
  ); @Override CompletableFuture<List<Acl>> deleteAcls(
      String clusterId,
      ResourceType resourceType,
      String resourceName,
      PatternType patternType,
      String principal,
      String host,
      Operation operation,
      Permission permission
  );  }