Thanks for the comments.
@David: yes, I have a source which is reading data from two topics and one
of them were empty while the second one was loaded with plenty of data.
So what do you suggest to solve this ?
Here is snippet of my code:
StreamsConfig config = new StreamsConfig(configProperties);
TopologyBuilder builder = new TopologyBuilder();
AppSettingsFetcher appSettingsFetcher = initAppSettingsFetcher();
StateStoreSupplier company_bucket= Stores.create("CBS")
.withKeys(Serdes.String())
.withValues(Serdes.String())
.persistent()
.build();
StateStoreSupplier profiles= Stores.create("PR")
.withKeys(Serdes.String())
.withValues(Serdes.String())
.persistent()
.build();
builder
.addSource("deltaSource", topicName, LoaderListener.LoadedDeltaTopic)
.addProcessor("deltaProcess1", () -> new DeltaProcessor(
), "deltaSource")
.addProcessor("deltaProcess2", () -> new LoadProcessor(
), "deltaProcess1")
.addStateStore(profiles, "deltaProcess2", "deltaProcess1")
.addStateStore(company_bucket, "deltaProcess2", "deltaProcess1");
KafkaStreams streams = new KafkaStreams(builder, config);
streams.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
@Override
public void uncaughtException(Thread t, Throwable e) {
e.printStackTrace();
}
});
streams.start();
On Wed, Nov 23, 2016 at 8:30 PM, David Garcia <dav...@spiceworks.com> wrote:
> If you are consuming from more than one topic/partition, punctuate is
> triggered when the “smallest” time-value changes. So, if there is a
> partition that doesn’t have any more messages on it, it will always have
> the smallest time-value and that time value won’t change…hence punctuate
> never gets called.
>
> -David
>
> On 11/23/16, 1:01 PM, "Matthias J. Sax" <matth...@confluent.io> wrote:
>
> Your understanding is correct:
>
> Punctuate is not triggered base on wall-clock time, but based in
> internally tracked "stream time" that is derived from
> TimestampExtractor.
> Even if you use WallclockTimestampExtractor, "stream time" is only
> advance if there are input records.
>
> Not sure why punctuate() is not triggered as you say that you do have
> arriving data.
>
> Can you share your code?
>
>
>
> -Matthias
>
>
> On 11/23/16 4:48 AM, shahab wrote:
> > Hello,
> >
> > I am using low level processor and I set the context.schedule(10000),
> > assuming that punctuate() method is invoked every 10 sec .
> > I have set
> > configProperties.put(StreamsConfig.TIMESTAMP_EXTRACTOR_CLASS_CONFIG,
> > WallclockTimestampExtractor.class.getCanonicalName()) )
> >
> > Although data is keep coming to the topology (as I have logged the
> incoming
> > tuples to process() ), punctuate() is never executed.
> >
> > What I am missing?
> >
> > best,
> > Shahab
> >
>
>
>
>
