KAFKA-20537: Return topic ID for existing topics

[lh0156]

Summary

Return the existing topic ID in the CreateTopicsResponse when a
requested topic already exists, and expose that ID through
CreateTopicsResult.topicId(topic).

Motivation

The controller already has the topic ID in the controller metadata
image, so returning it avoids an additional metadata lookup for clients
that need the ID. The AdminClient now preserves that ID independently
from the topic-creation status.

Compatibility

• CreateTopicsResult.topicId(topic) succeeds with the existing ID for
  TOPIC_ALREADY_EXISTS.
• CreateTopicsResult.values() and all() continue to fail with
  TopicExistsException, preserving their existing semantics.
• Older responses without topic metadata preserve the existing
  UnsupportedVersionException behavior for metadata accessors.

Validation

• ./gradlew :clients:test --tests org.apache.kafka.clients.admin.KafkaAdminClientTest --no-build-cache --console=plain (291 tests passed)
• ./gradlew :metadata:test --tests org.apache.kafka.controller.ReplicationControlManagerTest --no-build-cache --console=plain (75 tests passed)
• Checkstyle and SpotBugs completed successfully for the affected
  modules.

Reviewers: Sushant Mahajan smahajan@confluent.io

[Copilot]

Pull request overview

This PR updates the controller’s CreateTopics handling so that when a requested topic already exists, the CreateTopicsResponse includes the existing topic’s ID instead of the default zero UUID, avoiding an extra metadata lookup for clients that need the ID.

Changes:

• Populate CreatableTopicResult.topicId when the error is TOPIC_ALREADY_EXISTS.
• Extend the ReplicationControlManagerTest#testCreateTopics assertion to validate the existing topic ID is returned on the already-exists response.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated no comments.

File	Description
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java	Sets the topic ID in CreateTopicsResponseData for TOPIC_ALREADY_EXISTS results using the controller’s topicsByName mapping.
metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java	Updates testCreateTopics to expect the existing topic ID in the TOPIC_ALREADY_EXISTS response.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[smjn]

@lh0156 Do you want to update the admin client as well to actually display the topic id which you are returning?

[lh0156]

Implemented the AdminClient follow-up in cfc1ef1.

CreateTopicsResult.topicId(topic) now completes with the existing topic ID when the controller returns TOPIC_ALREADY_EXISTS, while values() and all() retain their existing TopicExistsException behavior. For older responses without topic metadata, topicId() preserves the existing UnsupportedVersionException behavior without changing values()/all().

Added focused coverage for the existing-topic response and the unsupported-metadata compatibility path. Verification on the latest trunk:

• :clients:test --tests org.apache.kafka.clients.admin.KafkaAdminClientTest (291 tests passed)
• :metadata:test --tests org.apache.kafka.controller.ReplicationControlManagerTest (75 tests passed)
• Checkstyle and SpotBugs completed successfully for the affected modules.

The two follow-up commits include Kafka's required Generated-by: OpenAI Codex (GPT-5) trailer.

[lh0156]

Status follow-up: the implementation and required JUnit/build checks are green. The only failing status is Gradle Build Scan / Java 25, which reports a build-scan publication error; rerunning that workflow requires repository-admin access. No code failure is indicated. The PR is ready for maintainer review once the infrastructure check is rerun or waived.