Repository: provectus/kafka-ui
Issue: Topic Messages: Preview AVRO broken in v0.5.0+
Labels: type/bug
scope/backend
status/accepted

**Describe the bug** (Actual behavior)
In v0.5.0 (and not currently fix on master), the preview feature of avro messages are garbagled:

![image](https://user-images.githubusercontent.com/26322144/212092253-a9a39b40-ac29-49cc-a7ea-1de30deee726.png)

**Expected behavior**
This is a regression, used to work in v0.4.0:

![image](https://user-images.githubusercontent.com/26322144/212092293-26a19dbc-8745-4e61-be37-60e941ad7b68.png)

**Steps to Reproduce**
```yaml
version: "3.8"
services:
    networks:
      - default
    ports:
      - "139:139"
      - "446:445"
    restart: unless-stopped

  zookeeper:
    image: confluentinc/cp-zookeeper:7.2.1
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-server:7.2.1
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "9094:9094"
      - "9092:9092"
      - "9101:9101"
    environment:
      HOSTNAME: "kafka"
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT,CLUSTER:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: INSIDE://:9092,OUTSIDE://kafka:9094,CLUSTER://kafka:1337
      KAFKA_LISTENERS: INSIDE://:9092,OUTSIDE://:9094,CLUSTER://:1337
      KAFKA_INTER_BROKER_LISTENER_NAME: INSIDE
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
      KAFKA_METRIC_REPORTERS: io.confluent.metrics.reporter.ConfluentMetricsReporter
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_CONFLUENT_BALANCER_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_JMX_PORT: 9101
      KAFKA_JMX_HOSTNAME: localhost
      KAFKA_CONFLUENT_SCHEMA_REGISTRY_URL: http://schema-registry:8081
      CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS: kafka:1337
      CONFLUENT_METRICS_REPORTER_TOPIC_REPLICAS: 1
      CONFLUENT_METRICS_ENABLE: 'true'
      CONFLUENT_SUPPORT_CUSTOMER_ID: 'anonymous'

  schema-registry:
    image: confluentinc/cp-schema-registry:7.2.1
    container_name: schema-registry
    ports:
      - "8081:8081"
    depends_on:
      - kafka
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'kafka:1337'
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081

  kafka-ui:
    image: provectuslabs/kafka-ui:v0.4.0  # Avro preview broken in v0.5.0
    container_name: kafka-ui
    ports:
      - "8080:8080"
    depends_on:
      - kafka
      - schema-registry
    restart: always
    environment:
      KAFKA_CLUSTERS_0_NAME: BG
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9094
      KAFKA_CLUSTERS_0_ZOOKEEPER: zookeeper:2181
      KAFKA_CLUSTERS_0_SCHEMAREGISTRY: 'http://schema-registry:8081'
      KAFKA_CLUSTERS_0_KAFKACONNECT_0_NAME: kafka-connect
      KAFKA_CLUSTERS_0_KAFKACONNECT_0_ADDRESS: "http://kafka-connect:8083"

networks:
  default:
    external: true
    name: kafka_overlay_network
```

**Additional context**
Since new "Value Serde" options are implemented in v0.5.0, it's quite clear that some bug was introduced.

Issue comments:
Hello there th0ger! 👋

Thank you and congratulations 🎉 for opening your very first issue in this project! 💖

In case you want to claim this issue, please comment down below! We will try to get back to you as soon as we can. 👀
@th0ger hey, is the schema for this topic located in SR?
> @th0ger hey, is the schema for this topic located in SR?

Yes
> > @th0ger hey, is the schema for this topic located in SR?
> 
> Yes

do you have schema registry available in serdes list within this topic? If yes, does selecting one help?
@th0ger also, please tell us if you are using TopicNameStrategy for subjects naming
If you mean this dropdown, then no schema:
![image](https://user-images.githubusercontent.com/26322144/212295098-e3ce3292-3e2f-424a-821c-76efc5e367e5.png)

We use TopicRecordNameStrategy, essentially something like:
```python
from confluent_kafka.schema_registry import topic_record_subject_name_strategy
conf = {'auto.register.schemas': True, 'use.latest.version': False, 'subject.name.strategy': topic_record_subject_name_strategy}
AvroSerializer(schema_registry_client=registry, schema_str=schema_str, conf=conf)
```

I noticed under Settings that 
```confluent.value.subject.name.strategy: io.confluent.kafka.serializers.subject.TopicNameStrategy```,
not sure where it comes from. We intend to use TopicRecordNameStrategy.

UPDATE: I managed to change above setting to `io.confluent.kafka.serializers.subject.TopicRecordNameStrategy` (via Confluent Control Center) but the issue persists.
@th0ger thank you for response, we will fix it soon
Hello @th0ger, 
we implemented fix on this issue - can you please check on `master` tag image?
Got no reply, closing as completed.
@Haarolean yea it's on my list
@th0ger okay, ping me if it doesn't help
@armenuikafka please check the test cases for that issue and add one, if missing.

Pull request review comments:

Patch that resolved the issue:
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/serdes/builtin/sr/SchemaRegistrySerde.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/serdes/builtin/sr/SchemaRegistrySerde.java
index 9c3d0a9d375..a4d7ee8891d 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/serdes/builtin/sr/SchemaRegistrySerde.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/serdes/builtin/sr/SchemaRegistrySerde.java
@@ -46,6 +46,7 @@ public static String name() {
   private List<String> schemaRegistryUrls;
   private String valueSchemaNameTemplate;
   private String keySchemaNameTemplate;
+  private boolean checkSchemaExistenceForDeserialize;
 
   private Map<SchemaType, MessageFormatter> schemaRegistryFormatters;
 
@@ -75,7 +76,9 @@ public void autoConfigure(PropertyResolver kafkaClusterProperties,
             kafkaClusterProperties.getProperty("schemaRegistrySSL.truststorePassword", String.class).orElse(null)
         ),
         kafkaClusterProperties.getProperty("schemaRegistryKeySchemaNameTemplate", String.class).orElse("%s-key"),
-        kafkaClusterProperties.getProperty("schemaRegistrySchemaNameTemplate", String.class).orElse("%s-value")
+        kafkaClusterProperties.getProperty("schemaRegistrySchemaNameTemplate", String.class).orElse("%s-value"),
+        kafkaClusterProperties.getProperty("schemaRegistryCheckSchemaExistenceForDeserialize", Boolean.class)
+            .orElse(false)
     );
   }
 
@@ -99,7 +102,9 @@ public void configure(PropertyResolver serdeProperties,
             serdeProperties.getProperty("truststorePassword", String.class).orElse(null)
         ),
         serdeProperties.getProperty("keySchemaNameTemplate", String.class).orElse("%s-key"),
-        serdeProperties.getProperty("schemaNameTemplate", String.class).orElse("%s-value")
+        serdeProperties.getProperty("schemaNameTemplate", String.class).orElse("%s-value"),
+        kafkaClusterProperties.getProperty("checkSchemaExistenceForDeserialize", Boolean.class)
+            .orElse(false)
     );
   }
 
@@ -108,12 +113,14 @@ void configure(
       List<String> schemaRegistryUrls,
       SchemaRegistryClient schemaRegistryClient,
       String keySchemaNameTemplate,
-      String valueSchemaNameTemplate) {
+      String valueSchemaNameTemplate,
+      boolean checkTopicSchemaExistenceForDeserialize) {
     this.schemaRegistryUrls = schemaRegistryUrls;
     this.schemaRegistryClient = schemaRegistryClient;
     this.keySchemaNameTemplate = keySchemaNameTemplate;
     this.valueSchemaNameTemplate = valueSchemaNameTemplate;
     this.schemaRegistryFormatters = MessageFormatter.createMap(schemaRegistryClient);
+    this.checkSchemaExistenceForDeserialize = checkTopicSchemaExistenceForDeserialize;
   }
 
   private static SchemaRegistryClient createSchemaRegistryClient(List<String> urls,
@@ -122,8 +129,7 @@ private static SchemaRegistryClient createSchemaRegistryClient(List<String> urls
                                                                  @Nullable String keyStoreLocation,
                                                                  @Nullable String keyStorePassword,
                                                                  @Nullable String trustStoreLocation,
-                                                                 @Nullable String trustStorePassword
-                                                                 ) {
+                                                                 @Nullable String trustStorePassword) {
     Map<String, String> configs = new HashMap<>();
     if (username != null && password != null) {
       configs.put(BASIC_AUTH_CREDENTIALS_SOURCE, "USER_INFO");
@@ -169,7 +175,8 @@ public Optional<String> getDescription() {
   @Override
   public boolean canDeserialize(String topic, Target type) {
     String subject = schemaSubject(topic, type);
-    return getSchemaBySubject(subject).isPresent();
+    return !checkSchemaExistenceForDeserialize
+        || getSchemaBySubject(subject).isPresent();
   }
 
   @Override


Test patch:
diff --git a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/serdes/builtin/sr/SchemaRegistrySerdeTest.java b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/serdes/builtin/sr/SchemaRegistrySerdeTest.java
index 092b3e8b3c9..4ea2bf3c2ac 100644
--- a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/serdes/builtin/sr/SchemaRegistrySerdeTest.java
+++ b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/serdes/builtin/sr/SchemaRegistrySerdeTest.java
@@ -22,6 +22,7 @@
 import org.apache.avro.io.Encoder;
 import org.apache.avro.io.EncoderFactory;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.CsvSource;
@@ -35,7 +36,7 @@ class SchemaRegistrySerdeTest {
   @BeforeEach
   void init() {
     serde = new SchemaRegistrySerde();
-    serde.configure(List.of("wontbeused"), registryClient, "%s-key", "%s-value");
+    serde.configure(List.of("wontbeused"), registryClient, "%s-key", "%s-value", true);
   }
 
   @ParameterizedTest
@@ -129,24 +130,61 @@ void deserializeReturnsJsonAvroMsgJsonRepresentation() throws RestClientExceptio
         .contains(Map.entry("schemaId", schemaId));
   }
 
+  @Nested
+  class SerdeWithDisabledSubjectExistenceCheck {
+
+    @BeforeEach
+    void init() {
+      serde.configure(List.of("wontbeused"), registryClient, "%s-key", "%s-value", false);
+    }
+
+    @Test
+    void canDeserializeAlwaysReturnsTrue() {
+      String topic = RandomString.make(10);
+      assertThat(serde.canDeserialize(topic, Serde.Target.KEY)).isTrue();
+      assertThat(serde.canDeserialize(topic, Serde.Target.VALUE)).isTrue();
+    }
+  }
+
+  @Nested
+  class SerdeWithEnabledSubjectExistenceCheck {
+
+    @BeforeEach
+    void init() {
+      serde.configure(List.of("wontbeused"), registryClient, "%s-key", "%s-value", true);
+    }
+
+    @Test
+    void canDeserializeReturnsTrueIfSubjectExists() throws Exception {
+      String topic = RandomString.make(10);
+      registryClient.register(topic + "-key", new AvroSchema("\"int\""));
+      registryClient.register(topic + "-value", new AvroSchema("\"int\""));
+
+      assertThat(serde.canDeserialize(topic, Serde.Target.KEY)).isTrue();
+      assertThat(serde.canDeserialize(topic, Serde.Target.VALUE)).isTrue();
+    }
+
+    @Test
+    void canDeserializeReturnsFalseIfSubjectDoesNotExist() {
+      String topic = RandomString.make(10);
+      assertThat(serde.canDeserialize(topic, Serde.Target.KEY)).isFalse();
+      assertThat(serde.canDeserialize(topic, Serde.Target.VALUE)).isFalse();
+    }
+  }
+
   @Test
   void canDeserializeAndCanSerializeReturnsTrueIfSubjectExists() throws Exception {
     String topic = RandomString.make(10);
     registryClient.register(topic + "-key", new AvroSchema("\"int\""));
     registryClient.register(topic + "-value", new AvroSchema("\"int\""));
 
-    assertThat(serde.canDeserialize(topic, Serde.Target.KEY)).isTrue();
-    assertThat(serde.canDeserialize(topic, Serde.Target.VALUE)).isTrue();
-
     assertThat(serde.canSerialize(topic, Serde.Target.KEY)).isTrue();
     assertThat(serde.canSerialize(topic, Serde.Target.VALUE)).isTrue();
   }
 
   @Test
-  void canDeserializeAndCanSerializeReturnsFalseIfSubjectDoesNotExist() {
+  void canSerializeReturnsFalseIfSubjectDoesNotExist() {
     String topic = RandomString.make(10);
-    assertThat(serde.canDeserialize(topic, Serde.Target.KEY)).isFalse();
-    assertThat(serde.canDeserialize(topic, Serde.Target.VALUE)).isFalse();
     assertThat(serde.canSerialize(topic, Serde.Target.KEY)).isFalse();
     assertThat(serde.canSerialize(topic, Serde.Target.VALUE)).isFalse();
   }
@@ -178,4 +216,4 @@ private byte[] jsonToAvro(String json, AvroSchema schema) {
     return output.toByteArray();
   }
 
-}
\ No newline at end of file
+}