BE: SR: Support SR vendor-specific content-types (#1697)

Author: Haarolean

api/src/main/java/io/kafbat/ui/service/KafkaClusterFactory.java

@@ -32,6 +32,7 @@
 import java.util.Properties;
 import java.util.stream.Stream;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.http.MediaType;
 import org.springframework.stereotype.Service;
 import org.springframework.util.StringUtils;
 import org.springframework.util.unit.DataSize;
@@ -48,6 +49,10 @@ public class KafkaClusterFactory {
   private static final DataSize DEFAULT_WEBCLIENT_BUFFER = DataSize.parse("20MB");
   private static final Duration DEFAULT_RESPONSE_TIMEOUT = Duration.ofSeconds(20);
 
+  // Confluent Schema Registry API content types (used by WarpStream and other compatible implementations)
+  private static final MediaType SR_V1_JSON = MediaType.parseMediaType("application/vnd.schemaregistry.v1+json");
+  private static final MediaType SR_JSON = MediaType.parseMediaType("application/vnd.schemaregistry+json");
+
   private final DataSize webClientMaxBuffSize;
   private final Duration responseTimeout;
   private final JmxMetricsRetriever jmxMetricsRetriever;
@@ -233,6 +238,7 @@ private ReactiveFailover<KafkaSrClientApi> schemaRegistryClient(ClustersProperti
         .configureSsl(clusterProperties.getSsl(), clusterProperties.getSchemaRegistrySsl())
         .configureBasicAuth(auth.getUsername(), auth.getPassword())
         .configureBufferSize(webClientMaxBuffSize)
+        .configureAdditionalDecoderMediaTypes(SR_V1_JSON, SR_JSON)
         .build();
     return ReactiveFailover.create(
         parseUrlList(clusterProperties.getSchemaRegistry()),

api/src/main/java/io/kafbat/ui/util/WebClientConfigurator.java

@@ -12,6 +12,7 @@
 import java.security.KeyStore;
 import java.time.Duration;
 import java.util.function.Consumer;
+import java.util.stream.Stream;
 import javax.annotation.Nullable;
 import javax.net.ssl.KeyManagerFactory;
 import javax.net.ssl.TrustManagerFactory;
@@ -33,9 +34,10 @@ public class WebClientConfigurator {
   private HttpClient httpClient = HttpClient
       .create()
       .proxyWithSystemProperties();
+  private ObjectMapper objectMapper = defaultOM();
 
   public WebClientConfigurator() {
-    configureObjectMapper(defaultOM());
+    configureObjectMapper(objectMapper);
   }
 
   private static ObjectMapper defaultOM() {
@@ -131,6 +133,7 @@ public WebClientConfigurator configureBufferSize(DataSize maxBuffSize) {
   }
 
   public void configureObjectMapper(ObjectMapper mapper) {
+    this.objectMapper = mapper;
     builder.codecs(codecs -> {
       codecs.defaultCodecs()
           .jackson2JsonEncoder(new Jackson2JsonEncoder(mapper, MediaType.APPLICATION_JSON));
@@ -139,6 +142,20 @@ public void configureObjectMapper(ObjectMapper mapper) {
     });
   }
 
+  public WebClientConfigurator configureAdditionalDecoderMediaTypes(MediaType... additionalMediaTypes) {
+    builder.codecs(codecs -> {
+      codecs.defaultCodecs()
+          .jackson2JsonEncoder(new Jackson2JsonEncoder(objectMapper, MediaType.APPLICATION_JSON));
+      MediaType[] allMediaTypes = Stream.concat(
+          Stream.of(MediaType.APPLICATION_JSON),
+          Stream.of(additionalMediaTypes)
+      ).toArray(MediaType[]::new);
+      codecs.defaultCodecs()
+          .jackson2JsonDecoder(new Jackson2JsonDecoder(objectMapper, allMediaTypes));
+    });
+    return this;
+  }
+
   public WebClientConfigurator configureCodecs(Consumer<ClientCodecConfigurer> configurer) {
     builder.codecs(configurer);
     return this;

api/src/test/java/io/kafbat/ui/util/WebClientConfiguratorTest.java

@@ -0,0 +1,91 @@
+package io.kafbat.ui.util;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.http.MediaType;
+import org.springframework.web.reactive.function.client.WebClient;
+
+class WebClientConfiguratorTest {
+
+  private final MockWebServer mockWebServer = new MockWebServer();
+
+  @BeforeEach
+  void startMockServer() throws IOException {
+    mockWebServer.start();
+  }
+
+  @AfterEach
+  void stopMockServer() throws IOException {
+    mockWebServer.close();
+  }
+
+  @Test
+  void decodesStandardApplicationJson() {
+    mockWebServer.enqueue(new MockResponse()
+        .addHeader("Content-Type", "application/json")
+        .setBody("{\"name\":\"test\"}"));
+
+    WebClient client = new WebClientConfigurator().build();
+    String body = client.get()
+        .uri(mockWebServer.url("/").toString())
+        .retrieve()
+        .bodyToMono(String.class)
+        .block();
+
+    assertThat(body).isEqualTo("{\"name\":\"test\"}");
+  }
+
+  @Test
+  void decodesVendorMediaTypeWhenConfigured() {
+    MediaType srV1Json = MediaType.parseMediaType("application/vnd.schemaregistry.v1+json");
+
+    mockWebServer.enqueue(new MockResponse()
+        .addHeader("Content-Type", "application/vnd.schemaregistry.v1+json")
+        .setBody("{\"compatibilityLevel\":\"BACKWARD\"}"));
+
+    WebClient client = new WebClientConfigurator()
+        .configureAdditionalDecoderMediaTypes(srV1Json)
+        .build();
+
+    CompatibilityConfig config = client.get()
+        .uri(mockWebServer.url("/config").toString())
+        .retrieve()
+        .bodyToMono(CompatibilityConfig.class)
+        .block();
+
+    assertThat(config).isNotNull();
+    assertThat(config.compatibilityLevel()).isEqualTo("BACKWARD");
+  }
+
+  @Test
+  void decodesMultipleVendorMediaTypes() {
+    MediaType srV1Json = MediaType.parseMediaType("application/vnd.schemaregistry.v1+json");
+    MediaType srJson = MediaType.parseMediaType("application/vnd.schemaregistry+json");
+
+    mockWebServer.enqueue(new MockResponse()
+        .addHeader("Content-Type", "application/vnd.schemaregistry+json")
+        .setBody("{\"compatibilityLevel\":\"FULL\"}"));
+
+    WebClient client = new WebClientConfigurator()
+        .configureAdditionalDecoderMediaTypes(srV1Json, srJson)
+        .build();
+
+    CompatibilityConfig config = client.get()
+        .uri(mockWebServer.url("/config").toString())
+        .retrieve()
+        .bodyToMono(CompatibilityConfig.class)
+        .block();
+
+    assertThat(config).isNotNull();
+    assertThat(config.compatibilityLevel()).isEqualTo("FULL");
+  }
+
+  record CompatibilityConfig(String compatibilityLevel) {
+  }
+}