@Test public void createAcl_createsAcl() throws Exception { expect(clusterManager.getCluster(CLUSTER_ID)) .andReturn( completedFuture( Optional.of(Cluster.create(CLUSTER_ID, null, emptyList())))); expect(adminClient.createAcls(singletonList(ACL_BINDING_1))).andReturn(createAclsResult); expect(createAclsResult.values()) .andReturn(Collections.singletonMap(ACL_BINDING_1, KafkaFuture.completedFuture(null))); replay(clusterManager, adminClient, createAclsResult); aclManager.createAcl( CLUSTER_ID, Acl.ResourceType.CLUSTER, "*", Acl.PatternType.LITERAL, "User:alice", "*", Acl.Operation.READ, Acl.Permission.ALLOW).get(); verify(adminClient); }
@Override public CompletableFuture<Void> createAcl( String clusterId, ResourceType resourceType, String resourceName, PatternType patternType, String principal, String host, Operation operation, Permission permission ) { AclBinding aclBinding = new AclBinding( new ResourcePattern( resourceType.toAdminResourceType(), resourceName, patternType.toAdminPatternType()), new AccessControlEntry( principal, host, operation.toAclOperation(), permission.toAclPermissionType())); return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenApply(cluster -> adminClient.createAcls(singletonList(aclBinding))) .thenCompose( createAclsResult -> KafkaFutures.toCompletableFuture(createAclsResult.values().get(aclBinding))); }
AclManagerImpl implements AclManager { @Override public CompletableFuture<Void> createAcl( String clusterId, ResourceType resourceType, String resourceName, PatternType patternType, String principal, String host, Operation operation, Permission permission ) { AclBinding aclBinding = new AclBinding( new ResourcePattern( resourceType.toAdminResourceType(), resourceName, patternType.toAdminPatternType()), new AccessControlEntry( principal, host, operation.toAclOperation(), permission.toAclPermissionType())); return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenApply(cluster -> adminClient.createAcls(singletonList(aclBinding))) .thenCompose( createAclsResult -> KafkaFutures.toCompletableFuture(createAclsResult.values().get(aclBinding))); } }
AclManagerImpl implements AclManager { @Override public CompletableFuture<Void> createAcl( String clusterId, ResourceType resourceType, String resourceName, PatternType patternType, String principal, String host, Operation operation, Permission permission ) { AclBinding aclBinding = new AclBinding( new ResourcePattern( resourceType.toAdminResourceType(), resourceName, patternType.toAdminPatternType()), new AccessControlEntry( principal, host, operation.toAclOperation(), permission.toAclPermissionType())); return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenApply(cluster -> adminClient.createAcls(singletonList(aclBinding))) .thenCompose( createAclsResult -> KafkaFutures.toCompletableFuture(createAclsResult.values().get(aclBinding))); } @Inject  AclManagerImpl(Admin adminClient, ClusterManager clusterManager); }
AclManagerImpl implements AclManager { @Override public CompletableFuture<Void> createAcl( String clusterId, ResourceType resourceType, String resourceName, PatternType patternType, String principal, String host, Operation operation, Permission permission ) { AclBinding aclBinding = new AclBinding( new ResourcePattern( resourceType.toAdminResourceType(), resourceName, patternType.toAdminPatternType()), new AccessControlEntry( principal, host, operation.toAclOperation(), permission.toAclPermissionType())); return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenApply(cluster -> adminClient.createAcls(singletonList(aclBinding))) .thenCompose( createAclsResult -> KafkaFutures.toCompletableFuture(createAclsResult.values().get(aclBinding))); } @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<Void> createAcl( String clusterId, ResourceType resourceType, String resourceName, PatternType patternType, String principal, String host, Operation operation, Permission permission ) { AclBinding aclBinding = new AclBinding( new ResourcePattern( resourceType.toAdminResourceType(), resourceName, patternType.toAdminPatternType()), new AccessControlEntry( principal, host, operation.toAclOperation(), permission.toAclPermissionType())); return clusterManager.getCluster(clusterId) .thenApply(cluster -> checkEntityExists(cluster, "Cluster %s cannot be found.", clusterId)) .thenApply(cluster -> adminClient.createAcls(singletonList(aclBinding))) .thenCompose( createAclsResult -> KafkaFutures.toCompletableFuture(createAclsResult.values().get(aclBinding))); } @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
  );  }