Package org.springframework.cloud.stream.binder.kafka.streams

Class KeyValueSerdeResolver

java.lang.Object
org.springframework.cloud.stream.binder.kafka.streams.KeyValueSerdeResolver
All Implemented Interfaces:
org.springframework.beans.factory.Aware, org.springframework.context.ApplicationContextAware
public class KeyValueSerdeResolver extends Object implements org.springframework.context.ApplicationContextAware
Resolver for key and value Serde. On the inbound, if native decoding is enabled, then any deserialization on the value is handled by Kafka. First, we look for any key/value Serde set on the binding itself, if that is not available then look at the common Serde set at the global level. If that fails, it falls back to byte[]. If native decoding is disabled, then the binder will do the deserialization on value and ignore any Serde set for value and rely on the contentType provided. Keys are always deserialized at the broker. Same rules apply on the outbound. If native encoding is enabled, then value serialization is done at the broker using any binder level Serde for value, if not using common Serde, if not, then byte[]. If native encoding is disabled, then the binder will do serialization using a contentType. Keys are always serialized by the broker. For state store, use serdes class specified in KafkaStreamsStateStore to create Serde accordingly.
Author:
Soby Chacko, Lei Chen, Eduard Domínguez, Chris Bono

  • Method Details

    • getInboundKeySerde

      public org.apache.kafka.common.serialization.Serde<?> getInboundKeySerde(KafkaStreamsConsumerProperties extendedConsumerProperties)
      Provide the Serde for inbound key.
      Parameters:
      extendedConsumerProperties - binding level extended KafkaStreamsConsumerProperties
      Returns:
      configurd Serde for the inbound key.

    • getInboundKeySerde

      public org.apache.kafka.common.serialization.Serde<?> getInboundKeySerde(KafkaStreamsConsumerProperties extendedConsumerProperties, org.springframework.core.ResolvableType resolvableType)

    • getInboundValueSerde

      public org.apache.kafka.common.serialization.Serde<?> getInboundValueSerde(ConsumerProperties consumerProperties, KafkaStreamsConsumerProperties extendedConsumerProperties)
      Provide the Serde for inbound value.
      Parameters:
      consumerProperties - ConsumerProperties on binding
      extendedConsumerProperties - binding level extended KafkaStreamsConsumerProperties
      Returns:
      configurd Serde for the inbound value.

    • getInboundValueSerde

      public org.apache.kafka.common.serialization.Serde<?> getInboundValueSerde(ConsumerProperties consumerProperties, KafkaStreamsConsumerProperties extendedConsumerProperties, org.springframework.core.ResolvableType resolvableType)

    • getOuboundKeySerde

      public org.apache.kafka.common.serialization.Serde<?> getOuboundKeySerde(KafkaStreamsProducerProperties properties)
      Provide the Serde for outbound key.
      Parameters:
      properties - binding level extended KafkaStreamsProducerProperties
      Returns:
      configurd Serde for the outbound key.

    • getOuboundKeySerde

      public org.apache.kafka.common.serialization.Serde<?> getOuboundKeySerde(KafkaStreamsProducerProperties properties, org.springframework.core.ResolvableType resolvableType)

    • getOutboundValueSerde

      public org.apache.kafka.common.serialization.Serde<?> getOutboundValueSerde(ProducerProperties producerProperties, KafkaStreamsProducerProperties kafkaStreamsProducerProperties)
      Provide the Serde for outbound value.
      Parameters:
      producerProperties - ProducerProperties on binding
      kafkaStreamsProducerProperties - binding level extended KafkaStreamsProducerProperties
      Returns:
      configurd Serde for the outbound value.

    • getOutboundValueSerde

      public org.apache.kafka.common.serialization.Serde<?> getOutboundValueSerde(ProducerProperties producerProperties, KafkaStreamsProducerProperties kafkaStreamsProducerProperties, org.springframework.core.ResolvableType resolvableType)

    • getStateStoreKeySerde

      public org.apache.kafka.common.serialization.Serde<?> getStateStoreKeySerde(String keySerdeString)
      Provide the Serde for state store.
      Parameters:
      keySerdeString - serde class used for key
      Returns:
      Serde for the state store key.

    • getStateStoreValueSerde

      public org.apache.kafka.common.serialization.Serde<?> getStateStoreValueSerde(String valueSerdeString)
      Provide the Serde for state store value.
      Parameters:
      valueSerdeString - serde class used for value
      Returns:
      Serde for the state store value.

    • setApplicationContext

      public void setApplicationContext(org.springframework.context.ApplicationContext applicationContext) throws org.springframework.beans.BeansException
      Specified by:
      setApplicationContext in interface org.springframework.context.ApplicationContextAware
      Throws:
      org.springframework.beans.BeansException