@Test public void testCreateAcls() throws Exception { try (MockKafkaAdminClientEnv env = mockClientEnv()) { env.kafkaClient().setNodeApiVersions(NodeApiVersions.create()); env.kafkaClient().prepareMetadataUpdate(env.cluster(), Collections.<String>emptySet()); env.kafkaClient().setNode(env.cluster().controller()); env.kafkaClient().prepareResponse(new CreateAclsResponse(0, asList(new AclCreationResponse(ApiError.NONE), new AclCreationResponse(ApiError.NONE)))); CreateAclsResult results = env.adminClient().createAcls(asList(ACL1, ACL2)); assertCollectionIs(results.values().keySet(), ACL1, ACL2); for (KafkaFuture<Void> future : results.values().values()) future.get(); results.all().get(); env.kafkaClient().prepareResponse(new CreateAclsResponse(0, asList( new AclCreationResponse(new ApiError(Errors.SECURITY_DISABLED, "Security is disabled")), new AclCreationResponse(ApiError.NONE)) )); results = env.adminClient().createAcls(asList(ACL1, ACL2)); assertCollectionIs(results.values().keySet(), ACL1, ACL2); assertFutureError(results.values().get(ACL1), SecurityDisabledException.class); results.values().get(ACL2).get(); assertFutureError(results.all(), SecurityDisabledException.class); } }
@Override public CreateAclsResult createAcls(Collection<AclBinding> acls, CreateAclsOptions options) { final long now = time.milliseconds(); final Map<AclBinding, KafkaFutureImpl<Void>> futures = new HashMap<>(); final List<AclCreation> aclCreations = new ArrayList<>(); for (AclBinding acl : acls) { if (futures.get(acl) == null) { KafkaFutureImpl<Void> future = new KafkaFutureImpl<>(); futures.put(acl, future); String indefinite = acl.toFilter().findIndefiniteField(); if (indefinite == null) { aclCreations.add(new AclCreation(acl)); } else { future.completeExceptionally(new InvalidRequestException("Invalid ACL creation: " + indefinite)); } } } runnable.call(new Call("createAcls", calcDeadlineMs(now, options.timeoutMs()), new LeastLoadedNodeProvider()) { @Override AbstractRequest.Builder createRequest(int timeoutMs) { return new CreateAclsRequest.Builder(aclCreations); } @Override void handleResponse(AbstractResponse abstractResponse) { CreateAclsResponse response = (CreateAclsResponse) abstractResponse; List<AclCreationResponse> responses = response.aclCreationResponses(); Iterator<AclCreationResponse> iter = responses.iterator(); for (AclCreation aclCreation : aclCreations) { KafkaFutureImpl<Void> future = futures.get(aclCreation.acl()); if (!iter.hasNext()) { future.completeExceptionally(new UnknownServerException( "The broker reported no creation result for the given ACL.")); } else { AclCreationResponse creation = iter.next(); if (creation.error().isFailure()) { future.completeExceptionally(creation.error().exception()); } else { future.complete(null); } } } } @Override void handleFailure(Throwable throwable) { completeAllExceptionally(futures.values(), throwable); } }, now); return new CreateAclsResult(new HashMap<AclBinding, KafkaFuture<Void>>(futures)); }
KafkaAdminClient extends AdminClient { @Override public CreateAclsResult createAcls(Collection<AclBinding> acls, CreateAclsOptions options) { final long now = time.milliseconds(); final Map<AclBinding, KafkaFutureImpl<Void>> futures = new HashMap<>(); final List<AclCreation> aclCreations = new ArrayList<>(); for (AclBinding acl : acls) { if (futures.get(acl) == null) { KafkaFutureImpl<Void> future = new KafkaFutureImpl<>(); futures.put(acl, future); String indefinite = acl.toFilter().findIndefiniteField(); if (indefinite == null) { aclCreations.add(new AclCreation(acl)); } else { future.completeExceptionally(new InvalidRequestException("Invalid ACL creation: " + indefinite)); } } } runnable.call(new Call("createAcls", calcDeadlineMs(now, options.timeoutMs()), new LeastLoadedNodeProvider()) { @Override AbstractRequest.Builder createRequest(int timeoutMs) { return new CreateAclsRequest.Builder(aclCreations); } @Override void handleResponse(AbstractResponse abstractResponse) { CreateAclsResponse response = (CreateAclsResponse) abstractResponse; List<AclCreationResponse> responses = response.aclCreationResponses(); Iterator<AclCreationResponse> iter = responses.iterator(); for (AclCreation aclCreation : aclCreations) { KafkaFutureImpl<Void> future = futures.get(aclCreation.acl()); if (!iter.hasNext()) { future.completeExceptionally(new UnknownServerException( "The broker reported no creation result for the given ACL.")); } else { AclCreationResponse creation = iter.next(); if (creation.error().isFailure()) { future.completeExceptionally(creation.error().exception()); } else { future.complete(null); } } } } @Override void handleFailure(Throwable throwable) { completeAllExceptionally(futures.values(), throwable); } }, now); return new CreateAclsResult(new HashMap<AclBinding, KafkaFuture<Void>>(futures)); } }
KafkaAdminClient extends AdminClient { @Override public CreateAclsResult createAcls(Collection<AclBinding> acls, CreateAclsOptions options) { final long now = time.milliseconds(); final Map<AclBinding, KafkaFutureImpl<Void>> futures = new HashMap<>(); final List<AclCreation> aclCreations = new ArrayList<>(); for (AclBinding acl : acls) { if (futures.get(acl) == null) { KafkaFutureImpl<Void> future = new KafkaFutureImpl<>(); futures.put(acl, future); String indefinite = acl.toFilter().findIndefiniteField(); if (indefinite == null) { aclCreations.add(new AclCreation(acl)); } else { future.completeExceptionally(new InvalidRequestException("Invalid ACL creation: " + indefinite)); } } } runnable.call(new Call("createAcls", calcDeadlineMs(now, options.timeoutMs()), new LeastLoadedNodeProvider()) { @Override AbstractRequest.Builder createRequest(int timeoutMs) { return new CreateAclsRequest.Builder(aclCreations); } @Override void handleResponse(AbstractResponse abstractResponse) { CreateAclsResponse response = (CreateAclsResponse) abstractResponse; List<AclCreationResponse> responses = response.aclCreationResponses(); Iterator<AclCreationResponse> iter = responses.iterator(); for (AclCreation aclCreation : aclCreations) { KafkaFutureImpl<Void> future = futures.get(aclCreation.acl()); if (!iter.hasNext()) { future.completeExceptionally(new UnknownServerException( "The broker reported no creation result for the given ACL.")); } else { AclCreationResponse creation = iter.next(); if (creation.error().isFailure()) { future.completeExceptionally(creation.error().exception()); } else { future.complete(null); } } } } @Override void handleFailure(Throwable throwable) { completeAllExceptionally(futures.values(), throwable); } }, now); return new CreateAclsResult(new HashMap<AclBinding, KafkaFuture<Void>>(futures)); } private  KafkaAdminClient(AdminClientConfig config, String clientId, Time time, Metadata metadata,
                     Metrics metrics, KafkaClient client, TimeoutProcessorFactory timeoutProcessorFactory); }
KafkaAdminClient extends AdminClient { @Override public CreateAclsResult createAcls(Collection<AclBinding> acls, CreateAclsOptions options) { final long now = time.milliseconds(); final Map<AclBinding, KafkaFutureImpl<Void>> futures = new HashMap<>(); final List<AclCreation> aclCreations = new ArrayList<>(); for (AclBinding acl : acls) { if (futures.get(acl) == null) { KafkaFutureImpl<Void> future = new KafkaFutureImpl<>(); futures.put(acl, future); String indefinite = acl.toFilter().findIndefiniteField(); if (indefinite == null) { aclCreations.add(new AclCreation(acl)); } else { future.completeExceptionally(new InvalidRequestException("Invalid ACL creation: " + indefinite)); } } } runnable.call(new Call("createAcls", calcDeadlineMs(now, options.timeoutMs()), new LeastLoadedNodeProvider()) { @Override AbstractRequest.Builder createRequest(int timeoutMs) { return new CreateAclsRequest.Builder(aclCreations); } @Override void handleResponse(AbstractResponse abstractResponse) { CreateAclsResponse response = (CreateAclsResponse) abstractResponse; List<AclCreationResponse> responses = response.aclCreationResponses(); Iterator<AclCreationResponse> iter = responses.iterator(); for (AclCreation aclCreation : aclCreations) { KafkaFutureImpl<Void> future = futures.get(aclCreation.acl()); if (!iter.hasNext()) { future.completeExceptionally(new UnknownServerException( "The broker reported no creation result for the given ACL.")); } else { AclCreationResponse creation = iter.next(); if (creation.error().isFailure()) { future.completeExceptionally(creation.error().exception()); } else { future.complete(null); } } } } @Override void handleFailure(Throwable throwable) { completeAllExceptionally(futures.values(), throwable); } }, now); return new CreateAclsResult(new HashMap<AclBinding, KafkaFuture<Void>>(futures)); } private  KafkaAdminClient(AdminClientConfig config, String clientId, Time time, Metadata metadata,
                     Metrics metrics, KafkaClient client, TimeoutProcessorFactory timeoutProcessorFactory); @Override void close(long duration, TimeUnit unit); @Override CreateTopicsResult createTopics(final Collection<NewTopic> newTopics,
                                           final CreateTopicsOptions options); @Override DeleteTopicsResult deleteTopics(final Collection<String> topicNames,
                                           DeleteTopicsOptions options); @Override ListTopicsResult listTopics(final ListTopicsOptions options); @Override DescribeTopicsResult describeTopics(final Collection<String> topicNames, DescribeTopicsOptions options); @Override DescribeClusterResult describeCluster(DescribeClusterOptions options); @Override DescribeAclsResult describeAcls(final AclBindingFilter filter, DescribeAclsOptions options); @Override CreateAclsResult createAcls(Collection<AclBinding> acls, CreateAclsOptions options); @Override DeleteAclsResult deleteAcls(Collection<AclBindingFilter> filters, DeleteAclsOptions options); @Override DescribeConfigsResult describeConfigs(Collection<ConfigResource> configResources, final DescribeConfigsOptions options); @Override AlterConfigsResult alterConfigs(Map<ConfigResource, Config> configs, final AlterConfigsOptions options); }
KafkaAdminClient extends AdminClient { @Override public CreateAclsResult createAcls(Collection<AclBinding> acls, CreateAclsOptions options) { final long now = time.milliseconds(); final Map<AclBinding, KafkaFutureImpl<Void>> futures = new HashMap<>(); final List<AclCreation> aclCreations = new ArrayList<>(); for (AclBinding acl : acls) { if (futures.get(acl) == null) { KafkaFutureImpl<Void> future = new KafkaFutureImpl<>(); futures.put(acl, future); String indefinite = acl.toFilter().findIndefiniteField(); if (indefinite == null) { aclCreations.add(new AclCreation(acl)); } else { future.completeExceptionally(new InvalidRequestException("Invalid ACL creation: " + indefinite)); } } } runnable.call(new Call("createAcls", calcDeadlineMs(now, options.timeoutMs()), new LeastLoadedNodeProvider()) { @Override AbstractRequest.Builder createRequest(int timeoutMs) { return new CreateAclsRequest.Builder(aclCreations); } @Override void handleResponse(AbstractResponse abstractResponse) { CreateAclsResponse response = (CreateAclsResponse) abstractResponse; List<AclCreationResponse> responses = response.aclCreationResponses(); Iterator<AclCreationResponse> iter = responses.iterator(); for (AclCreation aclCreation : aclCreations) { KafkaFutureImpl<Void> future = futures.get(aclCreation.acl()); if (!iter.hasNext()) { future.completeExceptionally(new UnknownServerException( "The broker reported no creation result for the given ACL.")); } else { AclCreationResponse creation = iter.next(); if (creation.error().isFailure()) { future.completeExceptionally(creation.error().exception()); } else { future.complete(null); } } } } @Override void handleFailure(Throwable throwable) { completeAllExceptionally(futures.values(), throwable); } }, now); return new CreateAclsResult(new HashMap<AclBinding, KafkaFuture<Void>>(futures)); } private  KafkaAdminClient(AdminClientConfig config, String clientId, Time time, Metadata metadata,
                     Metrics metrics, KafkaClient client, TimeoutProcessorFactory timeoutProcessorFactory); @Override void close(long duration, TimeUnit unit); @Override CreateTopicsResult createTopics(final Collection<NewTopic> newTopics,
                                           final CreateTopicsOptions options); @Override DeleteTopicsResult deleteTopics(final Collection<String> topicNames,
                                           DeleteTopicsOptions options); @Override ListTopicsResult listTopics(final ListTopicsOptions options); @Override DescribeTopicsResult describeTopics(final Collection<String> topicNames, DescribeTopicsOptions options); @Override DescribeClusterResult describeCluster(DescribeClusterOptions options); @Override DescribeAclsResult describeAcls(final AclBindingFilter filter, DescribeAclsOptions options); @Override CreateAclsResult createAcls(Collection<AclBinding> acls, CreateAclsOptions options); @Override DeleteAclsResult deleteAcls(Collection<AclBindingFilter> filters, DeleteAclsOptions options); @Override DescribeConfigsResult describeConfigs(Collection<ConfigResource> configResources, final DescribeConfigsOptions options); @Override AlterConfigsResult alterConfigs(Map<ConfigResource, Config> configs, final AlterConfigsOptions options);  }