Skip to content

Commit

Permalink
KAFKA-16865: Add IncludeTopicAuthorizedOperations option for Describe…
Browse files Browse the repository at this point in the history 
…TopicPartitionsRequest (apache#16136)


Reviewers: Mickael Maison <[email protected]>, Chia-Ping Tsai <[email protected]>, Calvin Liu <[email protected]>, Andrew Schofield <[email protected]>, Apoorv Mittal <[email protected]>
  • Loading branch information
@tinaselenge
tinaselenge authored Jun 12, 2024
1 parent 46eb081 commit 9368ef8
Show file tree
Hide file tree
Showing 2 changed files with 78 additions and 3 deletions.
8 changes: 5 additions & 3 deletions clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2245,7 +2245,7 @@ void handleResponse(AbstractResponse abstractResponse) {
continue;
}

TopicDescription currentTopicDescription = getTopicDescriptionFromDescribeTopicsResponseTopic(topic, nodes);
TopicDescription currentTopicDescription = getTopicDescriptionFromDescribeTopicsResponseTopic(topic, nodes, options.includeAuthorizedOperations());

if (partiallyFinishedTopicDescription != null && partiallyFinishedTopicDescription.name().equals(topicName)) {
// Add the partitions for the cursor topic of the previous batch.
Expand Down Expand Up @@ -2408,14 +2408,16 @@ void handleFailure(Throwable throwable) {

private TopicDescription getTopicDescriptionFromDescribeTopicsResponseTopic(
DescribeTopicPartitionsResponseTopic topic,
Map<Integer, Node> nodes
Map<Integer, Node> nodes,
boolean includeAuthorizedOperations
) {
List<DescribeTopicPartitionsResponsePartition> partitionInfos = topic.partitions();
List<TopicPartitionInfo> partitions = new ArrayList<>(partitionInfos.size());
for (DescribeTopicPartitionsResponsePartition partitionInfo : partitionInfos) {
partitions.add(DescribeTopicPartitionsResponse.partitionToTopicPartitionInfo(partitionInfo, nodes));
}
return new TopicDescription(topic.name(), topic.isInternal(), partitions, validAclOperations(topic.topicAuthorizedOperations()), topic.topicId());
Set<AclOperation> authorisedOperations = includeAuthorizedOperations ? validAclOperations(topic.topicAuthorizedOperations()) : null;
return new TopicDescription(topic.name(), topic.isInternal(), partitions, authorisedOperations, topic.topicId());
}

private TopicDescription getTopicDescriptionFromCluster(Cluster cluster, String topicName, Uuid topicId,
Expand Down
73 changes: 73 additions & 0 deletions clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1474,13 +1474,85 @@ public void testDescribeTopicsWithDescribeTopicPartitionsApiBasic() {
assertEquals(0, topicDescription.partitions().get(0).partition());
assertEquals(1, topicDescription.partitions().get(1).partition());
topicDescription = topicDescriptions.get(topicName1);
assertNull(topicDescription.authorizedOperations());
assertEquals(1, topicDescription.partitions().size());
} catch (Exception e) {
fail("describe using DescribeTopics API should not fail", e);
}
}
}

@Test
public void testDescribeTopicPartitionsApiWithAuthorizedOps() throws ExecutionException, InterruptedException {
try (AdminClientUnitTestEnv env = mockClientEnv()) {
env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
String topicName0 = "test-0";
Uuid topicId = Uuid.randomUuid();

int authorisedOperations = Utils.to32BitField(Utils.mkSet(AclOperation.DESCRIBE.code(), AclOperation.ALTER.code()));
env.kafkaClient().prepareResponse(
prepareDescribeClusterResponse(0,
env.cluster().nodes(),
env.cluster().clusterResource().clusterId(),
2,
authorisedOperations)
);

DescribeTopicPartitionsResponseData responseData = new DescribeTopicPartitionsResponseData();
responseData.topics().add(new DescribeTopicPartitionsResponseTopic()
.setErrorCode((short) 0)
.setTopicId(topicId)
.setName(topicName0)
.setIsInternal(false)
.setTopicAuthorizedOperations(authorisedOperations));
env.kafkaClient().prepareResponse(new DescribeTopicPartitionsResponse(responseData));

DescribeTopicsResult result = env.adminClient().describeTopics(
singletonList(topicName0), new DescribeTopicsOptions().includeAuthorizedOperations(true)
);

Map<String, TopicDescription> topicDescriptions = result.allTopicNames().get();
TopicDescription topicDescription = topicDescriptions.get(topicName0);
assertEquals(new HashSet<>(asList(AclOperation.DESCRIBE, AclOperation.ALTER)),
topicDescription.authorizedOperations());
}
}

@Test
public void testDescribeTopicPartitionsApiWithoutAuthorizedOps() throws ExecutionException, InterruptedException {
try (AdminClientUnitTestEnv env = mockClientEnv()) {
env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
String topicName0 = "test-0";
Uuid topicId = Uuid.randomUuid();

int authorisedOperations = Utils.to32BitField(Utils.mkSet(AclOperation.DESCRIBE.code(), AclOperation.ALTER.code()));
env.kafkaClient().prepareResponse(
prepareDescribeClusterResponse(0,
env.cluster().nodes(),
env.cluster().clusterResource().clusterId(),
2,
authorisedOperations)
);

DescribeTopicPartitionsResponseData responseData = new DescribeTopicPartitionsResponseData();
responseData.topics().add(new DescribeTopicPartitionsResponseTopic()
.setErrorCode((short) 0)
.setTopicId(topicId)
.setName(topicName0)
.setIsInternal(false)
.setTopicAuthorizedOperations(authorisedOperations));
env.kafkaClient().prepareResponse(new DescribeTopicPartitionsResponse(responseData));

DescribeTopicsResult result = env.adminClient().describeTopics(
singletonList(topicName0), new DescribeTopicsOptions().includeAuthorizedOperations(false)
);

Map<String, TopicDescription> topicDescriptions = result.allTopicNames().get();
TopicDescription topicDescription = topicDescriptions.get(topicName0);
assertNull(topicDescription.authorizedOperations());
}
}

@SuppressWarnings({"NPathComplexity", "CyclomaticComplexity"})
@Test
public void testDescribeTopicsWithDescribeTopicPartitionsApiEdgeCase() {
Expand Down Expand Up @@ -1554,6 +1626,7 @@ public void testDescribeTopicsWithDescribeTopicPartitionsApiEdgeCase() {
assertEquals(2, topicDescription.partitions().size());
topicDescription = topicDescriptions.get(topicName2);
assertEquals(2, topicDescription.partitions().size());
assertNull(topicDescription.authorizedOperations());
} catch (Exception e) {
fail("describe using DescribeTopics API should not fail", e);
}
Expand Down

0 comments on commit 9368ef8

Please sign in to comment.