Working with Time-Based Processing

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> source = builder.stream("events");
KTable<String, Long> counts = source
    .selectKey((key, value) -> value.split(":")[0])
    .groupBy((key, value) -> key)
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
    .count();

import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Produced;

KStream<String, String> inputStream = builder.stream("input-topic");

KStream<String, String> withEventTime = inputStream
    .selectKey((key, value) -> {
        // Extract timestamp from the event payload
        long eventTimestamp = extractTimestamp(value);
        return new org.apache.kafka.common.utils.Time().milliseconds();
    })
    .withTimestamps((key, value, timestamp) -> timestamp);

withEventTime.to("output-topic", Produced.with(Serdes.String(), Serdes.String()));

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> source = builder.stream("input-topic");

KStream<String, String> processed = source
    .filter((key, value) -> {
        // Use processing time for filtering or windowing
        long currentTime = System.currentTimeMillis();
        return true; // Example logic
    });

KafkaStreams streams = new KafkaStreams(builder.build(), config);
streams.start();

ProducerRecord<String, String> record = new ProducerRecord<>(
    "my-topic",
    "key1",
    "message-value"
);

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "timestamp-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.TIMESTAMP_EXTRACTOR_CLASS_CONFIG, CustomTimestampExtractor.class);

import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.SystemTime;

public class CustomTimestampExtractor implements TimestampExtractor {
    private final Time time = new SystemTime();

    @Override
    public long extract(ConsumerRecord<String, String> record, long partitionTime) {
        // Use a custom timestamp from the record's payload or metadata
        String payload = record.value();
        if (payload != null && payload.contains("timestamp")) {
            // Extract timestamp from payload (example logic)
            return parseTimestampFromPayload(payload);
        }
        return time.milliseconds(); // Fallback to current system time
    }

    private long parseTimestampFromPayload(String payload) {
        // Dummy implementation for illustration
        return System.currentTimeMillis();
    }
}

StreamsBuilder builder = new StreamsBuilder();
KTable<String, Long> windowedCounts = builder.stream("input-topic")
    .groupBy((key, value) -> key)
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
    .count();

StreamsBuilder builder = new StreamsBuilder();
KTable<String, Integer> table = builder.table("input-topic");

TimeWindows windows = TimeWindows.of(Duration.ofMinutes(5))
    .advanceBy(Duration.ofMinutes(1));

KGroupedStream<String, Integer> grouped = table.groupBy(KeyValue.pair("key", 10));
KTable<Windowed<String>, Long> windowedCount = grouped
    .windowedBy(windows)
    .count();

StreamsBuilder builder = new StreamsBuilder();

KTable<String, Long> windowedCounts = builder
    .stream("input-topic")
    .groupBy((key, value) -> key)
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5)).advanceBy(Duration.ofMinutes(1)))
    .count();

windowedCounts.toStream()
    .mapValues((key, value) -> String.format("Key: %s, Count: %d", key.key(), value))
    .to("output-topic");

StreamsBuilder builder = new StreamsBuilder();
KTable<String, Long> sessionTable = builder.table(
    "input-topic",
    Consumed.with(Serdes.String(), Serdes.Long())
).groupBy(
    (key, value) -> key,
    Grouped.with(Serdes.String(), Serdes.Long())
).windowedBy(
    SessionWindows.of(300000L)
).reduce(
    (value1, value2) -> value1 + value2
);

{
  "bootstrap.servers": "localhost:9092",
  "group.id": "my-consumer-group",
  "enable.auto.commit": false,
  "graceful.shutdown.timeout.ms": 30000
}

{
  "bootstrap.servers": "localhost:9092",
  "acks": "all",
  "retries": 3,
  "graceful.shutdown.timeout.ms": 30000
}

import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;

KafkaStreams streams = new KafkaStreams(
    builder.build(),
    new StreamsConfig(props)
);

// Define a watermark for event time
KStream<String, MyEvent> events = builder.stream("events-topic");

// Apply windowing with a 5-minute allowed lateness
KTable<String, Long> counts = events
    .groupBy(KeyValue::new)
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5)).grace(Duration.ofMinutes(1)))
    .count();

streams.start();