Skip to content

Rename TopicData to TopicMetadata #7

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

Merged
showuon merged 1 commit into from
Jun 18, 2026
Merged

Rename TopicData to TopicMetadata #7

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
14 changes: 7 additions & 7 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 @@ -4945,7 +4945,7 @@ public StartMirrorTopicsResult startMirrorTopics(String mirrorName, Set<String>
validateRegexPatterns(options.excludePatterns());

// Fetch source metadata so the controller can create topics synchronously
Map<String, StartMirrorTopicsRequestData.TopicData> topicMetadata;
Map<String, StartMirrorTopicsRequestData.TopicMetadata> topicMetadata;
if (!topics.isEmpty()) {
try {
topicMetadata = fetchSourceTopicMetadata(mirrorName, topics);
Expand All @@ -4966,9 +4966,9 @@ StartMirrorTopicsRequest.Builder createRequest(int timeoutMs) {
StartMirrorTopicsRequestData data = new StartMirrorTopicsRequestData();
data.setMirrorName(mirrorName);
topics.forEach(t -> {
StartMirrorTopicsRequestData.TopicData existing = topicMetadata.get(t);
StartMirrorTopicsRequestData.TopicMetadata existing = topicMetadata.get(t);
data.topics().add(existing != null ? existing
: new StartMirrorTopicsRequestData.TopicData().setTopicName(t));
: new StartMirrorTopicsRequestData.TopicMetadata().setTopicName(t));
});
data.setIncludePatterns(options.includePatterns());
data.setExcludePatterns(options.excludePatterns());
Expand Down Expand Up @@ -5002,7 +5002,7 @@ void handleFailure(Throwable throwable) {
return new StartMirrorTopicsResult(future);
}

private Map<String, StartMirrorTopicsRequestData.TopicData> fetchSourceTopicMetadata(
private Map<String, StartMirrorTopicsRequestData.TopicMetadata> fetchSourceTopicMetadata(
String mirrorName, Set<String> topics) throws Exception {
ConfigResource mirrorResource = new ConfigResource(ConfigResource.Type.CLUSTER_MIRROR, mirrorName);
var configResult = describeConfigs(List.of(mirrorResource)).all().get();
Expand All @@ -5018,9 +5018,9 @@ private Map<String, StartMirrorTopicsRequestData.TopicData> fetchSourceTopicMeta

try (Admin sourceAdmin = Admin.create(sourceProps)) {
var descriptions = sourceAdmin.describeTopics(topics).allTopicNames().get();
Map<String, StartMirrorTopicsRequestData.TopicData> metadata = new HashMap<>();
Map<String, StartMirrorTopicsRequestData.TopicMetadata> metadata = new HashMap<>();
descriptions.forEach((name, desc) ->
metadata.put(name, new StartMirrorTopicsRequestData.TopicData()
metadata.put(name, new StartMirrorTopicsRequestData.TopicMetadata()
.setTopicName(name)
.setTopicId(desc.topicId())
.setNumPartitions(desc.partitions().size())));
Expand All @@ -5042,7 +5042,7 @@ public StopMirrorTopicsResult stopMirrorTopics(String mirrorName, Set<String> to
StopMirrorTopicsRequest.Builder createRequest(int timeoutMs) {
StopMirrorTopicsRequestData data = new StopMirrorTopicsRequestData();
data.setMirrorName(mirrorName);
topics.forEach(t -> data.topics().add(new StopMirrorTopicsRequestData.TopicData().setTopicName(t)));
topics.forEach(t -> data.topics().add(new StopMirrorTopicsRequestData.TopicMetadata().setTopicName(t)));
data.setPatterns(options.patterns());
return new StopMirrorTopicsRequest.Builder(data);
}
Expand Down
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/common/requests/PauseMirrorTopicsRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ public Builder(String mirrorName, Set<String> topics) {
ApiKeys.PAUSE_MIRROR_TOPICS.latestVersion());
PauseMirrorTopicsRequestData data = new PauseMirrorTopicsRequestData();
data.setMirrorName(mirrorName);
topics.forEach(topic -> data.topics().add(new PauseMirrorTopicsRequestData.TopicData().setTopicName(topic)));
topics.forEach(topic -> data.topics().add(new PauseMirrorTopicsRequestData.TopicMetadata().setTopicName(topic)));
this.data = data;
}

Expand Down
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/common/requests/ResumeMirrorTopicsRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ public Builder(String mirrorName, Set<String> topics) {
ApiKeys.RESUME_MIRROR_TOPICS.latestVersion());
ResumeMirrorTopicsRequestData data = new ResumeMirrorTopicsRequestData();
data.setMirrorName(mirrorName);
topics.forEach(topic -> data.topics().add(new ResumeMirrorTopicsRequestData.TopicData().setTopicName(topic)));
topics.forEach(topic -> data.topics().add(new ResumeMirrorTopicsRequestData.TopicMetadata().setTopicName(topic)));
this.data = data;
}

Expand Down
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/common/requests/StartMirrorTopicsRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ public Builder(String mirrorName, Set<String> topics) {
ApiKeys.START_MIRROR_TOPICS.latestVersion());
StartMirrorTopicsRequestData data = new StartMirrorTopicsRequestData();
data.setMirrorName(mirrorName);
topics.forEach(topic -> data.topics().add(new StartMirrorTopicsRequestData.TopicData().setTopicName(topic)));
topics.forEach(topic -> data.topics().add(new StartMirrorTopicsRequestData.TopicMetadata().setTopicName(topic)));
this.data = data;
}

Expand Down
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/common/requests/StopMirrorTopicsRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ public Builder(String mirrorName, Set<String> topics) {
ApiKeys.STOP_MIRROR_TOPICS.latestVersion());
StopMirrorTopicsRequestData data = new StopMirrorTopicsRequestData();
data.setMirrorName(mirrorName);
topics.forEach(topic -> data.topics().add(new StopMirrorTopicsRequestData.TopicData().setTopicName(topic)));
topics.forEach(topic -> data.topics().add(new StopMirrorTopicsRequestData.TopicMetadata().setTopicName(topic)));
this.data = data;
}

Expand Down
4 changes: 2 additions & 2 deletions clients/src/main/resources/common/message/PauseMirrorTopicsRequest.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,8 @@
"flexibleVersions": "0+",
"fields": [
{ "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
"about": "The mirror name to pause the topics for." },
{ "name": "Topics", "type": "[]TopicData", "versions": "0+", "about": "The data for the topics.",
"about": "The cluster mirror name." },
{ "name": "Topics", "type": "[]TopicMetadata", "versions": "0+", "about": "The data for the topics.",
"fields": [
{ "name": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID."},
{ "name": "TopicName", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName",
Expand Down
2 changes: 1 addition & 1 deletion clients/src/main/resources/common/message/ReadMirrorStatesRequest.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
"fields": [
{ "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
"about": "The cluster mirror name." },
{ "name": "Topics", "type": "[]TopicData", "versions": "0+",
{ "name": "Topics", "type": "[]TopicMetadata", "versions": "0+",
"about": "The data for the topics.", "fields": [
{ "name": "Name", "type": "string", "versions": "0+", "entityType": "topicName",
"about": "The topic name." },
Expand Down
2 changes: 1 addition & 1 deletion clients/src/main/resources/common/message/ResumeMirrorTopicsRequest.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
"fields": [
{ "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
"about": "The cluster mirror name." },
{ "name": "Topics", "type": "[]TopicData", "versions": "0+", "about": "The data for the topics.",
{ "name": "Topics", "type": "[]TopicMetadata", "versions": "0+", "about": "The data for the topics.",
"fields": [
{ "name": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID."},
{ "name": "TopicName", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName",
Expand Down
2 changes: 1 addition & 1 deletion clients/src/main/resources/common/message/StartMirrorTopicsRequest.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
"fields": [
{ "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
"about": "The cluster mirror name." },
{ "name": "Topics", "type": "[]TopicData", "versions": "0+", "about": "The data for the topics.",
{ "name": "Topics", "type": "[]TopicMetadata", "versions": "0+", "about": "The data for the topics.",
"fields": [
{ "name": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID."},
{ "name": "TopicName", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName",
Expand Down
2 changes: 1 addition & 1 deletion clients/src/main/resources/common/message/StopMirrorTopicsRequest.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
"fields": [
{ "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
"about": "The cluster mirror name." },
{ "name": "Topics", "type": "[]TopicData", "versions": "0+", "about": "The data for the topics.",
{ "name": "Topics", "type": "[]TopicMetadata", "versions": "0+", "about": "The data for the topics.",
"fields": [
{ "name": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID."},
{ "name": "TopicName", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName",
Expand Down
4 changes: 2 additions & 2 deletions clients/src/main/resources/common/message/WriteMirrorStatesRequest.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,8 @@
"flexibleVersions": "0+",
"fields": [
{ "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
"about": "The mirror name." },
{ "name": "Topics", "type": "[]TopicData", "versions": "0+",
"about": "The cluster mirror name." },
{ "name": "Topics", "type": "[]TopicMetadata", "versions": "0+",
"about": "The data for the topics.", "fields": [
{ "name": "Name", "type": "string", "versions": "0+", "entityType": "topicName",
"about": "The topic name." },
Expand Down
20 changes: 10 additions & 10 deletions clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1223,8 +1223,8 @@ private AbstractResponse getResponse(ApiKeys apikey, short version) {
}

private StartMirrorTopicsRequest createStartMirrorTopicsRequest(short version) {
StartMirrorTopicsRequestData.TopicDataCollection topics = new StartMirrorTopicsRequestData.TopicDataCollection();
topics.add(new StartMirrorTopicsRequestData.TopicData()
StartMirrorTopicsRequestData.TopicMetadataCollection topics = new StartMirrorTopicsRequestData.TopicMetadataCollection();
topics.add(new StartMirrorTopicsRequestData.TopicMetadata()
.setTopicId(Uuid.randomUuid())
.setTopicName("topic")
);
Expand All @@ -1242,8 +1242,8 @@ private StartMirrorTopicsResponse createStartMirrorTopicsResponse() {
}

private StopMirrorTopicsRequest createStopMirrorTopicsRequest(short version) {
StopMirrorTopicsRequestData.TopicDataCollection topics = new StopMirrorTopicsRequestData.TopicDataCollection();
topics.add(new StopMirrorTopicsRequestData.TopicData()
StopMirrorTopicsRequestData.TopicMetadataCollection topics = new StopMirrorTopicsRequestData.TopicMetadataCollection();
topics.add(new StopMirrorTopicsRequestData.TopicMetadata()
.setTopicId(Uuid.randomUuid())
.setTopicName("topic")
);
Expand Down Expand Up @@ -1303,7 +1303,7 @@ public DescribeClusterMirrorsResponse createDescribeClusterMirrorsResponse() {
public ReadMirrorStatesRequest createReadMirrorStatesRequest(short version) {
ReadMirrorStatesRequestData data = new ReadMirrorStatesRequestData()
.setMirrorName("mirror")
.setTopics(List.of(new ReadMirrorStatesRequestData.TopicData()
.setTopics(List.of(new ReadMirrorStatesRequestData.TopicMetadata()
.setName("topic")
.setPartitions(List.of(new ReadMirrorStatesRequestData.PartitionData().setPartitionIndex(0)))
));
Expand All @@ -1325,7 +1325,7 @@ public ReadMirrorStatesResponse createReadMirrorStatesResponse() {
public WriteMirrorStatesRequest createWriteMirrorStatesRequest(short version) {
WriteMirrorStatesRequestData data = new WriteMirrorStatesRequestData()
.setMirrorName("mirror")
.setTopics(List.of(new WriteMirrorStatesRequestData.TopicData()
.setTopics(List.of(new WriteMirrorStatesRequestData.TopicMetadata()
.setName("topic")
.setPartitions(List.of(new WriteMirrorStatesRequestData.PartitionData()
.setPartitionIndex(0)
Expand All @@ -1346,8 +1346,8 @@ public WriteMirrorStatesResponse createWriteMirrorStatesResponse() {
}

public PauseMirrorTopicsRequest createPauseMirrorTopicsRequest(short version) {
PauseMirrorTopicsRequestData.TopicDataCollection topics = new PauseMirrorTopicsRequestData.TopicDataCollection();
topics.add(new PauseMirrorTopicsRequestData.TopicData()
PauseMirrorTopicsRequestData.TopicMetadataCollection topics = new PauseMirrorTopicsRequestData.TopicMetadataCollection();
topics.add(new PauseMirrorTopicsRequestData.TopicMetadata()
.setTopicId(Uuid.randomUuid())
.setTopicName("topic")
);
Expand All @@ -1366,8 +1366,8 @@ public PauseMirrorTopicsResponse createPauseMirrorTopicsResponse() {
}

public ResumeMirrorTopicsRequest createResumeMirrorTopicsRequest(short version) {
ResumeMirrorTopicsRequestData.TopicDataCollection topics = new ResumeMirrorTopicsRequestData.TopicDataCollection();
topics.add(new ResumeMirrorTopicsRequestData.TopicData()
ResumeMirrorTopicsRequestData.TopicMetadataCollection topics = new ResumeMirrorTopicsRequestData.TopicMetadataCollection();
topics.add(new ResumeMirrorTopicsRequestData.TopicMetadata()
.setTopicId(Uuid.randomUuid())
.setTopicName("topic")
);
Expand Down
14 changes: 7 additions & 7 deletions core/src/main/java/kafka/server/mirror/MirrorMetadataManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1250,7 +1250,7 @@ private void discoverTopicsByPattern(String mirrorName, ClusterMirrorConfig mirr
Set<String> configuredTopics = getConfiguredTopics(mirrorName, true);
final Pattern topicsExcludePattern = mirrorConfig.topicsExcludePattern();

List<StartMirrorTopicsRequestData.TopicData> newTopics;
List<StartMirrorTopicsRequestData.TopicMetadata> newTopics;
try {
Set<String> allSourceTopics = srcAdmin.listTopics()
.names().get(brokerConfig.requestTimeoutMs(), TimeUnit.MILLISECONDS);
Expand All @@ -1271,7 +1271,7 @@ private void discoverTopicsByPattern(String mirrorName, ClusterMirrorConfig mirr
cacheSourceLeaders(mirrorName, descriptions.values());

newTopics = descriptions.values().stream()
.map(td -> new StartMirrorTopicsRequestData.TopicData()
.map(td -> new StartMirrorTopicsRequestData.TopicMetadata()
.setTopicName(td.name())
.setTopicId(td.topicId())
.setNumPartitions(td.partitions().size()))
Expand All @@ -1286,7 +1286,7 @@ private void discoverTopicsByPattern(String mirrorName, ClusterMirrorConfig mirr
}

log.info("Discovered {} new topic(s) matching mirror.topics.include pattern for mirror {}: {}",
newTopics.size(), mirrorName, newTopics.stream().map(StartMirrorTopicsRequestData.TopicData::topicName).toList());
newTopics.size(), mirrorName, newTopics.stream().map(StartMirrorTopicsRequestData.TopicMetadata::topicName).toList());

StartMirrorTopicsRequestData data = new StartMirrorTopicsRequestData();
data.setMirrorName(mirrorName);
Expand Down Expand Up @@ -1459,10 +1459,10 @@ void writeStatesToRemoteCoordinator(String mirrorName,
// Send one batched request per coordinator node
nodeToTopicPartitions.forEach((node, topicPartitionsMap) -> {
WriteMirrorStatesRequestData data = new WriteMirrorStatesRequestData().setMirrorName(mirrorName);
List<WriteMirrorStatesRequestData.TopicData> topicDataList = new ArrayList<>();
List<WriteMirrorStatesRequestData.TopicMetadata> topicDataList = new ArrayList<>();

topicPartitionsMap.forEach((topic, partitionDataList) ->
topicDataList.add(new WriteMirrorStatesRequestData.TopicData()
topicDataList.add(new WriteMirrorStatesRequestData.TopicMetadata()
.setName(topic)
.setPartitions(partitionDataList)));

Expand Down Expand Up @@ -1517,10 +1517,10 @@ void readStatesFromRemoteCoordinator(String mirrorName,
// Send one batched request per coordinator node
nodeToTopicPartitions.forEach((node, topicPartitionsMap) -> {
ReadMirrorStatesRequestData data = new ReadMirrorStatesRequestData().setMirrorName(mirrorName);
List<ReadMirrorStatesRequestData.TopicData> topicDataList = new ArrayList<>();
List<ReadMirrorStatesRequestData.TopicMetadata> topicDataList = new ArrayList<>();

topicPartitionsMap.forEach((topic, partitionDataList) ->
topicDataList.add(new ReadMirrorStatesRequestData.TopicData()
topicDataList.add(new ReadMirrorStatesRequestData.TopicMetadata()
.setName(topic)
.setPartitions(partitionDataList)));

Expand Down