Allow producing messages with an empty key/value

* [ISSUE 1046] UI allows to submit message with empty key & value (#1264)

* [ISSUE 1046] UI allows to submit message with empty key & value

* Update Contract

(cherry picked from commit 4b730eb)

* Backend fix

* Refactoring

* Fix nullable & checkstyle

* Fix jsonnullable get

* Remove unnecessary check and add a test

Co-authored-by: Oleg Shur <[email protected]>

Author: Haarolean

kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/Config.java

@@ -1,5 +1,6 @@
 package com.provectus.kafka.ui.config;
 
+import com.fasterxml.jackson.databind.Module;
 import com.provectus.kafka.ui.model.JmxConnectionInfo;
 import com.provectus.kafka.ui.util.JmxPoolFactory;
 import java.util.Collections;
@@ -9,6 +10,7 @@
 import org.apache.commons.pool2.KeyedObjectPool;
 import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
 import org.apache.commons.pool2.impl.GenericKeyedObjectPoolConfig;
+import org.openapitools.jackson.nullable.JsonNullableModule;
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.web.ServerProperties;
@@ -78,4 +80,9 @@ public WebClient webClient(
         .codecs(c -> c.defaultCodecs().maxInMemorySize((int) maxBuffSize.toBytes()))
         .build();
   }
+
+  @Bean
+  public JsonNullableModule jsonNullableModule() {
+    return new JsonNullableModule();
+  }
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/serde/DeserializationService.java

@@ -39,7 +39,7 @@ private RecordSerDe createRecordDeserializerForCluster(KafkaCluster cluster) {
             objectMapper);
       } else {
         log.info("Using SchemaRegistryAwareRecordSerDe for cluster '{}'", cluster.getName());
-        return new SchemaRegistryAwareRecordSerDe(cluster);
+        return new SchemaRegistryAwareRecordSerDe(cluster, objectMapper);
       }
     } catch (Throwable e) {
       throw new RuntimeException("Can't init deserializer", e);

kafka-ui-api/src/main/java/com/provectus/kafka/ui/serde/SimpleRecordSerDe.java

@@ -13,6 +13,8 @@
 
 public class SimpleRecordSerDe implements RecordSerDe {
 
+  private static final ObjectMapper objectMapper = new ObjectMapper();
+
   @Override
   public DeserializedKeyValue deserialize(ConsumerRecord<Bytes, Bytes> msg) {
     var builder = DeserializedKeyValue.builder();
@@ -45,7 +47,7 @@ public TopicMessageSchemaDTO getTopicSchema(String topic) {
     final MessageSchemaDTO schema = new MessageSchemaDTO()
         .name("unknown")
         .source(MessageSchemaDTO.SourceEnum.UNKNOWN)
-        .schema(JsonSchema.stringSchema().toJson(new ObjectMapper()));
+        .schema(JsonSchema.stringSchema().toJson(objectMapper));
     return new TopicMessageSchemaDTO()
         .key(schema)
         .value(schema);

kafka-ui-api/src/main/java/com/provectus/kafka/ui/serde/schemaregistry/SchemaRegistryAwareRecordSerDe.java

@@ -10,7 +10,6 @@
 import com.provectus.kafka.ui.model.MessageSchemaDTO;
 import com.provectus.kafka.ui.model.TopicMessageSchemaDTO;
 import com.provectus.kafka.ui.serde.RecordSerDe;
-import com.provectus.kafka.ui.util.ConsumerRecordUtil;
 import com.provectus.kafka.ui.util.jsonschema.AvroJsonSchemaConverter;
 import com.provectus.kafka.ui.util.jsonschema.JsonSchema;
 import com.provectus.kafka.ui.util.jsonschema.ProtobufSchemaConverter;
@@ -27,7 +26,6 @@
 import io.confluent.kafka.schemaregistry.protobuf.ProtobufSchemaProvider;
 import java.net.URI;
 import java.nio.ByteBuffer;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -47,31 +45,32 @@ public class SchemaRegistryAwareRecordSerDe implements RecordSerDe {
 
   private static final int CLIENT_IDENTITY_MAP_CAPACITY = 100;
 
+  private static final StringMessageFormatter stringFormatter = new StringMessageFormatter();
+  private static final ProtobufSchemaConverter protoSchemaConverter = new ProtobufSchemaConverter();
+  private static final AvroJsonSchemaConverter avroSchemaConverter = new AvroJsonSchemaConverter();
+
   private final KafkaCluster cluster;
   private final Map<String, MessageFormatter> valueFormatMap = new ConcurrentHashMap<>();
   private final Map<String, MessageFormatter> keyFormatMap = new ConcurrentHashMap<>();
 
   @Nullable
   private final SchemaRegistryClient schemaRegistryClient;
-
   @Nullable
   private final AvroMessageFormatter avroFormatter;
-
   @Nullable
   private final ProtobufMessageFormatter protobufFormatter;
-
   @Nullable
   private final JsonSchemaMessageFormatter jsonSchemaMessageFormatter;
 
-  private final StringMessageFormatter stringFormatter = new StringMessageFormatter();
-  private final ProtobufSchemaConverter protoSchemaConverter = new ProtobufSchemaConverter();
-  private final AvroJsonSchemaConverter avroSchemaConverter = new AvroJsonSchemaConverter();
-  private final ObjectMapper objectMapper = new ObjectMapper();
+  private ObjectMapper objectMapper;
 
-  private static SchemaRegistryClient createSchemaRegistryClient(KafkaCluster cluster) {
+  private SchemaRegistryClient createSchemaRegistryClient(KafkaCluster cluster,
+                                                          ObjectMapper objectMapper) {
     if (cluster.getSchemaRegistry() == null) {
       throw new ValidationException("schemaRegistry is not specified");
     }
+    this.objectMapper = objectMapper;
+
     List<SchemaProvider> schemaProviders =
         List.of(new AvroSchemaProvider(), new ProtobufSchemaProvider(), new JsonSchemaProvider());
 
@@ -97,10 +96,10 @@ private static SchemaRegistryClient createSchemaRegistryClient(KafkaCluster clus
     );
   }
 
-  public SchemaRegistryAwareRecordSerDe(KafkaCluster cluster) {
+  public SchemaRegistryAwareRecordSerDe(KafkaCluster cluster, ObjectMapper objectMapper) {
     this.cluster = cluster;
     this.schemaRegistryClient = cluster.getSchemaRegistry() != null
-        ? createSchemaRegistryClient(cluster)
+        ? createSchemaRegistryClient(cluster, objectMapper)
         : null;
     if (schemaRegistryClient != null) {
       this.avroFormatter = new AvroMessageFormatter(schemaRegistryClient);
@@ -147,41 +146,45 @@ public ProducerRecord<byte[], byte[]> serialize(String topic,
                                                   @Nullable String key,
                                                   @Nullable String data,
                                                   @Nullable Integer partition) {
-    final Optional<SchemaMetadata> maybeValueSchema = getSchemaBySubject(topic, false);
     final Optional<SchemaMetadata> maybeKeySchema = getSchemaBySubject(topic, true);
+    final Optional<SchemaMetadata> maybeValueSchema = getSchemaBySubject(topic, false);
 
-    final byte[] serializedValue = data != null
-        ? serialize(maybeValueSchema, topic, data, false)
-        : null;
-    final byte[] serializedKey = key != null
-        ? serialize(maybeKeySchema, topic, key, true)
-        : null;
+    final byte[] serializedKey = maybeKeySchema.isPresent()
+        ? serialize(maybeKeySchema.get(), topic, key, true)
+        : serialize(key);
+
+    final byte[] serializedValue = maybeValueSchema.isPresent()
+        ? serialize(maybeValueSchema.get(), topic, data, false)
+        : serialize(data);
 
     return new ProducerRecord<>(topic, partition, serializedKey, serializedValue);
   }
 
   @SneakyThrows
-  private byte[] serialize(
-      Optional<SchemaMetadata> maybeSchema, String topic, String value, boolean isKey) {
-    if (maybeSchema.isPresent()) {
-      final SchemaMetadata schema = maybeSchema.get();
-
-      MessageReader<?> reader;
-      if (schema.getSchemaType().equals(MessageFormat.PROTOBUF.name())) {
-        reader = new ProtobufMessageReader(topic, isKey, schemaRegistryClient, schema);
-      } else if (schema.getSchemaType().equals(MessageFormat.AVRO.name())) {
-        reader = new AvroMessageReader(topic, isKey, schemaRegistryClient, schema);
-      } else if (schema.getSchemaType().equals(MessageFormat.JSON.name())) {
-        reader = new JsonSchemaMessageReader(topic, isKey, schemaRegistryClient, schema);
-      } else {
-        throw new IllegalStateException("Unsupported schema type: " + schema.getSchemaType());
-      }
-
-      return reader.read(value);
+  private byte[] serialize(SchemaMetadata schema, String topic, String value, boolean isKey) {
+    if (value == null) {
+      return null;
+    }
+    MessageReader<?> reader;
+    if (schema.getSchemaType().equals(MessageFormat.PROTOBUF.name())) {
+      reader = new ProtobufMessageReader(topic, isKey, schemaRegistryClient, schema);
+    } else if (schema.getSchemaType().equals(MessageFormat.AVRO.name())) {
+      reader = new AvroMessageReader(topic, isKey, schemaRegistryClient, schema);
+    } else if (schema.getSchemaType().equals(MessageFormat.JSON.name())) {
+      reader = new JsonSchemaMessageReader(topic, isKey, schemaRegistryClient, schema);
     } else {
-      // if no schema provided serialize input as raw string
-      return value.getBytes();
+      throw new IllegalStateException("Unsupported schema type: " + schema.getSchemaType());
+    }
+
+    return reader.read(value);
+  }
+
+  private byte[] serialize(String value) {
+    if (value == null) {
+      return null;
     }
+    // if no schema provided serialize input as raw string
+    return value.getBytes();
   }
 
   @Override

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/MessagesService.java

@@ -81,9 +81,6 @@ private Mono<Map<TopicPartition, Long>> offsetsForDeletion(KafkaCluster cluster,
 
   public Mono<RecordMetadata> sendMessage(KafkaCluster cluster, String topic,
                                           CreateTopicMessageDTO msg) {
-    if (msg.getKey() == null && msg.getContent() == null) {
-      throw new ValidationException("Invalid message: both key and value can't be null");
-    }
     if (msg.getPartition() != null
         && msg.getPartition() > metricsCache.get(cluster).getTopicDescriptions()
           .get(topic).partitions().size() - 1) {
@@ -100,8 +97,8 @@ public Mono<RecordMetadata> sendMessage(KafkaCluster cluster, String topic,
     try (KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(properties)) {
       ProducerRecord<byte[], byte[]> producerRecord = serde.serialize(
           topic,
-          msg.getKey(),
-          msg.getContent(),
+          msg.getKey().orElse(null),
+          msg.getContent().orElse(null),
           msg.getPartition()
       );
       producerRecord = new ProducerRecord<>(

kafka-ui-api/src/test/java/com/provectus/kafka/ui/serde/SchemaRegistryRecordDeserializerTest.java

@@ -17,7 +17,7 @@ class SchemaRegistryRecordDeserializerTest {
       new SchemaRegistryAwareRecordSerDe(
           KafkaCluster.builder()
               .schemaNameTemplate("%s-value")
-              .build()
+              .build(), new ObjectMapper()
       );
 
   @Test

kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/SendAndReadTests.java

@@ -465,6 +465,20 @@ void topicMessageMetadataJson() {
         });
   }
 
+  @Test
+  void noKeyAndNoContentPresentTest() {
+    new SendAndReadSpec()
+        .withMsgToSend(
+            new CreateTopicMessageDTO()
+                .key(null)
+                .content(null)
+        )
+        .doAssert(polled -> {
+          assertThat(polled.getKey()).isNull();
+          assertThat(polled.getContent()).isNull();
+        });
+  }
+
   @SneakyThrows
   private void assertJsonEqual(String actual, String expected) {
     var mapper = new ObjectMapper();

kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml

@@ -1870,12 +1870,16 @@ components:
           type: integer
         key:
           type: string
+          nullable: true
         headers:
           type: object
           additionalProperties:
             type: string
         content:
           type: string
+          nullable: true
+      required:
+        - partition
 
     TopicMessageSchema:
       type: object
@@ -2635,4 +2639,4 @@ components:
         - DELETE
         - COMPACT
         - COMPACT_DELETE
-        - UNKNOWN
+        - UNKNOWN

kafka-ui-react-app/src/components/Alerts/Alert.styled.ts

@@ -3,7 +3,7 @@ import styled from 'styled-components';
 
 export const Alert = styled.div<{ $type: AlertType }>`
   background-color: ${({ $type, theme }) => theme.alert.color[$type]};
-  width: 400px;
+  min-width: 400px;
   min-height: 64px;
   border-radius: 8px;
   padding: 12px;
@@ -20,8 +20,14 @@ export const Title = styled.div`
   font-size: 14px;
 `;
 
-export const Message = styled.p`
+export const Message = styled.div`
   font-weight: normal;
   font-size: 14px;
   margin: 3px 0;
+
+  ol,
+  ul {
+    padding-left: 25px;
+    list-style: auto;
+  }
 `;

kafka-ui-react-app/src/components/App.styled.ts

@@ -168,7 +168,7 @@ export const AlertsContainer = styled.div`
   width: 500px;
   position: fixed;
   bottom: 15px;
-  left: 15px;
+  right: 15px;
   z-index: 1000;
 
   @media screen and (max-width: 1023px) {