Repository: provectus/kafka-ui
Issue: Schema Registry configuration does not support clustered endpoint
Labels: type/bug
scope/backend
status/accepted

**Describe the bug**
All Kafka ecosystem tools support schema registry configuration in form `http(s)://host.1name:8081,http(s)://host2.name:8081,...`

This does not work with this UI, because it will try to do something like:

```
curl http(s)://host.1name:8081,http(s)://host2.name:8081,.../subjects
```

Which obviously crashes.

This is a bug since this contradicts to standard way of defining SR endpoints.


**Set up**
Docker


**Steps to Reproduce**
Steps to reproduce the behavior:

1. Have clustered SR
2. Use standard endpoint to clustered SR in configuration

**Expected behavior**
Works

**Actual behavior**

```
kafka_kafka-ui.1.iqy3stcgghfl@redacted    | 03:37:06.271 [reactor-http-epoll-8] ERROR org.springframework.boot.autoconfigure.web.reactive.error.AbstractErrorWebExceptionHandler - [66f50769]  500 Server Error for HTTP GET "/api/clusters/cluster/schemas"
kafka_kafka-ui.1.iqy3stcgghfl@redacted    | org.springframework.web.reactive.function.client.WebClientResponseException$NotFound: 404 Not Found from GET http://host1.name:8081,http://host2.name:8081,http://host3.name:8081/subjects
kafka_kafka-ui.1.iqy3stcgghfl@redacted    | 	at org.springframework.web.reactive.function.client.WebClientResponseException.create(WebClientResponseException.java:185) ~[spring-webflux-5.2.3.RELEASE.jar!/:5.2.3.RELEASE]
kafka_kafka-ui.1.iqy3stcgghfl@redacted    | 	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
kafka_kafka-ui.1.iqy3stcgghfl@redacted    | Error has been observed at the following site(s):
kafka_kafka-ui.1.iqy3stcgghfl@redacted    | 	|_ checkpoint ⇢ 404 from GET http://host1.name:8081,http://host2.name:8081,http://host3.name:8081/subjects [DefaultWebClient]
kafka_kafka-ui.1.iqy3stcgghfl@redacted    | 	|_ checkpoint ⇢ Handler com.provectus.kafka.ui.controller.SchemasController#getSchemas(String, ServerWebExchange) [DispatcherHandler]
kafka_kafka-ui.1.iqy3stcgghfl@redacted    | 	|_ checkpoint ⇢ com.provectus.kafka.ui.config.ReadOnlyModeFilter [DefaultWebFilterChain]
kafka_kafka-ui.1.iqy3stcgghfl@redacted    | 	|_ checkpoint ⇢ com.provectus.kafka.ui.config.CustomWebFilter [DefaultWebFilterChain]
kafka_kafka-ui.1.iqy3stcgghfl@redacted    | 	|_ checkpoint ⇢ org.springframework.security.web.server.authorization.AuthorizationWebFilter [DefaultWebFilterChain]
kafka_kafka-ui.1.iqy3stcgghfl@redacted    | 	|_ checkpoint ⇢ org.springframework.security.web.server.authorization.ExceptionTranslationWebFilter [DefaultWebFilterChain]
kafka_kafka-ui.1.iqy3stcgghfl@redacted    | 	|_ checkpoint ⇢ org.springframework.security.web.server.authentication.logout.LogoutWebFilter [DefaultWebFilterChain]
kafka_kafka-ui.1.iqy3stcgghfl@redacted    | 	|_ checkpoint ⇢ org.springframework.security.web.server.savedrequest.ServerRequestCacheWebFilter [DefaultWebFilterChain]
kafka_kafka-ui.1.iqy3stcgghfl@redacted    | 	|_ checkpoint ⇢ org.springframework.security.web.server.context.SecurityContextServerWebExchangeWebFilter [DefaultWebFilterChain]
kafka_kafka-ui.1.iqy3stcgghfl@redacted    | 	|_ checkpoint ⇢ org.springframework.security.web.server.context.ReactorContextWebFilter [DefaultWebFilterChain]
kafka_kafka-ui.1.iqy3stcgghfl@redacted    | 	|_ checkpoint ⇢ org.springframework.security.web.server.header.HttpHeaderWriterWebFilter [DefaultWebFilterChain]
kafka_kafka-ui.1.iqy3stcgghfl@redacted    | 	|_ checkpoint ⇢ org.springframework.security.config.web.server.ServerHttpSecurity$ServerWebExchangeReactorContextWebFilter [DefaultWebFilterChain]
kafka_kafka-ui.1.iqy3stcgghfl@redacted    | 	|_ checkpoint ⇢ org.springframework.security.web.server.WebFilterChainProxy [DefaultWebFilterChain]
kafka_kafka-ui.1.iqy3stcgghfl@redacted    | 	|_ checkpoint ⇢ HTTP GET "/api/clusters/cluster/schemas" [ExceptionHandlingWebHandler]
```



Issue comments:
Hi, thanks for creating this issue. We'll try to fix it in the next minor version. 
@germanosin Looks like it is still not fixed on master (based on Docker image from 2 days ago). It now accepts the comma-separated list, but it ONLY communicates with first host in the list. That is not how SchemaRegistry designed to work.

We had scheduled downtime on first node in cluster -- the entire stack works as expected, but kafka-ui could not read schemas.
@akamensky Thanks for checking this, this behaviour would be updated in the next versions.
I agree with @akamensk the implementation does not add much value besides being able to parse and select the first registry URI in a comma separated list. 

@akamensky I suppose you are expecting some kind of failover behavior ?
Something struck me, what if the multiple registries have different authentication credentials ?
@jonasgeiregat I am not sure how that would work. We have them all unauthenticated. But I imagine in clustered setup (where each registry provides exactly same data and uses same topics as data storage) the chances of individual instances having different credentials is very low as that would be an issue for failover cases for any clients.
We’re planning future improvements on this topic anyway. This should work fine for now. 
We’re planning future improvements on this topic anyway. This should work fine for now. 

> On 17 Mar 2022, at 13:49, Alexey Kamenskiy ***@***.***> wrote:
> 
> ﻿
> @jonasgeiregat I am not sure how that would work. We have them all unauthenticated. But I imagine in clustered setup (where each registry provides exactly same data and uses same topics as data storage) the chances of individual instances having different credentials is very low as that would be an issue for failover cases for any clients.
> 
> —
> Reply to this email directly, view it on GitHub, or unsubscribe.
> Triage notifications on the go with GitHub Mobile for iOS or Android. 
> You are receiving this because you modified the open/close state.


Pull request review comments:

Patch that resolved the issue:
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/mapper/ClusterMapper.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/mapper/ClusterMapper.java
index 99efa810d3d..3c53490e251 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/mapper/ClusterMapper.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/mapper/ClusterMapper.java
@@ -33,6 +33,8 @@
 import com.provectus.kafka.ui.model.schemaregistry.InternalCompatibilityLevel;
 import java.math.BigDecimal;
 import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
@@ -101,7 +103,11 @@ default InternalSchemaRegistry setSchemaRegistry(ClustersProperties.Cluster clus
     InternalSchemaRegistry.InternalSchemaRegistryBuilder internalSchemaRegistry =
         InternalSchemaRegistry.builder();
 
-    internalSchemaRegistry.url(clusterProperties.getSchemaRegistry());
+    internalSchemaRegistry.url(
+        clusterProperties.getSchemaRegistry() != null
+            ? Arrays.asList(clusterProperties.getSchemaRegistry().split(","))
+            : Collections.emptyList()
+    );
 
     if (clusterProperties.getSchemaRegistryAuth() != null) {
       internalSchemaRegistry.username(clusterProperties.getSchemaRegistryAuth().getUsername());
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalSchemaRegistry.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalSchemaRegistry.java
index 378f2706f58..0606579d78f 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalSchemaRegistry.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalSchemaRegistry.java
@@ -1,5 +1,6 @@
 package com.provectus.kafka.ui.model;
 
+import java.util.List;
 import lombok.Builder;
 import lombok.Data;
 
@@ -8,5 +9,10 @@
 public class InternalSchemaRegistry {
   private final String username;
   private final String password;
-  private final String url;
+  private final List<String> url;
+
+  public String getFirstUrl() {
+    return url != null  && !url.isEmpty() ? url.iterator().next() : null;
+  }
+
 }
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/serde/schemaregistry/SchemaRegistryAwareRecordSerDe.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/serde/schemaregistry/SchemaRegistryAwareRecordSerDe.java
index fd58804760e..9c218f28430 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/serde/schemaregistry/SchemaRegistryAwareRecordSerDe.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/serde/schemaregistry/SchemaRegistryAwareRecordSerDe.java
@@ -90,7 +90,7 @@ private static SchemaRegistryClient createSchemaRegistryClient(KafkaCluster clus
           "You specified password but do not specified username");
     }
     return new CachedSchemaRegistryClient(
-        Collections.singletonList(cluster.getSchemaRegistry().getUrl()),
+        cluster.getSchemaRegistry().getUrl(),
         CLIENT_IDENTITY_MAP_CAPACITY,
         schemaProviders,
         configs
@@ -218,7 +218,7 @@ public TopicMessageSchema getTopicSchema(String topic) {
   private String convertSchema(SchemaMetadata schema) {
 
     String jsonSchema;
-    URI basePath = new URI(cluster.getSchemaRegistry().getUrl())
+    URI basePath = new URI(cluster.getSchemaRegistry().getFirstUrl())
         .resolve(Integer.toString(schema.getId()));
     final ParsedSchema schemaById = Objects.requireNonNull(schemaRegistryClient)
         .getSchemaById(schema.getId());
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/SchemaRegistryService.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/SchemaRegistryService.java
index 611d6eedf42..c88394b5581 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/SchemaRegistryService.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/SchemaRegistryService.java
@@ -343,8 +343,9 @@ private WebClient.RequestBodySpec configuredWebClient(KafkaCluster cluster, Http
   private WebClient.RequestBodySpec configuredWebClient(InternalSchemaRegistry schemaRegistry,
                                                         HttpMethod method, String uri,
                                                         Object... params) {
-    return webClient.method(method)
-        .uri(schemaRegistry.getUrl() + uri, params)
+    return webClient
+        .method(method)
+        .uri(schemaRegistry.getFirstUrl() + uri, params)
         .headers(headers -> setBasicAuthIfEnabled(schemaRegistry, headers));
   }
 }


Test patch: