Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

KAFKA-16865: Add IncludeTopicAuthorizedOperations option for DescribeTopicPartitionsRequest #16136

Merged
merged 4 commits into from
Jun 12, 2024
Merged

KAFKA-16865: Add IncludeTopicAuthorizedOperations option for DescribeTopicPartitionsRequest #16136

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
6911a4c
KAFKA-16865: Add IncludeTopicAuthorizedOperations option for Describe…
tinaselenge May 30, 2024
3835aad
Add unit test
tinaselenge Jun 11, 2024
c4bc52b
Address comments on unit tests
tinaselenge Jun 11, 2024
2b3537a
Address review comment for adding an unit test for filtering out auth…
tinaselenge Jun 12, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2241,7 +2241,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 @@ -2404,14 +2404,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;
tinaselenge marked this conversation as resolved.
Show resolved Hide resolved
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
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1464,13 +1464,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)),
tinaselenge marked this conversation as resolved.
Show resolved Hide resolved
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 @@ -1547,6 +1619,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