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

Class CollectionSerde<E>

java.lang.Object
org.springframework.cloud.stream.binder.kafka.streams.serde.CollectionSerde<E>
Type Parameters:
E - type of the underlying object that the collection holds
All Implemented Interfaces:
Closeable, AutoCloseable, org.apache.kafka.common.serialization.Serde<Collection<E>>
public class CollectionSerde<E> extends Object implements org.apache.kafka.common.serialization.Serde<Collection<E>>
A convenient Serde for Collection implementations. Whenever a Kafka Stream application needs to collect data into a container object like Collection, then this Serde class can be used as a convenience for serialization needs. Some examples of where using this may handy is when the application needs to do aggregation or reduction operations where it needs to simply hold an Iterable type. By default, this Serde will use JsonSerde for serializing the inner objects. This can be changed by providing an explicit Serde during creation of this object. Here is an example of a possible use case: 
                .aggregate(ArrayList::new,
                                        (k, v, aggregates) -> {
                                                        aggregates.add(v);
                                                        return aggregates;
                                                },
                                        Materialized.<String, Collection<Foo>, WindowStore<Bytes, byte[]>>as(
                                                "foo-store")
                                                .withKeySerde(Serdes.String())
                                                .withValueSerde(new CollectionSerde<>(Foo.class, ArrayList.class)))
  *
Supported Collection types by this Serde are - ArrayList, LinkedList, PriorityQueue and HashSet. Deserializer will throw an exception if any other Collection types are used.
Since:
3.0.0
Author:
Soby Chacko

  • Constructor Details

    • CollectionSerde

      public CollectionSerde(org.apache.kafka.common.serialization.Serde<E> serde, Class<?> collectionsClass)
      Constructor to use when the application wants to specify the type of the Serde used for the inner object.
      Parameters:
      serde - specify an explicit Serde
      collectionsClass - type of the Collection class

    • CollectionSerde

      public CollectionSerde(Class<?> targetTypeForJsonSerde, Class<?> collectionsClass)
      Constructor to delegate serialization operations for the inner objects to JsonSerde.
      Parameters:
      targetTypeForJsonSerde - target type used by the JsonSerde
      collectionsClass - type of the Collection class

  • Method Details

    • serializer

      public org.apache.kafka.common.serialization.Serializer<Collection<E>> serializer()
      Specified by:
      serializer in interface org.apache.kafka.common.serialization.Serde<E>

    • deserializer

      public org.apache.kafka.common.serialization.Deserializer<Collection<E>> deserializer()
      Specified by:
      deserializer in interface org.apache.kafka.common.serialization.Serde<E>

    • configure

      public void configure(Map<String,?> configs, boolean isKey)
      Specified by:
      configure in interface org.apache.kafka.common.serialization.Serde<E>

    • close

      public void close()
      Specified by:
      close in interface AutoCloseable
      Specified by:
      close in interface Closeable
      Specified by:
      close in interface org.apache.kafka.common.serialization.Serde<E>