### Question:
KafkaAdminClient implements Closeable { public void removeAcls(Set<Acl> acls, Resource resource) { if (acls == null) throw new IllegalArgumentException("acls cannot be null"); if (resource == null) throw new IllegalArgumentException("resource cannot be null"); LOG.debug("Removing ACLs [{}] for resource [{}]", acls, resource); try { getAuthorizer().removeAcls(toImmutableScalaSet(acls), resource); } catch (ZkException | ZooKeeperClientException e) { throw new AdminOperationException("Unable to remove ACLs for resource: " + resource, e); } } KafkaAdminClient(Properties properties); Set<String> getTopics(); Set<TopicAndPartition> getPartitions(); Set<TopicAndPartition> getPartitions(String topic); Authorizer getAuthorizer(); Map<Resource, Set<Acl>> getAcls(); Map<Resource, Set<Acl>> getAcls(KafkaPrincipal principal); Set<Acl> getAcls(Resource resource); void addAcls(Set<Acl> acls, Resource resource); void removeAcls(Set<Acl> acls, Resource resource); void createTopic(String topic, int partitions, int replicationFactor); void createTopic(String topic, int partitions, int replicationFactor, Properties topicConfig); void deleteTopic(String topic); Properties getTopicConfig(String topic); void updateTopicConfig(String topic, Properties properties); int getTopicReplicationFactor(String topic); int getTopicPartitions(String topic); void addTopicPartitions(String topic, int partitions); AdminClient.ConsumerGroupSummary getConsumerGroupSummary(String consumerGroup); Collection<AdminClient.ConsumerSummary> getConsumerGroupSummaries(String consumerGroup); Map<TopicPartition, String> getConsumerGroupAssignments(String consumerGroup); @Override void close(); static final String ZOOKEEPER_SECURE; static final String DEFAULT_ZOOKEEPER_SECURE; static final String OPERATION_TIMEOUT_MS; static final String DEFAULT_OPERATION_TIMEOUT_MS; static final String OPERATION_SLEEP_MS; static final String DEFAULT_OPERATION_SLEEP_MS; }### Answer:
@Test public void removeAcls() { KafkaPrincipal user = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "my_user"); Resource topic1 = Resource.fromString(Topic.name() + Resource.Separator() + "topic1"); Set<Acl> readAcl = Collections.singleton(new Acl(user, Allow$.MODULE$, Acl.WildCardHost(), Read$.MODULE$)); client.addAcls(readAcl, topic1); assertThat(client.getAcls(topic1), is(readAcl)); client.removeAcls(readAcl, topic1); assertThat(client.getAcls(topic1), is(empty())); }
@Test (expected = IllegalArgumentException.class) public void removeAcls_nullAcls() { Resource topic1 = Resource.fromString(Topic.name() + Resource.Separator() + "topic1"); client.removeAcls(null, topic1); }
@Test (expected = IllegalArgumentException.class) public void removeAcls_nullResource() { KafkaPrincipal user = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "my_user"); Set<Acl> readAcl = Collections.singleton(new Acl(user, Allow$.MODULE$, Acl.WildCardHost(), Read$.MODULE$)); client.removeAcls(readAcl, null); }
@Test(expected = AdminOperationException.class) public void removeAcls_zkException() { KafkaPrincipal user = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "user"); Resource resource = Resource.fromString(Topic.name() + Resource.Separator() + "topic"); Set<Acl> readAcl = Collections.singleton(new Acl(user, Allow$.MODULE$, Acl.WildCardHost(), Read$.MODULE$)); failureClient.removeAcls(readAcl, resource); }