Hi Jonathan, Thanks for the suggestion. It is working fine with the kafka 2.2.1-rc1 vesrion. Do we have any tentative release date for this version. Please let me know if you have any info on this. Thanks in Advance.
Thanks Kalyani On 2019/05/24 08:00:33, Jonathan Santilli <[email protected]> wrote: > Hello Kalyani, > > try testing the RC kafka-2.2.1-rc1, for what you describe seems to be a > problem that has been solved in the version 2.2.1 ( > https://issues.apache.org/jira/browse/KAFKA-7895) (which is under voting > right now 2.2.1-RC1 > https://home.apache.org/~vahid/kafka-2.2.1-rc1/RELEASE_NOTES.html) > I have tested the App that was suffering that problem and now is solved. Of > course, you need to test your own App. > > Hope that helps. > > Cheers! > -- > Jonathan > > > > On Thu, May 23, 2019 at 5:37 PM kalyani yarlagadda < > [email protected]> wrote: > > > Hi, > > > > I need assistance in the below scenario. Please help me with this. > > > > I am using the hopping time window in Kafka streams. I am facing an issue > > on the restart of my Kafka application, the application is processing the > > data from the beginning offset. > > However, it is happening only when the topic has more than one partition. > > If the topic has only 1 partition then on the restart of the application > > the sliding window is working fine. > > > > *Kafka Version:* *2.1.0* > > > > *Eg:* > > > > Time Window is 4 hours > > advanceBy 5 minutes > > Application is started at time A and running for every 5minutes with the > > stream data of 4hours > > and now stopped at a timestamp say X, and restarted at timestamp Y > > > > *The behavior for a single partition*: After the restart, the streams are > > processed from time X to time Y for every 5minutes. > > *The behavior for a more than one partition: *After the restart, the > > streams are processed from time A to time Y for every 5minutes. > > > > *I am adding the POC code below* > > > > > > > > > > > > > > > > > > > > *// define the time window as a hopping time window TimeWindows timeWindow > > = TimeWindows.of(Duration.ofHours(4)) .advanceBy(Duration.ofMinutes(5)) > > .grace(Duration.ofMinutes(1)); KTable<Windowed<String>, MetricsTimeSeries> > > windowedMetricsTimeSeriesStream = builder.stream("metrics_ip", > > Consumed.with(Serdes.String(), new JSONSerde<>())) .groupByKey() > > .windowedBy(timeWindow) .aggregate(() -> new MetricsTimeSeries(), /* > > initializer */ * //*MetricsTimeSeries* is the aggregator class > > > > > > > > > > > > > > > > > > > > > > * (aggKey, newValue, aggValue) -> { aggValue.addDataPoint(newValue); return > > aggValue; }, /* adder */ > > Materialized.as("windowed_aggregated_metric_store_poc_partitions")) /* > > state store name */ > > > > .suppress(Suppressed.untilWindowCloses(BufferConfig.maxBytes(10*1024*1024).shutDownWhenFull())); > > windowedMetricsTimeSeriesStream .toStream() .map((key, value) -> //mapping > > logic goes here ) .to("metrics_op");* > > > > *Properties set to Kafka Streams:* > > > > > > > > > > > > > > *StreamsConfig.APPLICATION_ID_CONFIG - > > > > "streams_changedetection_poc_partitions"StreamsConfig.BOOTSTRAP_SERVERS_CONFIG > > - "kafka:9092"StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG - > > > > Serdes.String().getClass()StreamsConfig.DEFAULT_WINDOWED_KEY_SERDE_INNER_CLASS > > - Serdes.String().getClass()StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG > > - JSONSerde.class StreamsConfig.DEFAULT_WINDOWED_VALUE_SERDE_INNER_CLASS - > > JSONSerde.classConsumerConfig.AUTO_OFFSET_RESET_CONFIG - "latest"* > > > > > > > > > > > > Thanks in Advance. > > > > Kalyani Y, > > 9177982636 > > > > > -- > Santilli Jonathan >
