import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.*;

public class MetadataEventConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "search-indexer-group");
        props.put("key.deserializer",   "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset", "earliest");    // start from beginning if no offset
        props.put("enable.auto.commit", "false");       // manual commit for at-least-once

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(List.of("metadata-events"));

        try {
            while (true) {
                ConsumerRecords<String, String> records =
                    consumer.poll(Duration.ofMillis(100));

                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("topic=%s partition=%d offset=%d key=%s value=%s%n",
                        record.topic(), record.partition(), record.offset(),
                        record.key(), record.value());

                    // Process: update search index
                    processMetadataEvent(record.value());
                }

                // Commit after processing all records in batch
                consumer.commitSync();
            }
        } finally {
            consumer.close();
        }
    }

    static void processMetadataEvent(String event) {
        // Update Elasticsearch index, notify watchers, etc.
        System.out.println("Indexing: " + event);
    }
}

import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import java.util.Properties;

// Kafka Streams: process and transform streams within Kafka
public class EventAggregator {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG,    "event-aggregator");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        StreamsBuilder builder = new StreamsBuilder();

        // Read from input topic
        KStream<String, String> events =
            builder.stream("metadata-events");

        // Filter only schema change events
        KStream<String, String> schemaChanges =
            events.filter((key, value) -> value.contains("schema_change"));

        // Count events per table (tumbling window: 1 hour)
        KTable<Windowed<String>, Long> eventCounts = schemaChanges
            .groupByKey()
            .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofHours(1)))
            .count();

        // Write results to output topic
        eventCounts.toStream()
            .map((key, count) -> KeyValue.pair(key.key(), count.toString()))
            .to("schema-change-stats");

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}