<commit_before>package io.github.timothyrenner.kstreamex.windowdelay;

import org.apache.kafka.common.serialization.Serdes;

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.KStreamBuilder;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.Windowed;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.util.Random;
import java.util.Properties;

/** Demonstrates time-ordered windowing.
 *
 * @author TimothyRenner
 */
public class KafkaStreamWindowDelay {
    
    /** Runs the streams program, generating data onto the "timed-longs"
     *  topic and writing results to the "summed-longs" topic.
     *
     * @param args Not used.
     */
    public static void main(String[] args) {
        
        /**** Streams configuration. ******/
        Properties streamsConfig = new Properties();

        streamsConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, 
            "window-delay");
        streamsConfig.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, 
            "localhost:9092");
        streamsConfig.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, 
            "localhost:2181");

        streamsConfig.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG,
            Serdes.String().getClass().getName());
        streamsConfig.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG,
            Serdes.Long().getClass().getName());
        
        /***** Producer configuration. *****/
        Properties producerConfig = new Properties();

        producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 
            "localhost:9092");
        producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
            Serdes.String().serializer().getClass().getName());
        producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
            Serdes.Long().serializer().getClass().getName());
        
        // Build the topology.
        KStreamBuilder builder = new KStreamBuilder();

        KStream<String, Long> longs = builder.stream("timed-longs");

        // NOTE: For some odd reason there's a thirty second latency
        // when pushing these windows out to a topic. This might be
        // a mailing list item. It's exactly thirty seconds and doesn't
        // change when the window size or window wait period changes.
        KTable<Windowed<String>, Long> longSums = 
            longs.groupByKey().reduce(
                (v1, v2) -> v1 + v2,
                TimeWindows.of(10000L) // Ten seconds per window.<commit_msg>Enhanced comments regarding mystery latency.<commit_after>package io.github.timothyrenner.kstreamex.windowdelay;

import org.apache.kafka.common.serialization.Serdes;

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.KStreamBuilder;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.Windowed;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.util.Random;
import java.util.Properties;

/** Demonstrates time-ordered windowing.
 *
 * @author TimothyRenner
 */
public class KafkaStreamWindowDelay {
    
    /** Runs the streams program, generating data onto the "timed-longs"
     *  topic and writing results to the "summed-longs" topic.
     *
     * @param args Not used.
     */
    public static void main(String[] args) {
        
        /**** Streams configuration. ******/
        Properties streamsConfig = new Properties();

        streamsConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, 
            "window-delay");
        streamsConfig.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, 
            "localhost:9092");
        streamsConfig.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, 
            "localhost:2181");

        streamsConfig.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG,
            Serdes.String().getClass().getName());
        streamsConfig.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG,
            Serdes.Long().getClass().getName());
        
        /***** Producer configuration. *****/
        Properties producerConfig = new Properties();

        producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 
            "localhost:9092");
        producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
            Serdes.String().serializer().getClass().getName());
        producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
            Serdes.Long().serializer().getClass().getName());
        
        // Build the topology.
        KStreamBuilder builder = new KStreamBuilder();

        KStream<String, Long> longs = builder.stream("timed-longs");

        // NOTE: For some odd reason there's a thirty second latency
        // when pushing these windows out to a topic. 
        // It's exactly thirty seconds and doesn't
        // change when the window size or window wait period changes, which
        // smells like a configuration option somewhere, but I'm not totally
        // sure.
        KTable<Windowed<String>, Long> longSums = 
            longs.groupByKey().reduce(
                (v1, v2) -> v1 + v2,
                TimeWindows.of(10000L) // Ten seconds per window.
{'fraction_non_alphanumeric': 0.10849627963315453, 'fraction_numerical': 0.007267693372555805, 'mean_word_length': 3.0447865640307907, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6351601', 'n_tokens_mistral': 1710, 'n_tokens_neox': 1668, 'n_words': 322}