BE: SchemaRegistry references support

kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/TopicsController.java

@@ -123,9 +123,11 @@ public Mono<ResponseEntity<Void>> deleteTopic(
  .operationName("deleteTopic")
  .build();
 
- return accessControlService.validateAccess(context).then(
- topicsService.deleteTopic(getCluster(clusterName), topicName).map(ResponseEntity::ok)
- ).doOnEach(sig -> auditService.audit(context, sig));
+ return accessControlService.validateAccess(context)
+ .then(
+ topicsService.deleteTopic(getCluster(clusterName), topicName)
+ .thenReturn(ResponseEntity.ok().<Void>build())
+ ).doOnEach(sig -> auditService.audit(context, sig));
  }
 
 

kafka-ui-api/src/main/java/com/provectus/kafka/ui/mapper/KafkaSrMapper.java

@@ -3,18 +3,21 @@
 import com.provectus.kafka.ui.model.CompatibilityCheckResponseDTO;
 import com.provectus.kafka.ui.model.CompatibilityLevelDTO;
 import com.provectus.kafka.ui.model.NewSchemaSubjectDTO;
+import com.provectus.kafka.ui.model.SchemaReferenceDTO;
 import com.provectus.kafka.ui.model.SchemaSubjectDTO;
 import com.provectus.kafka.ui.model.SchemaTypeDTO;
 import com.provectus.kafka.ui.service.SchemaRegistryService;
 import com.provectus.kafka.ui.sr.model.Compatibility;
 import com.provectus.kafka.ui.sr.model.CompatibilityCheckResponse;
 import com.provectus.kafka.ui.sr.model.NewSubject;
+import com.provectus.kafka.ui.sr.model.SchemaReference;
 import com.provectus.kafka.ui.sr.model.SchemaType;
+import java.util.List;
 import java.util.Optional;
 import org.mapstruct.Mapper;
 
 
-@Mapper(componentModel = "spring")
+@Mapper
 public interface KafkaSrMapper {
 
  default SchemaSubjectDTO toDto(SchemaRegistryService.SubjectWithCompatibilityLevel s) {
@@ -24,9 +27,12 @@ default SchemaSubjectDTO toDto(SchemaRegistryService.SubjectWithCompatibilityLev
  .subject(s.getSubject())
  .schema(s.getSchema())
  .schemaType(SchemaTypeDTO.fromValue(Optional.ofNullable(s.getSchemaType()).orElse(SchemaType.AVRO).getValue()))
+ .references(toDto(s.getReferences()))
  .compatibilityLevel(s.getCompatibility().toString());
  }
 
+ List<SchemaReferenceDTO> toDto(List<SchemaReference> references);
+
  CompatibilityCheckResponseDTO toDto(CompatibilityCheckResponse ccr);
 
  CompatibilityLevelDTO.CompatibilityEnum toDto(Compatibility compatibility);

kafka-ui-api/src/main/java/com/provectus/kafka/ui/serdes/builtin/sr/SchemaRegistrySerde.java

@@ -20,6 +20,7 @@
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClientConfig;
 import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException;
+import io.confluent.kafka.schemaregistry.json.JsonSchema;
 import io.confluent.kafka.schemaregistry.json.JsonSchemaProvider;
 import io.confluent.kafka.schemaregistry.protobuf.ProtobufSchema;
 import io.confluent.kafka.schemaregistry.protobuf.ProtobufSchemaProvider;
@@ -217,7 +218,9 @@ private String convertSchema(SchemaMetadata schema, ParsedSchema parsedSchema) {
  case AVRO -> new AvroJsonSchemaConverter()
  .convert(basePath, ((AvroSchema) parsedSchema).rawSchema())
  .toJson();
- case JSON -> schema.getSchema();
+ case JSON ->
+ //need to use confluent JsonSchema since it includes resolved references
+ ((JsonSchema) parsedSchema).rawSchema().toString();
  };
  }
 

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/SchemaRegistryService.java

@@ -14,8 +14,7 @@
 import com.provectus.kafka.ui.sr.model.NewSubject;
 import com.provectus.kafka.ui.sr.model.SchemaSubject;
 import com.provectus.kafka.ui.util.ReactiveFailover;
-import com.provectus.kafka.ui.util.WebClientConfigurator;
-import java.io.IOException;
+import java.nio.charset.Charset;
 import java.util.List;
 import java.util.stream.Collectors;
 import lombok.AllArgsConstructor;
@@ -92,7 +91,7 @@ public Mono<SubjectWithCompatibilityLevel> getLatestSchemaVersionBySubject(Kafka
  private Mono<SubjectWithCompatibilityLevel> getSchemaSubject(KafkaCluster cluster, String schemaName,
  String version) {
  return api(cluster)
- .mono(c -> c.getSubjectVersion(schemaName, version))
+ .mono(c -> c.getSubjectVersion(schemaName, version, false))
  .zipWith(getSchemaCompatibilityInfoOrGlobal(cluster, schemaName))
  .map(t -> new SubjectWithCompatibilityLevel(t.getT1(), t.getT2()))
  .onErrorResume(WebClientResponseException.NotFound.class, th -> Mono.error(new SchemaNotFoundException()));

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/integration/odd/SchemaReferencesResolver.java

@@ -0,0 +1,55 @@
+package com.provectus.kafka.ui.service.integration.odd;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import com.provectus.kafka.ui.sr.api.KafkaSrClientApi;
+import com.provectus.kafka.ui.sr.model.SchemaReference;
+import java.util.List;
+import java.util.Optional;
+import javax.annotation.Nullable;
+import reactor.core.publisher.Mono;
+
+// logic copied from AbstractSchemaProvider:resolveReferences
+// https://github.com/confluentinc/schema-registry/blob/fd59613e2c5adf62e36705307f420712e4c8c1ea/client/src/main/java/io/confluent/kafka/schemaregistry/AbstractSchemaProvider.java#L54
+class SchemaReferencesResolver {
+
+ private final KafkaSrClientApi client;
+
+ SchemaReferencesResolver(KafkaSrClientApi client) {
+ this.client = client;
+ }
+
+ Mono<ImmutableMap<String, String>> resolve(List<SchemaReference> refs) {
+ return resolveReferences(refs, new Resolving(ImmutableMap.of(), ImmutableSet.of()))
+ .map(Resolving::resolved);
+ }
+
+ private record Resolving(ImmutableMap<String, String> resolved, ImmutableSet<String> visited) {
+
+ Resolving visit(String name) {
+ return new Resolving(resolved, ImmutableSet.<String>builder().addAll(visited).add(name).build());
+ }
+
+ Resolving resolve(String ref, String schema) {
+ return new Resolving(ImmutableMap.<String, String>builder().putAll(resolved).put(ref, schema).build(), visited);
+ }
+ }
+
+ private Mono<Resolving> resolveReferences(@Nullable List<SchemaReference> refs, Resolving initState) {
+ Mono<Resolving> result = Mono.just(initState);
+ for (SchemaReference reference : Optional.ofNullable(refs).orElse(List.of())) {
+ result = result.flatMap(state -> {
+ if (state.visited().contains(reference.getName())) {
+ return Mono.just(state);
+ } else {
+ final var newState = state.visit(reference.getName());
+ return client.getSubjectVersion(reference.getSubject(), String.valueOf(reference.getVersion()), true)
+ .flatMap(subj ->
+ resolveReferences(subj.getReferences(), newState)
+ .map(withNewRefs -> withNewRefs.resolve(reference.getName(), subj.getSchema())));
+ }
+ });
+ }
+ return result;
+ }
+}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/integration/odd/TopicsExporter.java

@@ -5,6 +5,7 @@
 import com.provectus.kafka.ui.model.Statistics;
 import com.provectus.kafka.ui.service.StatisticsCache;
 import com.provectus.kafka.ui.service.integration.odd.schema.DataSetFieldsExtractors;
+import com.provectus.kafka.ui.sr.model.SchemaSubject;
 import java.net.URI;
 import java.util.List;
 import java.util.Map;
@@ -24,6 +25,8 @@
 import org.springframework.web.reactive.function.client.WebClientResponseException;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.util.function.Tuple2;
+import reactor.util.function.Tuples;
 
 @Slf4j
 @RequiredArgsConstructor
@@ -100,12 +103,20 @@ private Mono<List<DataSetField>> getTopicSchema(KafkaCluster cluster,
  return Mono.just(List.of());
  }
  String subject = topic + (isKey ? "-key" : "-value");
- return cluster.getSchemaRegistryClient()
- .mono(client -> client.getSubjectVersion(subject, "latest"))
- .map(subj -> DataSetFieldsExtractors.extract(subj, topicOddrn, isKey))
+ return getSubjWithResolvedRefs(cluster, subject)
+ .map(t -> DataSetFieldsExtractors.extract(t.getT1(), t.getT2(), topicOddrn, isKey))
  .onErrorResume(WebClientResponseException.NotFound.class, th -> Mono.just(List.of()))
  .onErrorMap(WebClientResponseException.class, err ->
  new IllegalStateException("Error retrieving subject %s".formatted(subject), err));
  }
 
+ private Mono<Tuple2<SchemaSubject, Map<String, String>>> getSubjWithResolvedRefs(KafkaCluster cluster,
+ String subjectName) {
+ return cluster.getSchemaRegistryClient()
+ .mono(client ->
+ client.getSubjectVersion(subjectName, "latest", false)
+ .flatMap(subj -> new SchemaReferencesResolver(client).resolve(subj.getReferences())
+ .map(resolvedRefs -> Tuples.of(subj, resolvedRefs))));
+ }
+
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/integration/odd/schema/AvroExtractor.java

@@ -1,7 +1,7 @@
 package com.provectus.kafka.ui.service.integration.odd.schema;
 
 import com.google.common.collect.ImmutableSet;
-import com.provectus.kafka.ui.sr.model.SchemaSubject;
+import io.confluent.kafka.schemaregistry.avro.AvroSchema;
 import java.util.ArrayList;
 import java.util.List;
 import org.apache.avro.Schema;
@@ -14,8 +14,8 @@ final class AvroExtractor {
  private AvroExtractor() {
  }
 
- static List<DataSetField> extract(SchemaSubject subject, KafkaPath topicOddrn, boolean isKey) {
- var schema = new Schema.Parser().parse(subject.getSchema());
+ static List<DataSetField> extract(AvroSchema avroSchema, KafkaPath topicOddrn, boolean isKey) {
+ var schema = avroSchema.rawSchema();
  List<DataSetField> result = new ArrayList<>();
  result.add(DataSetFieldsExtractors.rootField(topicOddrn, isKey));
  extract(

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/integration/odd/schema/DataSetFieldsExtractors.java

@@ -2,20 +2,30 @@
 
 import com.provectus.kafka.ui.sr.model.SchemaSubject;
 import com.provectus.kafka.ui.sr.model.SchemaType;
+import io.confluent.kafka.schemaregistry.avro.AvroSchema;
+import io.confluent.kafka.schemaregistry.json.JsonSchema;
+import io.confluent.kafka.schemaregistry.protobuf.ProtobufSchema;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import org.opendatadiscovery.client.model.DataSetField;
 import org.opendatadiscovery.client.model.DataSetFieldType;
 import org.opendatadiscovery.oddrn.model.KafkaPath;
 
 public final class DataSetFieldsExtractors {
 
- public static List<DataSetField> extract(SchemaSubject subject, KafkaPath topicOddrn, boolean isKey) {
+ public static List<DataSetField> extract(SchemaSubject subject,
+ Map<String, String> resolvedRefs,
+ KafkaPath topicOddrn,
+ boolean isKey) {
  SchemaType schemaType = Optional.ofNullable(subject.getSchemaType()).orElse(SchemaType.AVRO);
  return switch (schemaType) {
- case AVRO -> AvroExtractor.extract(subject, topicOddrn, isKey);
- case JSON -> JsonSchemaExtractor.extract(subject, topicOddrn, isKey);
- case PROTOBUF -> ProtoExtractor.extract(subject, topicOddrn, isKey);
+ case AVRO -> AvroExtractor.extract(
+ new AvroSchema(subject.getSchema(), List.of(), resolvedRefs, null), topicOddrn, isKey);
+ case JSON -> JsonSchemaExtractor.extract(
+ new JsonSchema(subject.getSchema(), List.of(), resolvedRefs, null), topicOddrn, isKey);
+ case PROTOBUF -> ProtoExtractor.extract(
+ new ProtobufSchema(subject.getSchema(), List.of(), resolvedRefs, null, null), topicOddrn, isKey);
  };
  }
 

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/integration/odd/schema/JsonSchemaExtractor.java

@@ -30,8 +30,8 @@ final class JsonSchemaExtractor {
  private JsonSchemaExtractor() {
  }
 
- static List<DataSetField> extract(SchemaSubject subject, KafkaPath topicOddrn, boolean isKey) {
- Schema schema = new JsonSchema(subject.getSchema()).rawSchema();
+ static List<DataSetField> extract(JsonSchema jsonSchema, KafkaPath topicOddrn, boolean isKey) {
+ Schema schema = jsonSchema.rawSchema();
  List<DataSetField> result = new ArrayList<>();
  result.add(DataSetFieldsExtractors.rootField(topicOddrn, isKey));
  extract(

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/integration/odd/schema/ProtoExtractor.java

@@ -15,7 +15,6 @@
 import com.google.protobuf.UInt32Value;
 import com.google.protobuf.UInt64Value;
 import com.google.protobuf.Value;
-import com.provectus.kafka.ui.sr.model.SchemaSubject;
 import io.confluent.kafka.schemaregistry.protobuf.ProtobufSchema;
 import java.util.ArrayList;
 import java.util.List;
@@ -42,8 +41,8 @@ final class ProtoExtractor {
  private ProtoExtractor() {
  }
 
- static List<DataSetField> extract(SchemaSubject subject, KafkaPath topicOddrn, boolean isKey) {
- Descriptor schema = new ProtobufSchema(subject.getSchema()).toDescriptor();
+ static List<DataSetField> extract(ProtobufSchema protobufSchema, KafkaPath topicOddrn, boolean isKey) {
+ Descriptor schema = protobufSchema.toDescriptor();
  List<DataSetField> result = new ArrayList<>();
  result.add(DataSetFieldsExtractors.rootField(topicOddrn, isKey));
  var rootOddrn = topicOddrn.oddrn() + "/columns/" + (isKey ? "key" : "value");

kafka-ui-api/src/test/java/com/provectus/kafka/ui/SchemaRegistryServiceTests.java

@@ -2,6 +2,7 @@
 
 import com.provectus.kafka.ui.model.CompatibilityLevelDTO;
 import com.provectus.kafka.ui.model.NewSchemaSubjectDTO;
+import com.provectus.kafka.ui.model.SchemaReferenceDTO;
 import com.provectus.kafka.ui.model.SchemaSubjectDTO;
 import com.provectus.kafka.ui.model.SchemaSubjectsResponseDTO;
 import com.provectus.kafka.ui.model.SchemaTypeDTO;
@@ -190,6 +191,58 @@ void shouldCreateNewProtobufSchema() {
  Assertions.assertEquals(schema, actual.getSchema());
  }
 
+
+ @Test
+ void shouldCreateNewProtobufSchemaWithRefs() {
+ NewSchemaSubjectDTO requestBody = new NewSchemaSubjectDTO()
+ .schemaType(SchemaTypeDTO.PROTOBUF)
+ .subject(subject + "-ref")
+ .schema("""
+ syntax = "proto3";
+ message MyRecord {
+ int32 id = 1;
+ string name = 2;
+ }
+ """);
+
+ webTestClient
+ .post()
+ .uri("/api/clusters/{clusterName}/schemas", LOCAL)
+ .contentType(MediaType.APPLICATION_JSON)
+ .body(BodyInserters.fromPublisher(Mono.just(requestBody), NewSchemaSubjectDTO.class))
+ .exchange()
+ .expectStatus()
+ .isOk();
+
+ requestBody = new NewSchemaSubjectDTO()
+ .schemaType(SchemaTypeDTO.PROTOBUF)
+ .subject(subject)
+ .schema("""
+ syntax = "proto3";
+ import "MyRecord.proto";
+ message MyRecordWithRef {
+ int32 id = 1;
+ MyRecord my_ref = 2;
+ }
+ """)
+ .references(List.of(new SchemaReferenceDTO().name("MyRecord.proto").subject(subject + "-ref").version(1)));
+
+ SchemaSubjectDTO actual = webTestClient
+ .post()
+ .uri("/api/clusters/{clusterName}/schemas", LOCAL)
+ .contentType(MediaType.APPLICATION_JSON)
+ .body(BodyInserters.fromPublisher(Mono.just(requestBody), NewSchemaSubjectDTO.class))
+ .exchange()
+ .expectStatus()
+ .isOk()
+ .expectBody(SchemaSubjectDTO.class)
+ .returnResult()
+ .getResponseBody();
+
+ Assertions.assertNotNull(actual);
+ Assertions.assertEquals(requestBody.getReferences(), actual.getReferences());
+ }
+
  @Test
  public void shouldReturnBackwardAsGlobalCompatibilityLevelByDefault() {
  webTestClient
@@ -278,7 +331,7 @@ public void shouldOkWhenCreateNewSchemaThenGetAndUpdateItsCompatibilityLevel() {
  void shouldCreateNewSchemaWhenSubjectIncludesNonAsciiCharacters() {
  String schema =
  "{\"subject\":\"test/test\",\"schemaType\":\"JSON\",\"schema\":"
- + "\"{\\\"type\\\": \\\"string\\\"}\"}";
+  + "\"{\\\"type\\\": \\\"string\\\"}\"}";
 
  webTestClient
  .post()