Kafka Streams

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
    <version>3.6.0</version>
</dependency>

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.6.0</version>
</dependency>

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Produced;

import java.util.Arrays;
import java.util.Properties;

public class WordCountExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "word-count-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, 
                  Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, 
                  Serdes.String().getClass());

        StreamsBuilder builder = new StreamsBuilder();
        
        // Input stream
        KStream<String, String> textLines = builder.stream("text-input");
        
        // Processing pipeline
        KTable<String, Long> wordCounts = textLines
            .flatMapValues(textLine -> Arrays.asList(textLine.toLowerCase().split("\\W+")))
            .groupBy((key, word) -> word)
            .count();
        
        // Output stream
        wordCounts.toStream().to("word-count-output", Produced.with(Serdes.String(), Serdes.Long()));
        
        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
        
        // Shutdown hook
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}

import org.apache.kafka.streams.kstream.*;
import java.time.Duration;

public class FraudDetectionExample {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();
        
        // Input stream de transações
        KStream<String, Transaction> transactions = builder.stream("transactions");
        
        // Detectar múltiplas transações do mesmo usuário em janela de tempo
        KTable<Windowed<String>, Long> transactionCounts = transactions
            .groupBy((key, transaction) -> transaction.getUserId())
            .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
            .count();
        
        // Filtrar usuários com muitas transações (possível fraude)
        KStream<Windowed<String>, Long> suspiciousActivity = transactionCounts
            .toStream()
            .filter((windowedUserId, count) -> count > 10);
        
        // Enriquecer com dados do usuário
        KTable<String, User> users = builder.table("users");
        
        KStream<String, FraudAlert> fraudAlerts = transactions
            .join(users, 
                  (transaction, user) -> new TransactionWithUser(transaction, user),
                  Joined.with(Serdes.String(), transactionSerde, userSerde))
            .filter((key, txWithUser) -> isSuspicious(txWithUser))
            .mapValues(txWithUser -> new FraudAlert(
                txWithUser.getTransaction(),
                txWithUser.getUser(),
                "High frequency transactions detected"
            ));
        
        // Output para tópico de alertas
        fraudAlerts.to("fraud-alerts");
        
        KafkaStreams streams = new KafkaStreams(builder.build(), getProperties());
        streams.start();
    }
    
    private static boolean isSuspicious(TransactionWithUser txWithUser) {
        // Lógica de detecção de fraude
        return txWithUser.getTransaction().getAmount() > 10000 ||
               txWithUser.getUser().getRiskScore() > 0.8;
    }
}

import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.WindowedSerdes;
import java.time.Duration;

public class RealTimeAnalyticsExample {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();
        
        // Stream de eventos de cliques
        KStream<String, ClickEvent> clicks = builder.stream("click-events");
        
        // Agregações em janelas de tempo
        KTable<Windowed<String>, Long> clicksPerPage = clicks
            .groupBy((key, click) -> click.getPageId())
            .windowedBy(TimeWindows.of(Duration.ofMinutes(1)))
            .count();
        
        // Calcular taxa de cliques por sessão
        KTable<Windowed<String>, Double> clickRateBySession = clicks
            .groupBy((key, click) -> click.getSessionId())
            .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
            .aggregate(
                () -> new ClickStats(0, 0),
                (key, click, stats) -> {
                    stats.incrementClicks();
                    if (click.isConversion()) {
                        stats.incrementConversions();
                    }
                    return stats;
                },
                Materialized.with(Serdes.String(), clickStatsSerde)
            )
            .mapValues(stats -> stats.getConversionRate());
        
        // Stream de alertas para páginas com baixa performance
        KStream<String, Alert> performanceAlerts = clickRateBySession
            .toStream()
            .filter((windowedSession, rate) -> rate < 0.02)
            .map((windowedSession, rate) -> KeyValue.pair(
                windowedSession.key(),
                new Alert("Low conversion rate", rate, System.currentTimeMillis())
            ));
        
        // Output streams
        clicksPerPage.toStream().to("page-analytics");
        performanceAlerts.to("performance-alerts");
        
        KafkaStreams streams = new KafkaStreams(builder.build(), getProperties());
        streams.start();
    }
}

import org.apache.kafka.streams.state.QueryableStoreTypes;
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;

public class InteractiveQueriesExample {
    private KafkaStreams streams;
    
    public void setupStreams() {
        StreamsBuilder builder = new StreamsBuilder();
        
        // Criar uma KTable materializada
        KTable<String, Long> userClickCounts = builder
            .stream("user-clicks")
            .groupBy((key, click) -> click.getUserId())
            .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("user-click-counts"));
        
        streams = new KafkaStreams(builder.build(), getProperties());
        streams.start();
    }
    
    // Método para consultar o estado local
    public Long getUserClickCount(String userId) {
        ReadOnlyKeyValueStore<String, Long> store = streams.store(
            "user-click-counts",
            QueryableStoreTypes.keyValueStore()
        );
        
        return store.get(userId);
    }
    
    // REST endpoint para expor as consultas
    @GetMapping("/user/{userId}/clicks")
    public ResponseEntity<Long> getUserClicks(@PathVariable String userId) {
        Long clickCount = getUserClickCount(userId);
        return clickCount != null ? 
            ResponseEntity.ok(clickCount) : 
            ResponseEntity.notFound().build();
    }
}