12

I'm using kafka stream and I'm trying to materialize a KTable into a topic.

It works but it seems to be done every 30 secs or so.

How/When does Kafka Stream decides to materialize the current state of a KTable into a topic ?

Is there any way to shorten this time and to make it more "real-time" ?

Here is the actual code I'm using

// Stream of random ints: (1,1) -> (6,6) -> (3,3)
// one record every 500ms
KStream<Integer, Integer> kStream = builder.stream(Serdes.Integer(), Serdes.Integer(), RandomNumberProducer.TOPIC);

// grouping by key
KGroupedStream<Integer, Integer> byKey = kStream.groupByKey(Serdes.Integer(), Serdes.Integer());

// same behaviour with or without the TimeWindow
KTable<Windowed<Integer>, Long> count = byKey.count(TimeWindows.of(1000L),"total");

// same behaviour with only count.to(Serdes.Integer(), Serdes.Long(), RandomCountConsumer.TOPIC);
count.toStream().map((k,v) -> new KeyValue<>(k.key(), v)).to(Serdes.Integer(), Serdes.Long(), RandomCountConsumer.TOPIC);
3
  • Could you be more clear on what you're trying to accomplish? Some code? Do you mean you're doing something like: ktable.to("topic_name")?
    – groo
    Jun 23, 2017 at 11:00
  • yes this is exactly what I'm doing
    – thomas.g
    Jun 23, 2017 at 11:30
  • 1
    Ok, I understand your issue but unfortunately I have never tuned this specific case yet(I will actually need something similar to this soon as we're implementing a solution that requires this to be very fast / up to date) with that said I would start playing with the available configurations: kafka.apache.org/documentation/#configuration Possibly first with the following: broker level: log.flush.* ones, offsets.commit.timeout.ms, producer timeouts. topic level: flush.* ones stream level: commit.* ones If you find a solution please post it here, It will be useful
    – groo
    Jun 23, 2017 at 12:48

1 Answer 1

13

This is controlled by commit.interval.ms, which defaults to 30s. More details here: http://docs.confluent.io/current/streams/developer-guide.html

The semantics of caching is that data is flushed to the state store and forwarded to the next downstream processor node whenever the earliest of commit.interval.ms or cache.max.bytes.buffering (cache pressure) hits.

and here:

https://cwiki.apache.org/confluence/display/KAFKA/KIP-63%3A+Unify+store+and+downstream+caching+in+streams

1
  • the documentation does not describe the effects of these settings on speed
    – user482963
    Jun 20, 2019 at 6:39

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.