Pluggable SerDe SPI

[iliax]

SPI structure

public interface PluggableSerde extends Closeable {

  void configure(
      PropertyResolver serdeProperties,
      PropertyResolver kafkaClusterProperties,
      PropertyResolver globalProperties, // maybe redundant
      boolean isKey
  );

  // todo: need to think about what headers to use
  byte[] serialize(String topic, Headers headers, String input);

  boolean serializeImplemented();

  DeserializeResult deserialize(String topic, Headers headers, byte[] data);

  boolean deserializeImplemented();

  Optional<SchemaDescription> describeSchema(String topic);
}

//------------------------------------------------------------------------

interface PropertyResolver {
  <T> Optional<T> getProperty(String key, Class<T> targetType);
}

class SchemaDescription {

  public enum Format {
    STRING,
    JSON
  }

  private final Format format;

  // can change to String if we don't want to pull jackson dependency
  private final ObjectNode jsonSchema;

  public static SchemaDescription string() {
    return new SchemaDescription(Format.STRING, null);
  }

  public static SchemaDescription json() {
    return new SchemaDescription(Format.JSON, null);
  }

  public static SchemaDescription jsonWithSchema(ObjectNode jsonSchema) {
    return new SchemaDescription(Format.JSON, jsonSchema);
  }

  private SchemaDescription(Format format, @Nullable ObjectNode jsonSchema) {
    this.format = format;
    this.jsonSchema = jsonSchema;
  }

  // getters...
}

class DeserializeResult {

  private final String result;
  private final Map<String, Object> additionalProperties; // like schemaId, etc. Will be shown in UI

  public DeserializeResult(String result) {
    this(result, Map.of());
  }

  public DeserializeResult(String result, Map<String, Object> additionalProperties) {
    this.result = result;
    this.additionalProperties = additionalProperties;
  }

  // getters...
}

Serde properties example:

KAFKA_CLUSTERS_0_CUSTOM_SERDE_0_NAME=MyCustomSerde
KAFKA_CLUSTERS_0_CUSTOM_SERDE_0_JAR_LOCATION=/var/lib/kafka-ui/serdes/custSerde.jar

# optional (if there are multiple providers in jar file)
KAFKA_CLUSTERS_0_CUSTOM_SERDE_0_CLASSNAME=com.example.MyCustonSerde1

# custom serde properties (passed as configure(..) first param)
KAFKA_CLUSTERS_0_CUSTOM_SERDE_0_PROPS_JSON_BUFF_SIZE=1234
KAFKA_CLUSTERS_0_CUSTOM_SERDE_0_PROPS_FAIL_ON_SMTH=true

# optional
KAFKA_CLUSTERS_0_CUSTOM_SERDE_0_TOPICS_PATTERN_KEYS=topic1|topic2|topic3
KAFKA_CLUSTERS_0_CUSTOM_SERDE_0_TOPICS_PATTERN_VALUES=topic1|topic2|topic3

[iliax]

@Haarolean @germanosin

[germanosin]

Maybe we should configure custom serde on instance level and then pass it to cluster?

[iliax]

@germanosin serde can depend on cluster's config. If we move serde configuration to config's top level it is likely that users will configure one serde for each cluster and duplicate configs.
Suggested option makes its possible to set serde properties on serde, cluster, global levels.

[Haarolean]

I'd suggest renaming (de)serializeImplemented methods to can(De)Serialize. Implementing one doesn't exactly mean that we'd want to use that particular plugin.

[germanosin]

Current approach

public enum DeserializeFormat {
    NONE,
    STRING,
    JSON
}

public interface PluggableSerde extends Closeable {

  void configure(
      PropertyResolver serdeProperties,
      PropertyResolver kafkaClusterProperties,
      PropertyResolver globalProperties,
      boolean isKey
  );

  DeserializeResult deserialize(String topic, Headers headers, byte[] data);
  
  DeserializeFormat canDeserialize(String topic);

  byte[] serialize(String topic, Headers headers, String input);

  boolean canSerialize(String topic);

  Optional<SchemaDescription> describeSchema(String topic);

}


class DeserializeResult {

  private final String result;
  private final Map<String, Object> additionalProperties; // like schemaId, etc. Will be shown in UI

  public DeserializeResult(String result) {
    this(result, Map.of());
  }

  public DeserializeResult(String result, Map<String, Object> additionalProperties) {
    this.result = result;
    this.additionalProperties = additionalProperties;
  }

  // getters...
}

class SchemaDescription {
  private final String jsonSchema;

  private SchemaDescription(String jsonSchema) {
    this.jsonSchema = jsonSchema;
  }

  // getters...
}


KAFKA_CLUSTERS_0_DEFAULT_KEY_SERDE=MyCustomSerde1
KAFKA_CLUSTERS_0_DEFAULT_VALUE_SERDE=MyCustomSerde1

KAFKA_CLUSTERS_0_SERDE_0_NAME=MyCustomSerde1
KAFKA_CLUSTERS_0_SERDE_0_CLASSNAME=com.example.MyCustonSerde1
KAFKA_CLUSTERS_0_SERDE_0_LOCATION=/var/lib/kafka-ui/serdes // jar file or directory
 
# custom serde properties (passed as configure(..) first param)
KAFKA_CLUSTERS_0_SERDE_0_PROPS_JSON_BUFF_SIZE=1234
KAFKA_CLUSTERS_0_SERDE_0_PROPS_FAIL_ON_SMTH=true

KAFKA_CLUSTERS_0_SERDE_1_NAME=MyCustomSerde2
KAFKA_CLUSTERS_0_SERDE_1_CLASSNAME=com.example.MyCustonSerde2
KAFKA_CLUSTERS_0_SERDE_1_LOCATION=/var/lib/kafka-ui/serdes

# custom serde properties (passed as configure(..) first param)
KAFKA_CLUSTERS_0_SERDE_1_PROPS_JSON_BUFF_SIZE=1234
KAFKA_CLUSTERS_0_SERDE_1_PROPS_FAIL_ON_SMTH=true
KAFKA_CLUSTERS_0_SERDE_1_TOPICS_PATTERN_KEYS=topic1|topic2|topic3
KAFKA_CLUSTERS_0_SERDE_1_TOPICS_PATTERN_VALUES=topic1|topic2|topic3

KAFKA_CLUSTERS_0_SERDE_2_NAME=MyCustomSerde3
KAFKA_CLUSTERS_0_SERDE_2_CLASSNAME=com.example.MyCustonSerde3
KAFKA_CLUSTERS_0_SERDE_2_LOCATION=/var/lib/kafka-ui/otherserdes

# custom serde properties (passed as configure(..) first param)
KAFKA_CLUSTERS_0_SERDE_2_PROPS_JSON_BUFF_SIZE=1234
KAFKA_CLUSTERS_0_SERDE_2_PROPS_FAIL_ON_SMTH=true
KAFKA_CLUSTERS_0_SERDE_2_TOPICS_PATTERN_KEYS=topic4|topic5|topic6
KAFKA_CLUSTERS_0_SERDE_2_TOPICS_PATTERN_VALUES=topic4|topic5|topic6

// Predefined serdes
KAFKA_CLUSTERS_0_SERDE_3_NAME=LongSerde
KAFKA_CLUSTERS_0_SERDE_3_CLASSNAME=com.example.LongSerde2 // prepackaged class
KAFKA_CLUSTERS_0_SERDE_3_TOPICS_PATTERN_KEYS=topic7|topic8|topic9

[jdechicchis]

Will different deserializers be configurable per topic and/or per message or would the plugin 100% replace the current deserializer baked into kafkaui? If the former, should deserializeImplemented() take in arguments like deserializeImplemented(topic, msg)?

To clarify, suppose I have topics with Avro as well as protobuf. Can I use the built-in support for Avro while providing a plugin just for my protobuf topics?

[Haarolean]

@jdechicchis so are you interested in implementing this?:)

[jdechicchis]

Happy to help out if needed but my Java knowledge / experience here is a bit lacking so it would take some time. If you / someone else on the team has cycles that would be great, otherwise let me know. Definitely will refactor our plugin to conform to this API once it's ready though

[Haarolean]

@jdechicchis Yes, please feel free to tackle this issue. I've assigned it to you. Also don't hesitate to join our discord if you want to discuss details of implementation. Thanks!

[Haarolean]

@jdechicchis any luck? If you haven't started yet, we can start doing it.

[jdechicchis]

Sorry, haven’t had a chance to start this in earnest yet

[Haarolean]

https://docs.kafka-ui.provectus.io/configuration/serialization-serde