import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import java.time.Duration;
import java.util.Properties;

public class OrderProcessingStream {
    
    public static void main(String[] args) {
        Properties props = new Properties();
        
        // Application identity; used as consumer group ID
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-processing-app");
        
        // Bootstrap servers for initial metadata discovery
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        
        // Default serializers/deserializers for keys and values
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
        
        // Exactly-once semantics (v2 for improved performance)
        props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, 
            StreamsConfig.EXACTLY_ONCE_V2);
        // Automatically configures idempotent producers, transactional writes
        // Reduces latency by ~30% vs EOS v1
        
        // State store configuration
        props.put(StreamsConfig.STATE_DIR_CONFIG, "/tmp/kafka-streams");
        // Directory for RocksDB state stores; use SSD for best performance
        
        props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);
        // How often offsets are committed to __consumer_offsets
        // More frequent = lower latency, more overhead
        
        props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 10 * 1024 * 1024L);
        // 10MB cache for write-back buffering
        // Larger cache = fewer changelog writes, slower recovery
        
        // Thread and task configuration
        props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 4);
        // Number of stream threads per instance
        // Each thread processes a subset of tasks
        
        StreamsBuilder builder = new StreamsBuilder();
        
        // Source: Raw order events
        KStream<String, String> orders = builder.stream("raw-orders",
            Consumed.with(Serdes.String(), Serdes.String())
                .withTimestampExtractor(new WallclockTimestampExtractor()));
        // WallclockTimestampExtractor uses processing time (system clock)
        // For event-time: use ExtractRecordTimestamp (from message timestamp)
        
        // Source: User data (KTable for point lookups)
        KTable<String, String> users = builder.table("user-data",
            Consumed.with(Serdes.String(), Serdes.String()));
        // KTable enables point lookups by key for joins
        
        // Transform and enrich
        KStream<String, Order> enrichedOrders = orders
            .filter((key, value) -> value != null && !value.isEmpty())
            // Filter out null/empty records
            .mapValues(value -> parseOrder(value))
            // MapValues preserves key, transforms value
            .filter((key, order) -> order.getAmount() > 0)
            // Filter invalid orders
            .join(users,
                (order, user) -> order.withUser(user),
                // Join function: combine order with user data
                Joined.with(Serdes.String(), Serdes.String(), Serdes.String()));
            // Joined specifies key, value, and foreign value serdes
        
        // Branch by order type
        Map<String, KStream<String, Order>> branches = enrichedOrders
            .branch((key, order) -> order.getType().equals("PREMIUM"),
                Branched.withFunction((stream) -> {
                    stream.mapValues(o -> o.applyDiscount(0.10))
                          .to("premium-orders", Produced.with(Serdes.String(), orderSerde));
                    return stream;
                }, "premium"))
            .branch((key, order) -> order.getType().equals("STANDARD"),
                Branched.withFunction((stream) -> {
                    stream.mapValues(o -> o.applyDiscount(0.0))
                          .to("standard-orders", Produced.with(Serdes.String(), orderSerde));
                    return stream;
                }, "standard"));
        // Branched allows naming branches for monitoring
        
        // Aggregation: Count orders per user in 1-hour tumbling windows
        KTable<Windowed<String>, Long> orderCounts = enrichedOrders
            .groupByKey()
            // Group by key (user) for aggregation
            .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofHours(1)))
            // 1-hour tumbling window, no grace period
            .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as(
                    "order-count-store")
                .withValueSerde(Serdes.Long()));
            // Materialized creates a named state store
        
        // Aggregate: Total revenue per user in 5-minute hopping windows
        KTable<Windowed<String>, Double> revenueByUser = enrichedOrders
            .mapValues(order -> order.getAmount())
            .groupByKey()
            .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5))
                .advanceBy(Duration.ofMinutes(1)))
            // Hopping: 5-minute windows, advance every 1 minute (overlapping)
            .reduce(Double::sum, Materialized.<String, Double, WindowStore<Bytes, byte[]>>as(
                    "revenue-store")
                .withValueSerde(Serdes.Double()));
        
        // Session windowing for activity tracking
        KTable<Windowed<String>, Long> sessionCounts = enrichedOrders
            .map((key, order) -> KeyValue.pair(order.getUserId(), order))
            .groupByKey()
            .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(30)))
            // Session window: merges events within 30-minute inactivity gap
            .count(Materialized.<String, Long, SessionStore<Bytes, byte[]>>as(
                    "session-store")
                .withValueSerde(Serdes.Long()));
        
        // Sink: Write enriched orders
        enrichedOrders
            .mapValues(order -> order.toJson())
            .to("enriched-orders", Produced.with(Serdes.String(), Serdes.String()));
        
        StreamsTopology topology = builder.build();
        System.out.println(topology.describe());
        // Print topology for debugging
        
        KafkaStreams streams = new KafkaStreams(topology, props);
        streams.start();
        
        // Graceful shutdown
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            streams.close(Duration.ofSeconds(30));
            // Close waits up to 30s for in-flight records
            streams.cleanUp();
            // CleanUp removes local state stores (optional)
        }));
    }
    
    private static Order parseOrder(String json) {
        // JSON parsing implementation
        return new Order(); // Placeholder
    }
}

import org.apache.kafka.streams.processor.*;
import org.apache.kafka.streams.state.*;
import java.time.Duration;
import java.util.*;

public class FraudDetectionProcessor implements Processor<String, OrderEvent, String, Alert> {
    
    private ProcessorContext<String, Alert> context;
    private KeyValueStore<String, UserTransactionSummary> stateStore;
    private WindowStore<String, OrderEvent> recentOrders;
    
    @Override
    public void init(ProcessorContext<String, Alert> context) {
        this.context = context;
        
        // Get reference to state store (must be added to topology)
        this.stateStore = context.getStateStore("user-transaction-summary");
        this.recentOrders = context.getStateStore("recent-orders-window");
        
        // Schedule periodic punctuation for cleanup
        context.schedule(Duration.ofMinutes(5), 
            PunctuationType.WALL_CLOCK_TIME, 
            this::cleanupOldOrders);
        // Punctuation runs periodically for housekeeping tasks
        
        context.schedule(Duration.ofMinutes(1), 
            PunctuationType.WALL_CLOCK_TIME, 
            this::checkVelocity);
    }
    
    @Override
    public void process(Record<String, OrderEvent> record) {
        String userId = record.value().getUserId();
        OrderEvent event = record.value();
        
        // Store in window store for velocity checks
        recentOrders.put(record.key(), event, 
            Instant.ofEpochMilli(event.getTimestamp()));
        // Window store enables range queries by time
        
        // Get or initialize user summary
        UserTransactionSummary summary = stateStore.get(userId);
        if (summary == null) {
            summary = new UserTransactionSummary(userId);
        }
        
        // Fraud detection logic
        Alert alert = null;
        
        // Rule 1: Amount threshold
        if (event.getAmount() > summary.getAverageAmount() * 3) {
            alert = new Alert(userId, "HIGH_AMOUNT", 
                "Transaction amount 3x above average", event);
        }
        
        // Rule 2: Velocity check (too many transactions)
        long recentCount = countRecentTransactions(userId, Duration.ofMinutes(5));
        if (recentCount > summary.getTransactionsPerMinute() * 5) {
            alert = new Alert(userId, "HIGH_VELOCITY", 
                "Transaction velocity 5x above normal", event);
        }
        
        // Rule 3: Geographic anomaly
        if (summary.getLastLocation() != null && 
            !summary.getLastLocation().equals(event.getLocation())) {
            alert = new Alert(userId, "GEO_ANOMALY", 
                "Location changed from " + summary.getLastLocation() + 
                " to " + event.getLocation(), event);
        }
        
        // Update summary
        summary.update(event);
        stateStore.put(userId, summary);
        
        // Forward alert if fraud detected
        if (alert != null) {
            context.forward(new Record<>(userId, alert, record.timestamp()));
        }
    }
    
    private long countRecentTransactions(String userId, Duration window) {
        Instant from = Instant.now().minus(window);
        Instant to = Instant.now();
        
        long count = 0;
        try (KeyValueIterator<Windowed<String>, OrderEvent> iterator = 
                recentOrders.fetch(userId, from, to)) {
            while (iterator.hasNext()) {
                count++;
                iterator.next();
            }
        }
        return count;
    }
    
    private void cleanupOldOrders(Instant timestamp) {
        Instant cutoff = timestamp.minus(Duration.ofHours(1));
        try (KeyValueIterator<Windowed<String>, OrderEvent> iterator = 
                recentOrders.all()) {
            while (iterator.hasNext()) {
                KeyValue<Windowed<String>, OrderEvent> entry = iterator.next();
                if (entry.key.window().endTime().isBefore(cutoff)) {
                    recentOrders.delete(entry.key);
                }
            }
        }
    }
    
    private void checkVelocity(Instant timestamp) {
        // Periodic velocity check logic
    }
    
    @Override
    public void close() {}
}

{
  "name": "mysql-cdc-source",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql-primary.example.com",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "${file:/opt/kafka/secrets/mysql.properties:password}",
    "database.server.id": "184054",
    "database.server.name": "ecommerce",
    "database.include.list": "ecommerce",
    "table.include.list": "ecommerce.orders,ecommerce.users,ecommerce.products",
    
    "topic.prefix": "cdc",
    "topic.creation.default.replication.factor": 3,
    "topic.creation.default.partitions": 12,
    "topic.creation.default.cleanup.policy": "delete",
    "topic.creation.default.retention.ms": 604800000,
    
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://schema-registry:8081",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",
    
    "transforms": "route,unwrap,addTimestamp",
    "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.route.regex": "ecommerce\\.(.*)",
    "transforms.route.replacement": "cdc-$1",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.add.fields": "op,ts_ms,source.ts_ms",
    "transforms.addTimestamp.type": "org.apache.kafka.connect.transforms.InsertField$Value",
    "transforms.addTimestamp.timestamp.field": "processed_at",
    
    "heartbeat.interval.ms": "30000",
    "poll.interval.ms": "1000",
    "max.batch.size": "8192",
    "snapshot.mode": "initial",
    "snapshot.locking.mode": "none",
    "snapshot.select.statement.overrides": "ecommerce.orders",
    
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "dlq-mysql-cdc",
    "errors.deadletterqueue.topic.replication.factor": 3,
    "errors.log.enable": true,
    "errors.log.include.messages": true,
    
    "tasks.max": "3"
  }
}

{
  "name": "elasticsearch-sink",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "connection.url": "http://elasticsearch:9200",
    "connection.username": "${file:/opt/kafka/secrets/es.properties:username}",
    "connection.password": "${file:/opt/kafka/secrets/es.properties:password}",
    
    "topics": "enriched-orders,enriched-users,enriched-products",
    "key.ignore": false,
    "schema.ignore": false,
    
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",
    
    "transforms": "extractKey,indexByDate,typeRoute",
    "transforms.extractKey.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
    "transforms.extractKey.field": "order_id",
    "transforms.indexByDate.type": "org.apache.kafka.connect.transforms.TimestampRouter",
    "transforms.indexByDate.timestamp.format": "YYYY-MM-dd",
    "transforms.indexByDate.topic.format": "${topic}-$${timestamp}",
    "transforms.typeRoute.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.typeRoute.regex": "enriched-(.*)",
    "transforms.typeRoute.replacement": "kafka-$1",
    
    "batch.size": 200,
    "max.buffered.records": 500,
    "linger.ms": 1,
    "flush.timeout.ms": 10000,
    "max.retries": 5,
    "retry.backoff.ms": 100,
    
    "behavior.on.null.values": "delete",
    "behavior.on.malformed.documents": "warn",
    
    "write.method": "upsert",
    "document.id.strategy": "record_key",
    
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "dlq-elasticsearch",
    "errors.deadletterqueue.topic.replication.factor": 3,
    
    "tasks.max": "3"
  }
}

import io.confluent.kafka.streams.serdes.avro.GenericAvroSerde;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import java.util.Properties;
import java.util.HashMap;
import java.util.Map;

public class StreamsSchemaRegistryExample {
    
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "schema-registry-streams-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        
        // Schema Registry configuration
        props.put("schema.registry.url", "http://localhost:8081");
        props.put("basic.auth.credentials.source", "USER_INFO");
        props.put("basic.auth.user.info", "admin:secret");
        
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, GenericAvroSerde.class);
        
        StreamsBuilder builder = new StreamsBuilder();
        
        // Configure Avro serde
        Map<String, String> serdeConfig = new HashMap<>();
        serdeConfig.put("schema.registry.url", "http://localhost:8081");
        serdeConfig.put("basic.auth.credentials.source", "USER_INFO");
        serdeConfig.put("basic.auth.user.info", "admin:secret");
        
        GenericAvroSerde avroSerde = new GenericAvroSerde();
        avroSerde.configure(serdeConfig, false);
        // false = not a key serde (value serde)
        
        // Source stream with Avro deserialization
        KStream<String, GenericRecord> orders = builder.stream("order-events",
            Consumed.with(Serdes.String(), avroSerde));
        
        // Transform and filter
        KStream<String, GenericRecord> highValueOrders = orders
            .filter((key, order) -> {
                double amount = (double) order.get("amount");
                return amount > 1000;
            })
            .mapValues(order -> {
                // Create new Avro record with evolved schema
                Schema newSchema = SchemaBuilder.builder("com.example")
                    .record("HighValueOrder")
                    .fields()
                        .name("orderId")
                        .type(order.getSchema().getField("orderId").schema())
                        .noDefault()
                        .name("userId")
                        .type(order.getSchema().getField("userId").schema())
                        .noDefault()
                        .name("amount")
                        .type(order.getSchema().getField("amount").schema())
                        .noDefault()
                        .name("verified")
                        .type()
                        .booleanType()
                        .booleanDefault(false)
                    .endRecord();
                
                GenericRecord highValueOrder = new GenericData.Record(newSchema);
                highValueOrder.put("orderId", order.get("orderId"));
                highValueOrder.put("userId", order.get("userId"));
                highValueOrder.put("amount", order.get("amount"));
                highValueOrder.put("verified", false);
                
                return highValueOrder;
            });
        
        // Sink with Avro serialization
        highValueOrders.to("high-value-orders", 
            Produced.with(Serdes.String(), avroSerde));
        
        // Aggregation with windowing
        KTable<Windowed<String>, Long> orderCounts = orders
            .groupByKey()
            .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofHours(1)))
            .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as(
                    "order-count-store")
                .withValueSerde(Serdes.Long()));
        
        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
        
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}

#!/bin/bash
# Kafka Connect REST API management

CONNECT_URL="http://localhost:8083"

echo "=== Kafka Connect Management ==="

# List all connectors
echo "--- Active Connectors ---"
curl -s "$CONNECT_URL/connectors" | jq .

# Get connector status
echo ""
echo "--- MySQL CDC Source Status ---"
curl -s "$CONNECT_URL/connectors/mysql-cdc-source/status" | jq .

# Create a new connector
echo ""
echo "--- Creating S3 Sink Connector ---"
curl -s -X POST "$CONNECT_URL/connectors" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "s3-sink",
    "config": {
      "connector.class": "io.confluent.connect.s3.S3SinkConnector",
      "s3.bucket.name": "my-kafka-data",
      "s3.region": "us-east-1",
      "topics": "enriched-orders",
      "tasks.max": "3",
      "storage.class": "io.confluent.connect.s3.storage.S3Storage",
      "format.class": "io.confluent.connect.s3.format.json.JsonFormat",
      "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
      "path.format": '\''YYYY/MM/dd'\'',
      "partition.duration.ms": "3600000",
      "locale": "en-US",
      "timezone": "UTC",
      "flush.size": "1000",
      "rotate.interval.ms": "60000",
      "schema.compatibility": "BACKWARD"
    }
  }' | jq .

# Restart a connector
echo ""
echo "--- Restarting MySQL CDC Source ---"
curl -s -X POST "$CONNECT_URL/connectors/mysql-cdc-source/restart" | jq .

# Delete a connector
echo ""
echo "--- Deleting S3 Sink Connector ---"
curl -s -X DELETE "$CONNECT_URL/connectors/s3-sink"

# Get connector config
echo ""
echo "--- Connector Configuration ---"
curl -s "$CONNECT_URL/connectors/mysql-cdc-source/config" | jq .

# Get connector tasks
echo ""
echo "--- Connector Tasks ---"
curl -s "$CONNECT_URL/connectors/mysql-cdc-source/tasks" | jq .

# Pause/Resume connector
echo ""
echo "--- Pausing Connector ---"
curl -s -X PUT "$CONNECT_URL/connectors/mysql-cdc-source/pause"

echo ""
echo "--- Resuming Connector ---"
curl -s -X PUT "$CONNECT_URL/connectors/mysql-cdc-source/resume"