Documentation for pluggable serdes (#2702)

* 1. Minor serde's code improvements
2. Javadoc added to serde-related code

* Serdes configuration docs WIP

* wip

* DeserializeResult.toString impl added

* SchemeRegistry.canDeserialize checks subject existence

* Update Serialization.md

* Update README.md

Co-authored-by: iliax <ikuramshin@provectus.com>
Co-authored-by: Roman Zabaluev <rzabaluev@provectus.com>

README.md

@@ -30,6 +30,7 @@ the cloud.
 * **Browse Messages** — browse messages with JSON, plain text, and Avro encoding
 * **Dynamic Topic Configuration** — create and configure new topics with dynamic configuration
 * **Configurable Authentification** — secure your installation with optional Github/Gitlab/Google OAuth 2.0
+* **Custom serialization/deserialization plugins** - use a ready-to-go serde for your data like AWS Glue or Smile, or code your own!
 
 # The Interface
 UI for Apache Kafka wraps major functions of Apache Kafka with an intuitive user interface.
@@ -76,6 +77,7 @@ We have plenty of [docker-compose files](documentation/compose/DOCKER_COMPOSE.md
 - [AWS IAM configuration](documentation/guides/AWS_IAM.md)
 - [Docker-compose files](documentation/compose/DOCKER_COMPOSE.md)
 - [Connection to a secure broker](documentation/guides/SECURE_BROKER.md)
+- [Configure seriliazation/deserialization plugins or code your own](documentation/guides/Serialization.md)
 
 ### Configuration File
 Example of how to configure clusters in the [application-local.yml](https://github.com/provectus/kafka-ui/blob/master/kafka-ui-api/src/main/resources/application-local.yml) configuration file:

documentation/guides/Serialization.md

@@ -0,0 +1,169 @@
+## Serialization and deserialization and custom plugins
+
+Kafka-ui supports multiple ways to serialize/deserialize data.
+
+
+### Int32, Int64, UInt32, UInt64
+Big-endian 4/8 bytes representation of signed/unsigned integers.
+
+### Base64
+Base64 (RFC4648) binary data representation. Can be useful in case if the actual data is not important, but exactly the same (byte-wise) key/value should be send.
+
+### String 
+Treats binary data as a string in specified encoding. Default encoding is UTF-8.
+
+Class name: `com.provectus.kafka.ui.serdes.builtin.StringSerde`
+
+Sample configuration (if you want to overwrite default configuration):
+```yaml
+kafka:
+  clusters:
+    - name: Cluster1
+      # Other Cluster configuration omitted ... 
+      serdes:
+          # registering String serde with custom config
+        - name: AsciiString
+          className: com.provectus.kafka.ui.serdes.builtin.StringSerde
+          properties:
+            encoding: "ASCII"
+        
+          # overriding build-it String serde config   
+        - name: String 
+          properties:
+            encoding: "UTF-16"
+```
+
+### Protobuf
+[Deprecated configuration is here](Protobuf.md)
+
+Class name: `com.provectus.kafka.ui.serdes.builtin.ProtobufFileSerde`
+
+Sample configuration:
+```yaml
+kafka:
+  clusters:
+    - name: Cluster1
+      # Other Cluster configuration omitted ... 
+      serdes:
+        - name: ProtobufFile
+          properties:
+            # path to the protobuf schema file
+            protobufFile: path/to/my.proto
+            # default protobuf type that is used for KEY serialization/deserialization
+            # optional
+            protobufMessageNameForKey: my.Type1
+            # mapping of topic names to protobuf types, that will be used for KEYS  serialization/deserialization
+            # optional
+            protobufMessageNameForKeyByTopic:
+              topic1: my.KeyType1
+              topic2: my.KeyType2
+            # default protobuf type that is used for VALUE serialization/deserialization
+            # optional, if not set - first type in file will be used as default
+            protobufMessageName: my.Type1
+            # mapping of topic names to protobuf types, that will be used for VALUES  serialization/deserialization
+            # optional
+            protobufMessageNameByTopic:
+              topic1: my.Type1
+              topic2: my.Type2
+```
+
+### SchemaRegistry
+SchemaRegistry serde is automatically configured if schema registry properties set on cluster level.
+But you can override them or add new SchemaRegistry serde that will connect to another schema-registry instance. 
+
+Class name: `com.provectus.kafka.ui.serdes.builtin.sr.SchemaRegistrySerde`
+
+Sample configuration:
+```yaml
+kafka:
+  clusters:
+    - name: Cluster1
+      # this schema-registry will be used by SchemaRegistrySerde by default
+      schemaRegistry: http://main-schema-registry:8081
+      serdes:
+        - name: SchemaRegistry
+          properties:
+            # but you can override cluster-level properties
+            url:  http://another-schema-registry:8081
+            # auth properties, optional
+            username: nameForAuth
+            password: P@ssW0RdForAuth
+        
+          # and also add another SchemaRegistry serde
+        - name: AnotherOneSchemaRegistry
+          className: com.provectus.kafka.ui.serdes.builtin.sr.SchemaRegistrySerde
+          properties:
+            url:  http://another-yet-schema-registry:8081
+            username: nameForAuth_2
+            password: P@ssW0RdForAuth_2
+```
+
+## Setting serdes for specific topics
+You can specify preferable serde for topics key/value. This serde will be chosen by default in UI on topic's view/produce pages. 
+To do so, set `topicValuesPattern/topicValuesPattern` properties for the selected serde. Kafka-ui will choose a first serde that matches specified pattern.
+
+Sample configuration:
+```yaml
+kafka:
+  clusters:
+    - name: Cluster1
+      serdes:
+        - name: String
+          topicKeysPattern: click-events|imp-events
+        
+        - name: Int64
+          topicKeysPattern: ".*-events"
+        
+        - name: SchemaRegistry
+          properties:
+            url:  http://schema-registry:8081
+          topicValuesPattern: click-events|imp-events
+```
+
+
+## Default serdes
+You can specify which serde will be chosen in UI by default if no other serdes selected via `topicKeysPattern/topicValuesPattern` settings.
+
+Sample configuration:
+```yaml
+kafka:
+  clusters:
+    - name: Cluster1
+      defaultKeySerde: Int34
+      defaultValueSerde: String
+      serdes:
+        - name: Int32
+          topicKeysPattern: click-events|imp-events
+```
+
+## Fallback
+If selected serde couldn't be applied (exception was thrown), then fallback (String serde with UTF-8 encoding) serde will be applied. Such messages will be specially highlighted in UI.
+
+## Custom pluggable serde registration
+You can implement your own serde and register it in kafka-ui application.
+To do so:
+1. Add `kafka-ui-serde-api` dependency (should be downloadable via maven central)
+2. Implement `com.provectus.kafka.ui.serde.api.Serde` interface. See javadoc for implementation requirements.
+3. Pack your serde into uber jar, or provide directory with no-dependency jar and it's dependencies jars
+
+
+Example pluggable serdes :
+https://github.com/provectus/kafkaui-smile-serde
+https://github.com/provectus/kafkaui-glue-sr-serde
+
+Sample configuration:
+```yaml
+kafka:
+  clusters:
+    - name: Cluster1
+      serdes:
+        - name: MyCustomSerde
+          className: my.lovely.org.KafkaUiSerde
+          filePath: /var/lib/kui-serde/my-kui-serde.jar
+          
+        - name: MyCustomSerde2
+          className: my.lovely.org.KafkaUiSerde2
+          filePath: /var/lib/kui-serde2
+          properties:
+            prop1: v1
+```

kafka-ui-api/src/main/java/com/provectus/kafka/ui/serdes/ClusterSerdes.java

@@ -1,6 +1,8 @@
 package com.provectus.kafka.ui.serdes;
 
 import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+import com.google.common.collect.ImmutableMap;
 import com.provectus.kafka.ui.config.ClustersProperties;
 import com.provectus.kafka.ui.exception.ValidationException;
 import com.provectus.kafka.ui.serde.api.PropertyResolver;
@@ -14,6 +16,7 @@ import com.provectus.kafka.ui.serdes.builtin.UInt32Serde;
 import com.provectus.kafka.ui.serdes.builtin.UInt64Serde;
 import com.provectus.kafka.ui.serdes.builtin.UuidBinarySerde;
 import com.provectus.kafka.ui.serdes.builtin.sr.SchemaRegistrySerde;
+import java.io.Closeable;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Optional;
@@ -26,22 +29,22 @@ import lombok.extern.slf4j.Slf4j;
 import org.springframework.core.env.Environment;
 
 @Slf4j
-public class ClusterSerdes {
+public class ClusterSerdes implements Closeable {
 
   private static final CustomSerdeLoader CUSTOM_SERDE_LOADER = new CustomSerdeLoader();
 
   private static final Map<String, Class<? extends BuiltInSerde>> BUILT_IN_SERDES =
-      Map.of(
-          StringSerde.name(), StringSerde.class,
-          Int32Serde.name(), Int32Serde.class,
-          Int64Serde.name(), Int64Serde.class,
-          UInt32Serde.name(), UInt32Serde.class,
-          UInt64Serde.name(), UInt64Serde.class,
-          UuidBinarySerde.name(), UuidBinarySerde.class,
-          Base64Serde.name(), Base64Serde.class,
-          SchemaRegistrySerde.name(), SchemaRegistrySerde.class,
-          ProtobufFileSerde.name(), ProtobufFileSerde.class
-      );
+      ImmutableMap.<String, Class<? extends BuiltInSerde>>builder()
+          .put(StringSerde.name(), StringSerde.class)
+          .put(Int32Serde.name(), Int32Serde.class)
+          .put(Int64Serde.name(), Int64Serde.class)
+          .put(UInt32Serde.name(), UInt32Serde.class)
+          .put(UInt64Serde.name(), UInt64Serde.class)
+          .put(Base64Serde.name(), Base64Serde.class)
+          .put(SchemaRegistrySerde.name(), SchemaRegistrySerde.class)
+          .put(ProtobufFileSerde.name(), ProtobufFileSerde.class)
+          .put(UuidBinarySerde.name(), UuidBinarySerde.class)
+          .build();
 
   // using linked map to keep order of serdes added to it
   private final Map<String, SerdeInstance> serdes = new LinkedHashMap<>();
@@ -64,6 +67,9 @@ public class ClusterSerdes {
     ClustersProperties.Cluster clusterProp = clustersProperties.getClusters().get(clusterIndex);
     for (int i = 0; i < clusterProp.getSerde().size(); i++) {
       var sendeConf = clusterProp.getSerde().get(i);
+      if (Strings.isNullOrEmpty(sendeConf.getName())) {
+        throw new ValidationException("'name' property not set for serde: " + sendeConf);
+      }
       if (serdes.containsKey(sendeConf.getName())) {
         throw new ValidationException("Multiple serdes with same name: " + sendeConf.getName());
       }
@@ -154,6 +160,14 @@ public class ClusterSerdes {
                                    PropertyResolver serdeProps,
                                    PropertyResolver clusterProps,
                                    PropertyResolver globalProps) {
+    if (Strings.isNullOrEmpty(serdeConfig.getClassName())) {
+      throw new ValidationException(
+          "'className' property not set for custom serde " + serdeConfig.getName());
+    }
+    if (Strings.isNullOrEmpty(serdeConfig.getFilePath())) {
+      throw new ValidationException(
+          "'filePath' property not set for custom serde " + serdeConfig.getName());
+    }
     var loaded = CUSTOM_SERDE_LOADER.loadAndConfigure(
         serdeConfig.getClassName(), serdeConfig.getFilePath(), serdeProps, clusterProps, globalProps);
     return new SerdeInstance(
@@ -215,4 +229,8 @@ public class ClusterSerdes {
         .orElse(serdes.get(StringSerde.name()));
   }
 
+  @Override
+  public void close() {
+    serdes.values().forEach(SerdeInstance::close);
+  }
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/serdes/ProducerRecordCreator.java

@@ -25,14 +25,11 @@ public class ProducerRecordCreator {
         partition,
         key == null ? null : keySerializer.serialize(key),
         value == null ? null : valuesSerializer.serialize(value),
-        createHeaders(headers)
+        headers == null ? null : createHeaders(headers)
     );
   }
 
-  private Iterable<Header> createHeaders(@Nullable Map<String, String> clientHeaders) {
-    if (clientHeaders == null) {
-      return new RecordHeaders();
-    }
+  private Iterable<Header> createHeaders(Map<String, String> clientHeaders) {
     RecordHeaders headers = new RecordHeaders();
     clientHeaders.forEach((k, v) -> headers.add(new RecordHeader(k, v.getBytes())));
     return headers;

kafka-ui-api/src/main/java/com/provectus/kafka/ui/serdes/SerdeInstance.java

@@ -2,16 +2,18 @@ package com.provectus.kafka.ui.serdes;
 
 import com.provectus.kafka.ui.serde.api.SchemaDescription;
 import com.provectus.kafka.ui.serde.api.Serde;
+import java.io.Closeable;
 import java.util.Optional;
 import java.util.function.Supplier;
 import java.util.regex.Pattern;
 import javax.annotation.Nullable;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 
-
+@Slf4j
 @RequiredArgsConstructor
-public class SerdeInstance {
+public class SerdeInstance implements Closeable {
 
   @Getter
   private final String name;
@@ -68,4 +70,16 @@ public class SerdeInstance {
       return (headers, data) -> wrapWithClassloader(() -> deserializer.deserialize(headers, data));
     });
   }
+
+  @Override
+  public void close() {
+    wrapWithClassloader(() -> {
+      try {
+        serde.close();
+      } catch (Exception e) {
+        log.error("Error closing serde " + name, e);
+      }
+      return null;
+    });
+  }
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/serdes/builtin/sr/SchemaRegistrySerde.java

@@ -163,7 +163,8 @@ public class SchemaRegistrySerde implements BuiltInSerde {
 
   @Override
   public boolean canDeserialize(String topic, Target type) {
-    return true;
+    String subject = schemaSubject(topic, type);
+    return getSchemaBySubject(subject).isPresent();
   }
 
   @Override

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

@@ -9,6 +9,7 @@ import com.provectus.kafka.ui.serdes.ClusterSerdes;
 import com.provectus.kafka.ui.serdes.ConsumerRecordDeserializer;
 import com.provectus.kafka.ui.serdes.ProducerRecordCreator;
 import com.provectus.kafka.ui.serdes.SerdeInstance;
+import java.io.Closeable;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -21,7 +22,7 @@ import org.springframework.stereotype.Component;
 
 @Slf4j
 @Component
-public class DeserializationService {
+public class DeserializationService implements Closeable {
 
   private final Map<KafkaCluster, ClusterSerdes> clusterSerdes = new ConcurrentHashMap<>();
 
@@ -137,4 +138,8 @@ public class DeserializationService {
         .preferred(preferred);
   }
 
+  @Override
+  public void close() {
+    clusterSerdes.values().forEach(ClusterSerdes::close);
+  }
 }

kafka-ui-api/src/test/java/com/provectus/kafka/ui/serdes/builtin/sr/SchemaRegistrySerdeTest.java

@@ -130,10 +130,25 @@ class SchemaRegistrySerdeTest {
   }
 
   @Test
-  void canDeserializeReturnsTrueAlways() {
+  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() {
+    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();
   }
 
   private void assertJsonsEqual(String expected, String actual) throws JsonProcessingException {

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

@@ -1,31 +1,54 @@
 package com.provectus.kafka.ui.serde.api;
 
+import java.util.Collections;
 import java.util.Map;
+import java.util.Objects;
 
+/**
+ * Result of {@code Deserializer} work.
+ */
 public final class DeserializeResult {
 
   public enum Type {
     STRING, JSON
   }
 
+  // nullable
   private final String result;
   private final Type type;
   private final Map<String, Object> additionalProperties;
 
+  /**
+   * @param result string representation of deserialized binary data
+   * @param type type of string - can it be converted to json or not
+   * @param additionalProperties additional information about deserialized value (will be shown in UI)
+   */
   public DeserializeResult(String result, Type type, Map<String, Object> additionalProperties) {
     this.result = result;
-    this.type = type;
-    this.additionalProperties = additionalProperties;
+    this.type = type != null ? type : Type.STRING;
+    this.additionalProperties = additionalProperties != null ? additionalProperties : Collections.emptyMap();
   }
 
+  /**
+   * @return string representation of deserialized binary data, can be null
+   */
   public String getResult() {
     return result;
   }
 
+  /**
+   * @return additional information about deserialized value.
+   * Will be show as json dictionary in UI (serialized with Jackson object mapper).
+   * It is recommended to use primitive types and strings for values.
+   */
   public Map<String, Object> getAdditionalProperties() {
     return additionalProperties;
   }
 
+  /**
+   * @return type of deserialized result. Will be used as hint for some internal logic
+   * (ex. if type==STRING smart filters won't try to parse it as json for further usage)
+   */
   public Type getType() {
     return type;
   }
@@ -38,22 +61,25 @@ public final class DeserializeResult {
     if (o == null || getClass() != o.getClass()) {
       return false;
     }
-
     DeserializeResult that = (DeserializeResult) o;
-    if (!result.equals(that.result)) {
-      return false;
-    }
-    if (type != that.type) {
-      return false;
-    }
-    return additionalProperties.equals(that.additionalProperties);
+    return Objects.equals(result, that.result)
+        && type == that.type
+        && additionalProperties.equals(that.additionalProperties);
   }
 
   @Override
   public int hashCode() {
-    int result = this.result.hashCode();
-    result = 31 * result + type.hashCode();
-    result = 31 * result + additionalProperties.hashCode();
-    return result;
+    return Objects.hash(result, type, additionalProperties);
+  }
+
+  @Override
+  public String toString() {
+    return "DeserializeResult{"
+        + "result='" + result
+        + '\''
+        + ", type=" + type
+        + ", additionalProperties="
+        + additionalProperties
+        + '}';
   }
 }

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

@@ -4,12 +4,42 @@ import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 
+/**
+ * Provides access to configuration properties.
+ *
+ * @implNote Actual implementation uses {@code org.springframework.boot.context.properties.bind.Binder} class
+ * to bind values to target types. Target type params can be custom configs classes, not only simple types and strings.
+ *
+ */
 public interface PropertyResolver {
 
+  /**
+   * Get property value by name.
+   *
+   * @param key property name
+   * @param targetType type of property value
+   * @return property value or empty {@code Optional} if property not found
+   */
   <T> Optional<T> getProperty(String key, Class<T> targetType);
 
+
+  /**
+   * Get list-property value by name
+   *
+   * @param key list property name
+   * @param itemType type of list element
+   * @return list property value or empty {@code Optional} if property not found
+   */
   <T> Optional<List<T>> getListProperty(String key, Class<T> itemType);
 
+  /**
+   * Get map-property value by name
+   *
+   * @param key  map-property name
+   * @param keyType type of map key
+   * @param valueType type of map value
+   * @return map-property value or empty {@code Optional} if property not found
+   */
   <K, V> Optional<Map<K, V>> getMapProperty(String key, Class<K> keyType, Class<V> valueType);
 
 }

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

@@ -2,21 +2,35 @@ package com.provectus.kafka.ui.serde.api;
 
 import java.util.Map;
 
+/**
+ * Description of topic's key/value schema.
+ */
 public final class SchemaDescription {
 
-  // can be json schema or plain text
   private final String schema;
   private final Map<String, Object> additionalProperties;
 
+  /**
+   *
+   * @param schema schema descriptions.
+   *              If contains json-schema (preferred) UI will use it for validation and sample data generation.
+   * @param additionalProperties additional properties about schema (may be rendered in UI in the future)
+   */
   public SchemaDescription(String schema, Map<String, Object> additionalProperties) {
     this.schema = schema;
     this.additionalProperties = additionalProperties;
   }
 
+  /**
+   * @return schema description text. Preferably contains json-schema. Can be null.
+   */
   public String getSchema() {
     return schema;
   }
 
+  /**
+   * @return additional properties about schema
+   */
   public Map<String, Object> getAdditionalProperties() {
     return additionalProperties;
   }

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

@@ -1,37 +1,106 @@
 package com.provectus.kafka.ui.serde.api;
 
+import java.io.Closeable;
 import java.util.Optional;
 
-public interface Serde {
+/**
+ * Main interface of  serialization/deserialization logic.
+ * It provides ability to serialize, deserialize topic's keys and values, and optionally provides
+ * information about data schema inside topic.
+ * <p/>
+ * <b>Lifecycle:</b><br/>
+ * 1. on application startup kafka-ui scans configs and finds all custom serde definitions<br/>
+ * 2. for each custom serde its own separated child-first classloader is created<br/>
+ * 3. kafka-ui loads class defined in configuration and instantiates instance of that class using default, non-arg constructor<br/>
+ * 4. {@code configure(...)} method called<br/>
+ * 5. various methods called during application runtime<br/>
+ * 6. on application shutdown kafka-ui calls {@code close()} method on serde instance<br/>
+ * <p/>
+ * <b>Implementation considerations:</b><br/>
+ * 1. Implementation class should have default/non-arg contructor<br/>
+ * 2. All methods except {@code configure(...)} and {@code close()} can be called from different threads. So, your code should be thread-safe.<br/>
+ * 3. All methods will be executed in separate child-first classloader.<br/>
+ */
+public interface Serde extends Closeable {
 
+  /**
+   * Kafka record's part that Serde will be applied to.
+   */
   enum Target {
     KEY, VALUE
   }
 
+  /**
+   * Reads configuration using property resolvers and sets up serde's internal state.
+   *
+   * @param serdeProperties        specific serde instance's properties
+   * @param kafkaClusterProperties properties of the custer for what serde is instantiated
+   * @param globalProperties       global application properties
+   */
   void configure(
       PropertyResolver serdeProperties,
       PropertyResolver kafkaClusterProperties,
       PropertyResolver globalProperties
   );
 
+  /**
+   * @return Serde's description. Treated as Markdown text. Will be shown in UI.
+   */
   Optional<String> getDescription();
 
+  /**
+   * @return SchemaDescription for specified topic's key/value.
+   * {@code Optional.empty} if there is not information about schema.
+   */
   Optional<SchemaDescription> getSchema(String topic, Target type);
 
+  /**
+   * @return true if this Serde can be applied to specified topic's key/value deserialization
+   */
   boolean canDeserialize(String topic, Target type);
 
+  /**
+   * @return true if this Serde can be applied to specified topic's key/value serialization
+   */
   boolean canSerialize(String topic, Target type);
 
+  /**
+   * Closes resources opened by Serde.
+   */
+  @Override
+  default void close() {
+    //intentionally left blank
+  }
+
   //----------------------------------------------------------------------------
 
+  /**
+   * Creates {@code Serializer} for specified topic's key/value.
+   * Kafka-ui doesn't cache  {@code Serializes} - new one will be created each time user's message needs to be serialized.
+   * (Unless kafka-ui supports batch inserts).
+   */
   Serializer serializer(String topic, Target type);
 
+  /**
+   * Creates {@code Deserializer} for specified topic's key/value.
+   * {@code Deserializer} will be created for each kafka polling and will be used for all messages within that polling cycle.
+   */
   Deserializer deserializer(String topic, Target type);
 
+  /**
+   * Serializes client's input to {@code bytes[]} that will be sent to kafka as key/value (depending on what {@code Type} it was created for).
+   */
   interface Serializer {
+
+    /**
+     * @param input string entered by user into UI text field.<br/> Note: this input is not formatted in any way.
+     */
     byte[] serialize(String input);
   }
 
+  /**
+   * Deserializes polled record's key/value (depending on what {@code Type} it was created for).
+   */
   interface Deserializer {
     DeserializeResult deserialize(RecordHeaders headers, byte[] data);
   }