Issue with ArrayList Serde in Kafka Streams API

Question

Based on my previous question, I am still trying to figure out what's the issue with my code.

I've got a most basic topic possible: keys and values are a type of Long and this is my producer code:

public class DemoProducer {
  public static void main(String... args) {
    Producer<Long, Long> producer = new KafkaProducer<>(createProperties());

    LongStream.range(1, 100)
        .forEach(
            i ->
                LongStream.range(100, 115)
                    .forEach(j -> producer.send(new ProducerRecord<>("test", i, j))));

    producer.close();
  }

  private static final Properties createProperties() {
    final Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
    props.put(ProducerConfig.ACKS_CONFIG, "all");
    props.put(ProducerConfig.RETRIES_CONFIG, 0);
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class.getName());
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, LongSerializer.class.getName());

    return props;
  }
}

I'd like to group things up using key and put values in a ArrayList using Kafka Streams API.

This is my Stream app that's supposed to do the transformation and put things to new topic - test-aggregated:

public class DemoStreams {
  public static void main(String... args) {
    final Serde<Long> longSerde = Serdes.Long();

    KStreamBuilder builder = new KStreamBuilder();

    builder
        .stream(longSerde, longSerde, "test")
        .groupByKey(longSerde, longSerde)
        .aggregate(
            ArrayList::new,
            (subscriberId, reportId, queue) -> {
              queue.add(reportId);
              return queue;
            },
            new ArrayListSerde<>(longSerde))
        .to(longSerde, new ArrayListSerde<>(longSerde), "test-aggregated");

    final KafkaStreams streams = new KafkaStreams(builder, createProperties());

    streams.cleanUp();
    streams.start();

    Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
  }

  private static Properties createProperties() {
    final Properties properties = new Properties();
    String longSerdes = Serdes.Long().getClass().getName();
    properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "aggregation-app");
    properties.put(StreamsConfig.CLIENT_ID_CONFIG, "aggregation-app-client");
    properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
    properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, longSerdes);
    properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, ArrayListSerde.class);
    properties.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * 1000);
    properties.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);

    return properties;
  }
}

I have implemented my Serde as follows:

ArrayListSerde

public class ArrayListSerde<T> implements Serde<ArrayList<T>> {

  private final Serde<ArrayList<T>> inner;

  public ArrayListSerde(Serde<T> serde) {
    inner =
        Serdes.serdeFrom(
            new ArrayListSerializer<>(serde.serializer()),
            new ArrayListDeserializer<>(serde.deserializer()));
  }

  @Override
  public Serializer<ArrayList<T>> serializer() {
    return inner.serializer();
  }

  @Override
  public Deserializer<ArrayList<T>> deserializer() {
    return inner.deserializer();
  }

  @Override
  public void configure(Map<String, ?> configs, boolean isKey) {
    inner.serializer().configure(configs, isKey);
    inner.deserializer().configure(configs, isKey);
  }

  @Override
  public void close() {
    inner.serializer().close();
    inner.deserializer().close();
  }
}

ArrayListSerializer

public class ArrayListSerializer<T> implements Serializer<ArrayList<T>> {

  private Serializer<T> inner;

  public ArrayListSerializer(Serializer<T> inner) {
    this.inner = inner;
  }

  // Default constructor needed by Kafka
  public ArrayListSerializer() {}

  @Override
  public void configure(Map<String, ?> configs, boolean isKey) {
    // do nothing
  }

  @Override
  public byte[] serialize(String topic, ArrayList<T> queue) {
    final int size = queue.size();
    final ByteArrayOutputStream baos = new ByteArrayOutputStream();
    final DataOutputStream dos = new DataOutputStream(baos);
    final Iterator<T> iterator = queue.iterator();
    try {
      dos.writeInt(size);
      while (iterator.hasNext()) {
        final byte[] bytes = inner.serialize(topic, iterator.next());
        dos.writeInt(bytes.length);
        dos.write(bytes);
      }
    } catch (IOException e) {
      throw new RuntimeException("Unable to serialize ArrayList", e);
    }
    return baos.toByteArray();
  }

  @Override
  public void close() {
    inner.close();
  }
}

ArrayListDeserializer

public class ArrayListDeserializer<T> implements Deserializer<ArrayList<T>> {
  private final Deserializer<T> valueDeserializer;

  public ArrayListDeserializer(final Deserializer<T> valueDeserializer) {
    this.valueDeserializer = valueDeserializer;
  }

  @Override
  public void configure(Map<String, ?> configs, boolean isKey) {
    // do nothing
  }

  @Override
  public ArrayList<T> deserialize(String topic, byte[] bytes) {
    if (bytes == null || bytes.length == 0) {
      return null;
    }

    final ArrayList<T> arrayList = new ArrayList<>();
    final DataInputStream dataInputStream = new DataInputStream(new ByteArrayInputStream(bytes));

    try {
      final int records = dataInputStream.readInt();
      for (int i = 0; i < records; i++) {
        final byte[] valueBytes = new byte[dataInputStream.readInt()];
        dataInputStream.read(valueBytes);
        arrayList.add(valueDeserializer.deserialize(topic, valueBytes));
      }
    } catch (IOException e) {
      throw new RuntimeException("Unable to deserialize ArrayList", e);
    }

    return arrayList;
  }

  @Override
  public void close() {
    // do nothing
  }
}

However I end up getting this Exception:

Exception in thread "permission-agg4-client-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: stream-thread [aggregation-app-client-StreamThread-1] Failed to rebalance.
    at org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:543)
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:490)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:480)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:457)
Caused by: org.apache.kafka.streams.errors.StreamsException: Failed to configure value serde class utils.ArrayListSerde
    at org.apache.kafka.streams.StreamsConfig.defaultValueSerde(StreamsConfig.java:770)
    at org.apache.kafka.streams.processor.internals.AbstractProcessorContext.<init>(AbstractProcessorContext.java:59)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.<init>(ProcessorContextImpl.java:40)
    at org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:138)
    at org.apache.kafka.streams.processor.internals.StreamThread.createStreamTask(StreamThread.java:1078)
    at org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:255)
    at org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.createTasks(StreamThread.java:245)
    at org.apache.kafka.streams.processor.internals.StreamThread.addStreamTasks(StreamThread.java:1147)
    at org.apache.kafka.streams.processor.internals.StreamThread.access$800(StreamThread.java:68)
    at org.apache.kafka.streams.processor.internals.StreamThread$RebalanceListener.onPartitionsAssigned(StreamThread.java:184)
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:265)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:367)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:316)
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:297)
    at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1078)
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1043)
    at org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:536)
    ... 3 more
Caused by: org.apache.kafka.common.KafkaException: Could not instantiate class utils.ArrayListSerde Does it have a public no-argument constructor?
    at org.apache.kafka.common.utils.Utils.newInstance(Utils.java:286)
    at org.apache.kafka.common.config.AbstractConfig.getConfiguredInstance(AbstractConfig.java:246)
    at org.apache.kafka.streams.StreamsConfig.defaultValueSerde(StreamsConfig.java:764)
    ... 19 more
Caused by: java.lang.InstantiationException: utils.ArrayListSerde
    at java.lang.Class.newInstance(Class.java:427)
    at org.apache.kafka.common.utils.Utils.newInstance(Utils.java:282)
    ... 21 more
Caused by: java.lang.NoSuchMethodException: utils.ArrayListSerde.<init>()
    at java.lang.Class.getConstructor0(Class.java:3082)
    at java.lang.Class.newInstance(Class.java:412)
    ... 22 more

I was trying to implement Serde based on PriorityQueue example found in Confluent's GitHub page: https://github.com/confluentinc/kafka-streams-examples/tree/3.3.0-post/src/main/java/io/confluent/examples/streams/utils

Answer 1

As the error indicates, all Serdes need to have a non-argument constructor:

Caused by: org.apache.kafka.common.KafkaException: Could not instantiate class utils.ArrayListSerde Does it have a public no-argument constructor?

You class ArrayListSerde does only have constructor:

public ArrayListSerde(Serde<T> serde) { ... }

Thus, we get this error.

Compare ArrayListSerializer:

// Default constructor needed by Kafka
public ArrayListSerializer() {}

Update:

A standard implementation for ListSerde is WIP and should be included in a future release making custom List Serde obsolete: https://issues.apache.org/jira/browse/KAFKA-8326