From 5e74bda56ec19e95c3ed11b349360691ad6e6914 Mon Sep 17 00:00:00 2001
From: German Osin <german.osin@gmail.com>
Date: Fri, 17 Sep 2021 11:35:34 +0300
Subject: [PATCH] ISSUE-886 Schema registry multiple urls (#888)

---
 .../java/com/provectus/kafka/ui/mapper/ClusterMapper.java | 8 +++++++-
 .../provectus/kafka/ui/model/InternalSchemaRegistry.java  | 8 +++++++-
 .../schemaregistry/SchemaRegistryAwareRecordSerDe.java    | 4 ++--
 .../provectus/kafka/ui/service/SchemaRegistryService.java | 5 +++--
 4 files changed, 19 insertions(+), 6 deletions(-)

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 99efa810d3..3c53490e25 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.InternalCompatibilityCheck;
 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 @@ public interface ClusterMapper {
     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 378f2706f5..0606579d78 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 @@ import lombok.Data;
 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 fd58804760..9c218f2843 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 @@ public class SchemaRegistryAwareRecordSerDe implements RecordSerDe {
           "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 class SchemaRegistryAwareRecordSerDe implements RecordSerDe {
   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 611d6eedf4..c88394b558 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 @@ public class SchemaRegistryService {
   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));
   }
 }