Kafka Streams: How to fix Serde casting error
When I simulate word count case with using aggregate function, I encounter Serde casting issue.
Exception in thread “aggregation-transformation-application-43485635-2d3c-4edc-b13c-c6505a793d18-StreamThread-1” org.apache.kafka.streams.errors.StreamsException: Deserialization exception handler is set to fail upon a deserialization error. If … Read more
What happen if kafka files are deleted?
This is definitively not the way to do it, and it should probably be handled by the cleanup policy but that’s not the point. Let’s imagine the files in log.dirs has been deleted, … Read more
How to send a KeyValue list to Kafka?
I am trying to send a List[KeyValue] to the topic, in a Kafka Streams app. But the stream expects a single KeyValue. How I can send the KeyValues to the stream, instead … Read more
Creating kafka topics with Java doesn’t work
I’m trying to create Kafka topics using Java. But I get a Exception in thread “main” java.lang.NoSuchFieldError: DEFAULT_SSL_PRINCIPAL_MAPPING_RULES and I can’t fix it.
My goal is to create a topic so that when … Read more
Consuming from kafka topic using spring only on receiving certain events
I created a kafka consumer in my spring boot application which listens to a topic – my_topic and on reading an event my_event, I perform some action. Right now … Read more
How to flush data batch in Kafka Consumer when there are no more records in topic
Consider this Kafka consumer that receives data from topic, buffers it into PreparedStatement and when 100K records is batched, it issues INSERT query into … Read more
Is it possible to change the replication factor of an existing kafka topic using java?
I am using AdminClient to alter other configurations of existing kafka topic. But I cannot find a method to change the replication factor of the … Read more
Understanding Kafka Message Byte Size
How do I get the size of a single record in Kafka?
There’s some exposition as to why I need this.
This does not appear to be the serializedValueSize exposed on the ConsumerRecord or RecordMetadata … Read more
How to view and set offsets.retention.minutes using kafka-configs
From the documentation, this is a broker config, but details are thin.
1) What is the correct way to set this to indefinite? If I do, will I still be able to … Read more
Is it correct to use database as a storage for states of messages consumed from kafka?
Currently I have implemented a kafka consumer that works as follows:
Inside a while loop:
Consume message from kafka
Put consumed message into seperate … Read more