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
