Skip to content

Dinf 2950 Fix schema prefix removal and ensure consistent topic name tranformation #1471

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

Closed
wants to merge 4 commits into from
Closed

Dinf 2950 Fix schema prefix removal and ensure consistent topic name tranformation #1471

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
673d4a1
Fix: Remove environment prefix from topic names before schema lookup …
Nov 5, 2025
6baf000
DINF-2950: Fix schema prefix removal and ensure consistent topic name…
Nov 5, 2025
2ba01b5
Merge branch 'main' into DINF-2950-schema-prefix-fix
ashok-pnt Nov 5, 2025
36cc1c4
Fix checkstyle: Move ternary operator to new line per code style rules
Nov 5, 2025
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
84 changes: 84 additions & 0 deletions 0001-Fix-Remove-environment-prefix-from-topic-names-befor.patch
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,84 @@
From 673d4a12f8196e100f131cc397cf76fd06f7e261 Mon Sep 17 00:00:00 2001
From: Ashok B <[email protected]>
Date: Wed, 5 Nov 2025 16:20:00 +0530
Subject: [PATCH] Fix: Remove environment prefix from topic names before schema
lookup and update tests for consistency

---
.../builtin/sr/SchemaRegistrySerde.java | 9 ++++++++-
.../integration/odd/TopicsExporter.java | 7 ++++++-
.../builtin/sr/SchemaRegistrySerdeTest.java | 20 +++++++++++++++++++
3 files changed, 34 insertions(+), 2 deletions(-)

diff --git a/api/src/main/java/io/kafbat/ui/serdes/builtin/sr/SchemaRegistrySerde.java b/api/src/main/java/io/kafbat/ui/serdes/builtin/sr/SchemaRegistrySerde.java
index 91c2375d..3d566dcf 100644
--- a/api/src/main/java/io/kafbat/ui/serdes/builtin/sr/SchemaRegistrySerde.java
+++ b/api/src/main/java/io/kafbat/ui/serdes/builtin/sr/SchemaRegistrySerde.java
@@ -243,7 +243,14 @@ public class SchemaRegistrySerde implements BuiltInSerde {
}

private String schemaSubject(String topic, Target type) {
- return String.format(type == Target.KEY ? keySchemaNameTemplate : valueSchemaNameTemplate, topic);
+ // Transform topic name from {env}.{topic_name} to {topic_name}
+ // Example: "development.integration_events.connection_management.entities.connection"
+ // becomes "integration_events.connection_management.entities.connection"
+ String transformedTopic = topic.contains(".") ?
+ topic.substring(topic.indexOf(".") + 1) : topic;
+
+ // Apply the schema naming template with the transformed topic name
+ return String.format(type == Target.KEY ? keySchemaNameTemplate : valueSchemaNameTemplate, transformedTopic);
}

@Override
diff --git a/api/src/main/java/io/kafbat/ui/service/integration/odd/TopicsExporter.java b/api/src/main/java/io/kafbat/ui/service/integration/odd/TopicsExporter.java
index d34e0a61..1960b97a 100644
--- a/api/src/main/java/io/kafbat/ui/service/integration/odd/TopicsExporter.java
+++ b/api/src/main/java/io/kafbat/ui/service/integration/odd/TopicsExporter.java
@@ -103,7 +103,12 @@ class TopicsExporter {
if (cluster.getSchemaRegistryClient() == null) {
return Mono.just(List.of());
}
- String subject = topic + (isKey ? "-key" : "-value");
+ // Transform topic name from {env}.{topic_name} to {topic_name}
+ // Example: "development.integration_events.connection_management.entities.connection"
+ // becomes "integration_events.connection_management.entities.connection"
+ String transformedTopic = topic.contains(".") ?
+ topic.substring(topic.indexOf(".") + 1) : topic;
+ String subject = transformedTopic + (isKey ? "-key" : "-value");
return getSubjWithResolvedRefs(cluster, subject)
.map(t -> DataSetFieldsExtractors.extract(t.getT1(), t.getT2(), topicOddrn, isKey))
.onErrorResume(WebClientResponseException.NotFound.class, th -> Mono.just(List.of()))
diff --git a/api/src/test/java/io/kafbat/ui/serdes/builtin/sr/SchemaRegistrySerdeTest.java b/api/src/test/java/io/kafbat/ui/serdes/builtin/sr/SchemaRegistrySerdeTest.java
index d66a8d00..54b1875f 100644
--- a/api/src/test/java/io/kafbat/ui/serdes/builtin/sr/SchemaRegistrySerdeTest.java
+++ b/api/src/test/java/io/kafbat/ui/serdes/builtin/sr/SchemaRegistrySerdeTest.java
@@ -70,6 +70,26 @@ class SchemaRegistrySerdeTest {
assertThat(serde.getSchema(topic, Serde.Target.VALUE)).isEmpty();
}

+ @Test
+ @SneakyThrows
+ void transformsEnvironmentPrefixedTopicNames() {
+ // Test with environment-prefixed topic name
+ String topic = "development.integration_events.connection_management.entities.connection";
+ String expectedSubject = "integration_events.connection_management.entities.connection-key";
+
+ int schemaId = registryClient.register(expectedSubject, new AvroSchema("{ \"type\": \"int\" }"));
+ int registeredVersion = registryClient.getLatestSchemaMetadata(expectedSubject).getVersion();
+
+ var schemaOptional = serde.getSchema(topic, Serde.Target.KEY);
+ assertThat(schemaOptional).isPresent();
+
+ SchemaDescription schemaDescription = schemaOptional.get();
+ assertThat(schemaDescription.getAdditionalProperties())
+ .containsEntry("subject", expectedSubject)
+ .containsEntry("schemaId", schemaId)
+ .containsEntry("latestVersion", registeredVersion);
+ }
+
@Test
void serializeTreatsInputAsJsonAvroSchemaPayload() throws RestClientException, IOException {
AvroSchema schema = new AvroSchema(
--
2.50.1

Empty file added DINF-2950.patch
View file
Open in desktop
Empty file.
10 changes: 9 additions & 1 deletion api/src/main/java/io/kafbat/ui/serdes/builtin/sr/SchemaRegistrySerde.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -243,7 +243,15 @@ private <T> Optional<T> wrapWith404Handler(Callable<T> call) {
}

private String schemaSubject(String topic, Target type) {
return String.format(type == Target.KEY ? keySchemaNameTemplate : valueSchemaNameTemplate, topic);
// Transform topic name from {env}.{topic_name} to {topic_name}
// Example: "development.integration_events.connection_management.entities.connection"
// becomes "integration_events.connection_management.entities.connection"
String transformedTopic = topic.contains(".")
Copy link
Member

@Haarolean Haarolean Nov 6, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You have a wrong assumption that every user of the app has the same schema naming as you do. We won't proceed with such changes

? topic.substring(topic.indexOf(".") + 1)
: topic;

// Apply the schema naming template with the transformed topic name
return String.format(type == Target.KEY ? keySchemaNameTemplate : valueSchemaNameTemplate, transformedTopic);
}

@Override
Expand Down
8 changes: 7 additions & 1 deletion api/src/main/java/io/kafbat/ui/service/integration/odd/TopicsExporter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,7 +103,13 @@ private Mono<List<DataSetField>> getTopicSchema(KafkaCluster cluster,
if (cluster.getSchemaRegistryClient() == null) {
return Mono.just(List.of());
}
String subject = topic + (isKey ? "-key" : "-value");
// Transform topic name from {env}.{topic_name} to {topic_name}
// Example: "development.integration_events.connection_management.entities.connection"
// becomes "integration_events.connection_management.entities.connection"
String transformedTopic = topic.contains(".")
? topic.substring(topic.indexOf(".") + 1)
: topic;
String subject = transformedTopic + (isKey ? "-key" : "-value");
return getSubjWithResolvedRefs(cluster, subject)
.map(t -> DataSetFieldsExtractors.extract(t.getT1(), t.getT2(), topicOddrn, isKey))
.onErrorResume(WebClientResponseException.NotFound.class, th -> Mono.just(List.of()))
Expand Down
20 changes: 20 additions & 0 deletions api/src/test/java/io/kafbat/ui/serdes/builtin/sr/SchemaRegistrySerdeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,26 @@ void returnsEmptyDescriptorIfSchemaNotRegisteredInSR() {
assertThat(serde.getSchema(topic, Serde.Target.VALUE)).isEmpty();
}

@Test
@SneakyThrows
void transformsEnvironmentPrefixedTopicNames() {
// Test with environment-prefixed topic name
String topic = "development.integration_events.connection_management.entities.connection";
String expectedSubject = "integration_events.connection_management.entities.connection-key";

int schemaId = registryClient.register(expectedSubject, new AvroSchema("{ \"type\": \"int\" }"));
int registeredVersion = registryClient.getLatestSchemaMetadata(expectedSubject).getVersion();

var schemaOptional = serde.getSchema(topic, Serde.Target.KEY);
assertThat(schemaOptional).isPresent();

SchemaDescription schemaDescription = schemaOptional.get();
assertThat(schemaDescription.getAdditionalProperties())
.containsEntry("subject", expectedSubject)
.containsEntry("schemaId", schemaId)
.containsEntry("latestVersion", registeredVersion);
}

@Test
void serializeTreatsInputAsJsonAvroSchemaPayload() throws RestClientException, IOException {
AvroSchema schema = new AvroSchema(
Expand Down
Loading