@Test public void testDeleteAcls() 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 DeleteAclsResponse(0, asList( new AclFilterResponse(asList(new AclDeletionResult(ACL1), new AclDeletionResult(ACL2))), new AclFilterResponse(new ApiError(Errors.SECURITY_DISABLED, "No security"), Collections.<AclDeletionResult>emptySet())))); DeleteAclsResult results = env.adminClient().deleteAcls(asList(FILTER1, FILTER2)); Map<AclBindingFilter, KafkaFuture<FilterResults>> filterResults = results.values(); FilterResults filter1Results = filterResults.get(FILTER1).get(); assertEquals(null, filter1Results.values().get(0).exception()); assertEquals(ACL1, filter1Results.values().get(0).binding()); assertEquals(null, filter1Results.values().get(1).exception()); assertEquals(ACL2, filter1Results.values().get(1).binding()); assertFutureError(filterResults.get(FILTER2), SecurityDisabledException.class); assertFutureError(results.all(), SecurityDisabledException.class); env.kafkaClient().prepareResponse(new DeleteAclsResponse(0, asList( new AclFilterResponse(asList(new AclDeletionResult(ACL1), new AclDeletionResult(new ApiError(Errors.SECURITY_DISABLED, "No security"), ACL2))), new AclFilterResponse(Collections.<AclDeletionResult>emptySet())))); results = env.adminClient().deleteAcls(asList(FILTER1, FILTER2)); assertTrue(results.values().get(FILTER2).get().values().isEmpty()); assertFutureError(results.all(), SecurityDisabledException.class); env.kafkaClient().prepareResponse(new DeleteAclsResponse(0, asList( new AclFilterResponse(asList(new AclDeletionResult(ACL1))), new AclFilterResponse(asList(new AclDeletionResult(ACL2)))))); results = env.adminClient().deleteAcls(asList(FILTER1, FILTER2)); Collection<AclBinding> deleted = results.all().get(); assertCollectionIs(deleted, ACL1, ACL2); } }
@Override public DeleteAclsResult deleteAcls(Collection<AclBindingFilter> filters, DeleteAclsOptions options) { final long now = time.milliseconds(); final Map<AclBindingFilter, KafkaFutureImpl<FilterResults>> futures = new HashMap<>(); final List<AclBindingFilter> filterList = new ArrayList<>(); for (AclBindingFilter filter : filters) { if (futures.get(filter) == null) { filterList.add(filter); futures.put(filter, new KafkaFutureImpl<FilterResults>()); } } runnable.call(new Call("deleteAcls", calcDeadlineMs(now, options.timeoutMs()), new LeastLoadedNodeProvider()) { @Override AbstractRequest.Builder createRequest(int timeoutMs) { return new DeleteAclsRequest.Builder(filterList); } @Override void handleResponse(AbstractResponse abstractResponse) { DeleteAclsResponse response = (DeleteAclsResponse) abstractResponse; List<AclFilterResponse> responses = response.responses(); Iterator<AclFilterResponse> iter = responses.iterator(); for (AclBindingFilter filter : filterList) { KafkaFutureImpl<FilterResults> future = futures.get(filter); if (!iter.hasNext()) { future.completeExceptionally(new UnknownServerException( "The broker reported no deletion result for the given filter.")); } else { AclFilterResponse deletion = iter.next(); if (deletion.error().isFailure()) { future.completeExceptionally(deletion.error().exception()); } else { List<FilterResult> filterResults = new ArrayList<>(); for (AclDeletionResult deletionResult : deletion.deletions()) { filterResults.add(new FilterResult(deletionResult.acl(), deletionResult.error().exception())); } future.complete(new FilterResults(filterResults)); } } } } @Override void handleFailure(Throwable throwable) { completeAllExceptionally(futures.values(), throwable); } }, now); return new DeleteAclsResult(new HashMap<AclBindingFilter, KafkaFuture<FilterResults>>(futures)); }
KafkaAdminClient extends AdminClient { @Override public DeleteAclsResult deleteAcls(Collection<AclBindingFilter> filters, DeleteAclsOptions options) { final long now = time.milliseconds(); final Map<AclBindingFilter, KafkaFutureImpl<FilterResults>> futures = new HashMap<>(); final List<AclBindingFilter> filterList = new ArrayList<>(); for (AclBindingFilter filter : filters) { if (futures.get(filter) == null) { filterList.add(filter); futures.put(filter, new KafkaFutureImpl<FilterResults>()); } } runnable.call(new Call("deleteAcls", calcDeadlineMs(now, options.timeoutMs()), new LeastLoadedNodeProvider()) { @Override AbstractRequest.Builder createRequest(int timeoutMs) { return new DeleteAclsRequest.Builder(filterList); } @Override void handleResponse(AbstractResponse abstractResponse) { DeleteAclsResponse response = (DeleteAclsResponse) abstractResponse; List<AclFilterResponse> responses = response.responses(); Iterator<AclFilterResponse> iter = responses.iterator(); for (AclBindingFilter filter : filterList) { KafkaFutureImpl<FilterResults> future = futures.get(filter); if (!iter.hasNext()) { future.completeExceptionally(new UnknownServerException( "The broker reported no deletion result for the given filter.")); } else { AclFilterResponse deletion = iter.next(); if (deletion.error().isFailure()) { future.completeExceptionally(deletion.error().exception()); } else { List<FilterResult> filterResults = new ArrayList<>(); for (AclDeletionResult deletionResult : deletion.deletions()) { filterResults.add(new FilterResult(deletionResult.acl(), deletionResult.error().exception())); } future.complete(new FilterResults(filterResults)); } } } } @Override void handleFailure(Throwable throwable) { completeAllExceptionally(futures.values(), throwable); } }, now); return new DeleteAclsResult(new HashMap<AclBindingFilter, KafkaFuture<FilterResults>>(futures)); } }
KafkaAdminClient extends AdminClient { @Override public DeleteAclsResult deleteAcls(Collection<AclBindingFilter> filters, DeleteAclsOptions options) { final long now = time.milliseconds(); final Map<AclBindingFilter, KafkaFutureImpl<FilterResults>> futures = new HashMap<>(); final List<AclBindingFilter> filterList = new ArrayList<>(); for (AclBindingFilter filter : filters) { if (futures.get(filter) == null) { filterList.add(filter); futures.put(filter, new KafkaFutureImpl<FilterResults>()); } } runnable.call(new Call("deleteAcls", calcDeadlineMs(now, options.timeoutMs()), new LeastLoadedNodeProvider()) { @Override AbstractRequest.Builder createRequest(int timeoutMs) { return new DeleteAclsRequest.Builder(filterList); } @Override void handleResponse(AbstractResponse abstractResponse) { DeleteAclsResponse response = (DeleteAclsResponse) abstractResponse; List<AclFilterResponse> responses = response.responses(); Iterator<AclFilterResponse> iter = responses.iterator(); for (AclBindingFilter filter : filterList) { KafkaFutureImpl<FilterResults> future = futures.get(filter); if (!iter.hasNext()) { future.completeExceptionally(new UnknownServerException( "The broker reported no deletion result for the given filter.")); } else { AclFilterResponse deletion = iter.next(); if (deletion.error().isFailure()) { future.completeExceptionally(deletion.error().exception()); } else { List<FilterResult> filterResults = new ArrayList<>(); for (AclDeletionResult deletionResult : deletion.deletions()) { filterResults.add(new FilterResult(deletionResult.acl(), deletionResult.error().exception())); } future.complete(new FilterResults(filterResults)); } } } } @Override void handleFailure(Throwable throwable) { completeAllExceptionally(futures.values(), throwable); } }, now); return new DeleteAclsResult(new HashMap<AclBindingFilter, KafkaFuture<FilterResults>>(futures)); } private  KafkaAdminClient(AdminClientConfig config, String clientId, Time time, Metadata metadata,
                     Metrics metrics, KafkaClient client, TimeoutProcessorFactory timeoutProcessorFactory); }
KafkaAdminClient extends AdminClient { @Override public DeleteAclsResult deleteAcls(Collection<AclBindingFilter> filters, DeleteAclsOptions options) { final long now = time.milliseconds(); final Map<AclBindingFilter, KafkaFutureImpl<FilterResults>> futures = new HashMap<>(); final List<AclBindingFilter> filterList = new ArrayList<>(); for (AclBindingFilter filter : filters) { if (futures.get(filter) == null) { filterList.add(filter); futures.put(filter, new KafkaFutureImpl<FilterResults>()); } } runnable.call(new Call("deleteAcls", calcDeadlineMs(now, options.timeoutMs()), new LeastLoadedNodeProvider()) { @Override AbstractRequest.Builder createRequest(int timeoutMs) { return new DeleteAclsRequest.Builder(filterList); } @Override void handleResponse(AbstractResponse abstractResponse) { DeleteAclsResponse response = (DeleteAclsResponse) abstractResponse; List<AclFilterResponse> responses = response.responses(); Iterator<AclFilterResponse> iter = responses.iterator(); for (AclBindingFilter filter : filterList) { KafkaFutureImpl<FilterResults> future = futures.get(filter); if (!iter.hasNext()) { future.completeExceptionally(new UnknownServerException( "The broker reported no deletion result for the given filter.")); } else { AclFilterResponse deletion = iter.next(); if (deletion.error().isFailure()) { future.completeExceptionally(deletion.error().exception()); } else { List<FilterResult> filterResults = new ArrayList<>(); for (AclDeletionResult deletionResult : deletion.deletions()) { filterResults.add(new FilterResult(deletionResult.acl(), deletionResult.error().exception())); } future.complete(new FilterResults(filterResults)); } } } } @Override void handleFailure(Throwable throwable) { completeAllExceptionally(futures.values(), throwable); } }, now); return new DeleteAclsResult(new HashMap<AclBindingFilter, KafkaFuture<FilterResults>>(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 DeleteAclsResult deleteAcls(Collection<AclBindingFilter> filters, DeleteAclsOptions options) { final long now = time.milliseconds(); final Map<AclBindingFilter, KafkaFutureImpl<FilterResults>> futures = new HashMap<>(); final List<AclBindingFilter> filterList = new ArrayList<>(); for (AclBindingFilter filter : filters) { if (futures.get(filter) == null) { filterList.add(filter); futures.put(filter, new KafkaFutureImpl<FilterResults>()); } } runnable.call(new Call("deleteAcls", calcDeadlineMs(now, options.timeoutMs()), new LeastLoadedNodeProvider()) { @Override AbstractRequest.Builder createRequest(int timeoutMs) { return new DeleteAclsRequest.Builder(filterList); } @Override void handleResponse(AbstractResponse abstractResponse) { DeleteAclsResponse response = (DeleteAclsResponse) abstractResponse; List<AclFilterResponse> responses = response.responses(); Iterator<AclFilterResponse> iter = responses.iterator(); for (AclBindingFilter filter : filterList) { KafkaFutureImpl<FilterResults> future = futures.get(filter); if (!iter.hasNext()) { future.completeExceptionally(new UnknownServerException( "The broker reported no deletion result for the given filter.")); } else { AclFilterResponse deletion = iter.next(); if (deletion.error().isFailure()) { future.completeExceptionally(deletion.error().exception()); } else { List<FilterResult> filterResults = new ArrayList<>(); for (AclDeletionResult deletionResult : deletion.deletions()) { filterResults.add(new FilterResult(deletionResult.acl(), deletionResult.error().exception())); } future.complete(new FilterResults(filterResults)); } } } } @Override void handleFailure(Throwable throwable) { completeAllExceptionally(futures.values(), throwable); } }, now); return new DeleteAclsResult(new HashMap<AclBindingFilter, KafkaFuture<FilterResults>>(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);  }