From f7ce3471491fe23a94a77c927e2fdb63fc88600e Mon Sep 17 00:00:00 2001
From: Marsel <49659004+MarselAhmetov@users.noreply.github.com>
Date: Fri, 30 Jul 2021 17:51:56 +0500
Subject: [PATCH] fix schemaRegistry property by adding checks (#752)

* fix schemaRegistry property by adding checks

* style fix

* pull request fix

Co-authored-by: marselakhmetov <makhmetov@provectus.com>
---
 .../java/com/provectus/kafka/ui/mapper/ClusterMapper.java    | 4 +++-
 .../serde/schemaregistry/SchemaRegistryAwareRecordSerDe.java | 5 +++--
 2 files changed, 6 insertions(+), 3 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 380fe7d853..9d2e3fc09a 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
@@ -35,6 +35,7 @@ import java.math.BigDecimal;
 import java.nio.file.Path;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Properties;
 import java.util.stream.Collectors;
 import org.apache.kafka.clients.admin.ConfigEntry;
@@ -91,7 +92,8 @@ public interface ClusterMapper {
   Partition toPartition(InternalPartition topic);
 
   default InternalSchemaRegistry setSchemaRegistry(ClustersProperties.Cluster clusterProperties) {
-    if (clusterProperties == null) {
+    if (clusterProperties == null
+        || clusterProperties.getSchemaRegistry() == null) {
       return 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 03f1431b82..747f5c61af 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
@@ -68,8 +68,9 @@ public class SchemaRegistryAwareRecordSerDe implements RecordSerDe {
   private final ObjectMapper objectMapper = new ObjectMapper();
 
   private static SchemaRegistryClient createSchemaRegistryClient(KafkaCluster cluster) {
-    Objects.requireNonNull(cluster.getSchemaRegistry());
-    Objects.requireNonNull(cluster.getSchemaRegistry().getUrl());
+    if (cluster.getSchemaRegistry() == null) {
+      throw new ValidationException("schemaRegistry is not specified");
+    }
     List<SchemaProvider> schemaProviders =
         List.of(new AvroSchemaProvider(), new ProtobufSchemaProvider(), new JsonSchemaProvider());