WebFlux + Kafka + ZooKeeper Mastery: The Complete Guide to Reactive Streaming Architecture

public interface Publisher<T> {
    void subscribe(Subscriber<? super T> subscriber);
}

public interface Subscriber<T> {
    void onSubscribe(Subscription subscription);
    void onNext(T item);
    void onError(Throwable throwable);
    void onComplete();
}

public interface Subscription {
    void request(long n);
    void cancel();
}

public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
}

Producer (1000 msg/s) --> Buffer (overflows!) --> Consumer (100 msg/s)

With backpressure:
Producer (adjusts to 100 msg/s) <-- request(100) -- Consumer (100 msg/s)

Flux.range(1, Integer.MAX_VALUE)
    .onBackpressureBuffer(256)          // Buffer up to 256 items
    .onBackpressureDrop(dropped ->      // Drop when buffer full
        log.warn("Dropped: {}", dropped))
    .onBackpressureLatest()             // Keep only latest item
    .onBackpressureError()              // Signal error when overwhelmed
    .subscribe(item -> processSlowly(item));

Spring MVC (Thread-per-request):
  Thread-1: [Request] --> [DB Wait......] --> [Response]
  Thread-2: [Request] --> [API Wait.........] --> [Response]
  Thread-3: [Request] --> [File Wait....] --> [Response]
  (200 threads = 200 concurrent requests)

Spring WebFlux (Event Loop):
  EventLoop-1: [Req1][Req2][Req3][Req1-resume][Req4][Req2-resume]...
  EventLoop-2: [Req5][Req6][Req5-resume][Req7][Req8]...
  (4 threads = thousands of concurrent requests)

                    ┌─────────────────────────┐
                    │     Boss EventLoopGroup  │
                    │   (accepts connections)  │
                    └────────────┬────────────┘
                                 │
                    ┌────────────▼────────────┐
                    │   Worker EventLoopGroup  │
                    │   (handles I/O events)   │
                    ├─────────────────────────┤
                    │  EventLoop-1  Channel-A  │
                    │  EventLoop-2  Channel-B  │
                    │  EventLoop-3  Channel-C  │
                    │  EventLoop-4  Channel-D  │
                    └─────────────────────────┘

// Mono: single value or empty
Mono<User> findUser = userRepository.findById(userId);
Mono<Void> saveResult = userRepository.save(user);

// Flux: stream of values
Flux<Order> orders = orderRepository.findByUserId(userId);
Flux<Long> interval = Flux.interval(Duration.ofSeconds(1));
Flux<ServerSentEvent<String>> sseStream = Flux.create(sink -> {
    // push events to sink
});

// Static factory methods
Mono.just("hello");
Mono.empty();
Mono.error(new RuntimeException("failure"));
Mono.fromCallable(() -> expensiveComputation());
Mono.defer(() -> Mono.just(dynamicValue()));

Flux.just("a", "b", "c");
Flux.fromIterable(myList);
Flux.range(1, 100);
Flux.interval(Duration.ofMillis(500));
Flux.merge(flux1, flux2);
Flux.concat(flux1, flux2);
Flux.zip(flux1, flux2, (a, b) -> a + b);

// Transformation
flux.map(String::toUpperCase)
    .flatMap(name -> userService.findByName(name))   // async 1:N
    .concatMap(user -> enrichUser(user))              // sequential async
    .switchMap(query -> search(query))                // cancel previous
    .flatMapSequential(id -> fetchById(id))           // ordered parallel

// Filtering
flux.filter(user -> user.isActive())
    .distinct()
    .take(10)
    .skip(5)
    .takeUntil(event -> event.isTerminal())

// Error Handling
flux.onErrorReturn("default")
    .onErrorResume(ex -> fallbackFlux())
    .retry(3)
    .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
        .maxBackoff(Duration.ofSeconds(30))
        .filter(ex -> ex instanceof TransientException))
    .timeout(Duration.ofSeconds(5))

// Combining
Flux.merge(fastSource, slowSource)          // interleaved
Flux.concat(firstBatch, secondBatch)        // sequential
Flux.zip(names, ages, Person::new)          // paired
Mono.zip(userMono, profileMono, ordersMono) // parallel join
    .map(tuple -> buildResponse(tuple.getT1(), tuple.getT2(), tuple.getT3()))

// Side Effects (for logging/debugging only)
flux.doOnNext(item -> log.debug("Processing: {}", item))
    .doOnError(ex -> log.error("Error: {}", ex.getMessage()))
    .doOnComplete(() -> log.info("Stream completed"))
    .doOnSubscribe(sub -> log.info("Subscribed"))

@RestController
@RequestMapping("/api/users")
public class UserController {

    private final UserService userService;

    @GetMapping("/{id}")
    public Mono<ResponseEntity<User>> getUser(@PathVariable String id) {
        return userService.findById(id)
            .map(ResponseEntity::ok)
            .defaultIfEmpty(ResponseEntity.notFound().build());
    }

    @GetMapping(produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<User> streamUsers() {
        return userService.streamAllUsers();
    }

    @PostMapping
    @ResponseStatus(HttpStatus.CREATED)
    public Mono<User> createUser(@Valid @RequestBody Mono<User> userMono) {
        return userMono.flatMap(userService::create);
    }
}

@Configuration
public class UserRouter {

    @Bean
    public RouterFunction<ServerResponse> userRoutes(UserHandler handler) {
        return RouterFunctions.route()
            .path("/api/users", builder -> builder
                .GET("/{id}", handler::getUser)
                .GET("", handler::listUsers)
                .POST("", handler::createUser)
                .PUT("/{id}", handler::updateUser)
                .DELETE("/{id}", handler::deleteUser))
            .build();
    }
}

@Component
public class UserHandler {

    private final UserService userService;

    public Mono<ServerResponse> getUser(ServerRequest request) {
        String id = request.pathVariable("id");
        return userService.findById(id)
            .flatMap(user -> ServerResponse.ok()
                .contentType(MediaType.APPLICATION_JSON)
                .bodyValue(user))
            .switchIfEmpty(ServerResponse.notFound().build());
    }

    public Mono<ServerResponse> listUsers(ServerRequest request) {
        return ServerResponse.ok()
            .contentType(MediaType.APPLICATION_JSON)
            .body(userService.findAll(), User.class);
    }
}

@Configuration
@EnableR2dbcRepositories
public class DatabaseConfig extends AbstractR2dbcConfiguration {

    @Override
    @Bean
    public ConnectionFactory connectionFactory() {
        return ConnectionFactories.get(ConnectionFactoryOptions.builder()
            .option(DRIVER, "postgresql")
            .option(HOST, "localhost")
            .option(PORT, 5432)
            .option(DATABASE, "mydb")
            .option(USER, "user")
            .option(PASSWORD, "password")
            .option(MAX_SIZE, 20)
            .build());
    }
}

public interface UserRepository extends ReactiveCrudRepository<User, String> {

    Flux<User> findByStatusOrderByCreatedAtDesc(String status);

    @Query("SELECT * FROM users WHERE email = :email")
    Mono<User> findByEmail(String email);

    @Query("SELECT * FROM users WHERE department = :dept")
    Flux<User> findByDepartment(String dept);
}

@Repository
public class CustomUserRepository {

    private final DatabaseClient databaseClient;

    public Flux<User> searchUsers(String keyword, int limit) {
        return databaseClient.sql(
                "SELECT * FROM users WHERE name ILIKE :keyword LIMIT :limit")
            .bind("keyword", "%" + keyword + "%")
            .bind("limit", limit)
            .map(row -> new User(
                row.get("id", String.class),
                row.get("name", String.class),
                row.get("email", String.class)
            ))
            .all();
    }
}

┌─────────────────────────────────────────────────────────────┐
│                     Kafka Cluster                           │
│                                                             │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐                  │
│  │ Broker 1 │  │ Broker 2 │  │ Broker 3 │                  │
│  │          │  │          │  │          │                  │
│  │ Topic-A  │  │ Topic-A  │  │ Topic-A  │                  │
│  │ Part-0   │  │ Part-1   │  │ Part-2   │                  │
│  │ (leader) │  │ (leader) │  │ (leader) │                  │
│  │          │  │          │  │          │                  │
│  │ Topic-A  │  │ Topic-A  │  │ Topic-A  │                  │
│  │ Part-1   │  │ Part-2   │  │ Part-0   │                  │
│  │ (replica)│  │ (replica)│  │ (replica)│                  │
│  └──────────┘  └──────────┘  └──────────┘                  │
│                                                             │
│  ┌──────────────────────────────────────────────────────┐   │
│  │  Metadata: ZooKeeper (legacy) or KRaft (Kafka 3.5+) │   │
│  └──────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────┘

@Configuration
public class KafkaProducerConfig {

    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka1:9092,kafka2:9092,kafka3:9092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

        // Reliability settings
        config.put(ProducerConfig.ACKS_CONFIG, "all");           // Wait for all ISR replicas
        config.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
        config.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
        config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

        // Performance tuning
        config.put(ProducerConfig.BATCH_SIZE_CONFIG, 32768);     // 32KB batch
        config.put(ProducerConfig.LINGER_MS_CONFIG, 20);         // Wait 20ms for batching
        config.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "zstd");
        config.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 67108864); // 64MB buffer

        return new DefaultKafkaProducerFactory<>(config);
    }

    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

@Service
@RequiredArgsConstructor
public class OrderEventProducer {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    // Fire-and-forget (highest throughput)
    public void sendFireAndForget(OrderEvent event) {
        kafkaTemplate.send("orders", event.getOrderId(), event);
    }

    // Synchronous send (highest reliability)
    public void sendSync(OrderEvent event) throws Exception {
        kafkaTemplate.send("orders", event.getOrderId(), event)
            .get(10, TimeUnit.SECONDS);
    }

    // Asynchronous send with callback (balanced)
    public CompletableFuture<SendResult<String, Object>> sendAsync(OrderEvent event) {
        return kafkaTemplate.send("orders", event.getOrderId(), event)
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("Failed to send order event: {}", event.getOrderId(), ex);
                } else {
                    log.info("Order event sent to partition {} offset {}",
                        result.getRecordMetadata().partition(),
                        result.getRecordMetadata().offset());
                }
            });
    }

    // Send with headers
    public void sendWithHeaders(OrderEvent event) {
        ProducerRecord<String, Object> record = new ProducerRecord<>("orders",
            null, null, event.getOrderId(), event);
        record.headers()
            .add("event-type", "ORDER_CREATED".getBytes())
            .add("source", "order-service".getBytes())
            .add("correlation-id", UUID.randomUUID().toString().getBytes());
        kafkaTemplate.send(record);
    }
}

@Configuration
@EnableKafka
public class KafkaConsumerConfig {

    @Bean
    public ConsumerFactory<String, Object> consumerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka1:9092,kafka2:9092");
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processing-group");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);

        // Offset management
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); // Manual commit

        // Performance tuning
        config.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        config.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1024);
        config.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500);
        config.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 1048576); // 1MB

        return new DefaultKafkaConsumerFactory<>(config);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        factory.setCommonErrorHandler(new DefaultErrorHandler(
            new DeadLetterPublishingRecoverer(kafkaTemplate()),
            new FixedBackOff(1000L, 3)
        ));
        return factory;
    }
}

@Component
@RequiredArgsConstructor
public class OrderEventConsumer {

    private final OrderService orderService;

    // Single record processing
    @KafkaListener(topics = "orders", groupId = "order-processing-group")
    public void processOrder(ConsumerRecord<String, OrderEvent> record,
                             Acknowledgment ack) {
        try {
            log.info("Processing order: {} from partition: {} offset: {}",
                record.value().getOrderId(),
                record.partition(),
                record.offset());
            orderService.process(record.value());
            ack.acknowledge();
        } catch (Exception e) {
            log.error("Failed to process order: {}", record.value().getOrderId(), e);
            // Don't acknowledge — will be retried by error handler
        }
    }

    // Batch processing
    @KafkaListener(topics = "analytics-events", groupId = "analytics-batch-group",
                   containerFactory = "batchKafkaListenerContainerFactory")
    public void processBatch(List<ConsumerRecord<String, AnalyticsEvent>> records,
                             Acknowledgment ack) {
        log.info("Processing batch of {} records", records.size());
        try {
            List<AnalyticsEvent> events = records.stream()
                .map(ConsumerRecord::value)
                .collect(Collectors.toList());
            analyticsService.bulkInsert(events);
            ack.acknowledge();
        } catch (Exception e) {
            log.error("Batch processing failed", e);
        }
    }

    // Filtering with headers
    @KafkaListener(topics = "orders",
                   filter = "orderCreatedFilter",
                   groupId = "notification-group")
    public void handleOrderCreated(OrderEvent event) {
        notificationService.sendOrderConfirmation(event);
    }
}

@Configuration
public class KafkaTransactionalConfig {

    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
        config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        config.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-tx-");
        config.put(ProducerConfig.ACKS_CONFIG, "all");
        return new DefaultKafkaProducerFactory<>(config);
    }

    @Bean
    public KafkaTransactionManager<String, Object> kafkaTransactionManager() {
        return new KafkaTransactionManager<>(producerFactory());
    }
}

@Service
@RequiredArgsConstructor
public class TransactionalOrderService {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    @Transactional("kafkaTransactionManager")
    public void processOrderTransactionally(OrderEvent order) {
        // All messages in this method are sent atomically
        kafkaTemplate.send("orders-processed", order.getOrderId(), order);
        kafkaTemplate.send("inventory-updates", order.getProductId(),
            new InventoryEvent(order.getProductId(), -order.getQuantity()));
        kafkaTemplate.send("payment-requests", order.getOrderId(),
            new PaymentEvent(order.getOrderId(), order.getTotalAmount()));
        // If any send fails, all are rolled back
    }
}

@Configuration
@EnableKafkaStreams
public class KafkaStreamsConfig {

    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    public KafkaStreamsConfiguration kafkaStreamsConfiguration() {
        Map<String, Object> config = new HashMap<>();
        config.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-analytics");
        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
        config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        config.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
        config.put(StreamsConfig.STATE_DIR_CONFIG, "/var/kafka-streams");
        return new KafkaStreamsConfiguration(config);
    }
}

@Component
public class OrderAnalyticsTopology {

    @Autowired
    public void buildPipeline(StreamsBuilder streamsBuilder) {
        // Real-time order analytics pipeline
        KStream<String, OrderEvent> orders = streamsBuilder
            .stream("orders", Consumed.with(Serdes.String(), orderEventSerde()));

        // 1. Order count per product in tumbling windows
        KTable<Windowed<String>, Long> productCounts = orders
            .groupBy((key, order) -> order.getProductId(),
                Grouped.with(Serdes.String(), orderEventSerde()))
            .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
            .count(Materialized.as("product-counts-store"));

        // 2. Revenue aggregation per category
        KTable<String, Double> revenueByCategory = orders
            .groupBy((key, order) -> order.getCategory(),
                Grouped.with(Serdes.String(), orderEventSerde()))
            .aggregate(
                () -> 0.0,
                (key, order, total) -> total + order.getTotalAmount(),
                Materialized.with(Serdes.String(), Serdes.Double())
            );

        // 3. Join with customer data for enrichment
        KTable<String, CustomerProfile> customers = streamsBuilder
            .table("customer-profiles",
                Consumed.with(Serdes.String(), customerProfileSerde()));

        KStream<String, EnrichedOrder> enrichedOrders = orders
            .selectKey((key, order) -> order.getCustomerId())
            .join(customers,
                (order, customer) -> new EnrichedOrder(order, customer),
                Joined.with(Serdes.String(), orderEventSerde(), customerProfileSerde()));

        enrichedOrders.to("enriched-orders",
            Produced.with(Serdes.String(), enrichedOrderSerde()));

        // 4. Fraud detection with session windows
        orders
            .filter((key, order) -> order.getTotalAmount() > 1000)
            .groupByKey()
            .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(10)))
            .count()
            .toStream()
            .filter((windowedKey, count) -> count >= 3)
            .mapValues((key, count) -> "ALERT: " + count + " high-value orders in session")
            .to("fraud-alerts");
    }
}

{
  "name": "postgresql-source-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "password",
    "database.dbname": "orders_db",
    "database.server.name": "order-server",
    "table.include.list": "public.orders,public.order_items",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_slot",
    "publication.name": "dbz_publication",
    "topic.prefix": "cdc",
    "transforms": "route",
    "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
    "transforms.route.replacement": "cdc-$3"
  }
}

{
  "name": "elasticsearch-sink-connector",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "connection.url": "http://elasticsearch:9200",
    "topics": "enriched-orders",
    "type.name": "_doc",
    "key.ignore": false,
    "schema.ignore": true,
    "behavior.on.null.values": "delete",
    "write.method": "upsert",
    "batch.size": 200,
    "max.buffered.records": 5000,
    "flush.timeout.ms": 10000
  }
}

// Producer with Avro schema
@Configuration
public class AvroProducerConfig {

    @Bean
    public ProducerFactory<String, GenericRecord> avroProducerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
            io.confluent.kafka.serializers.KafkaAvroSerializer.class);
        config.put("schema.registry.url", "http://schema-registry:8081");
        config.put("auto.register.schemas", true);
        return new DefaultKafkaProducerFactory<>(config);
    }
}

ZooKeeper Mode:                        KRaft Mode:
┌────────┐    ┌────────────┐           ┌──────────────────────┐
│ Broker │───▶│ ZooKeeper  │           │ Controller (Raft)    │
│ Broker │───▶│ Ensemble   │           │  Node 1 (leader)     │
│ Broker │───▶│ (3-5 nodes)│           │  Node 2 (follower)   │
└────────┘    └────────────┘           │  Node 3 (follower)   │
                                       └──────────┬───────────┘
                                                  │
                                       ┌──────────▼───────────┐
                                       │ Brokers              │
                                       │  Broker 1            │
                                       │  Broker 2            │
                                       │  Broker 3            │
                                       └──────────────────────┘

# Verify Kafka version (must be 3.5+)
kafka-broker-api-versions.sh --bootstrap-server localhost:9092 | head -1

# Check current ZooKeeper metadata
kafka-metadata.sh --snapshot /var/kafka-logs/__cluster_metadata-0/00000000000000000000.log \
  --cluster-id $(kafka-storage.sh random-uuid)

# Verify no deprecated features in use
kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type brokers

# kraft/server.properties for controller nodes
process.roles=controller
node.id=1
controller.quorum.voters=1@controller1:9093,2@controller2:9093,3@controller3:9093
controller.listener.names=CONTROLLER
listeners=CONTROLLER://:9093
log.dirs=/var/kraft-controller-logs

# Format controller storage
kafka-storage.sh format \
  --config kraft/server.properties \
  --cluster-id $(kafka-storage.sh random-uuid) \
  --release-version 3.7

# Start migration (ZooKeeper mode -> dual mode)
kafka-metadata.sh --migrate \
  --zookeeper-connect zk1:2181,zk2:2181,zk3:2181 \
  --bootstrap-server kafka1:9092

# Verify migration status
kafka-metadata.sh --status \
  --bootstrap-server kafka1:9092

# Finalize (point of no return)
kafka-metadata.sh --finalize \
  --bootstrap-server kafka1:9092

# Verify all topics are accessible
kafka-topics.sh --bootstrap-server localhost:9092 --list

# Verify consumer groups
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

# Check controller status
kafka-metadata.sh --controllers \
  --bootstrap-server localhost:9092

@RestController
@RequestMapping("/api/stream")
public class SSEController {

    private final StockPriceService stockPriceService;

    @GetMapping(value = "/prices", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<StockPrice>> streamPrices(
            @RequestParam List<String> symbols) {
        return stockPriceService.getPriceStream(symbols)
            .map(price -> ServerSentEvent.<StockPrice>builder()
                .id(String.valueOf(price.getTimestamp()))
                .event("price-update")
                .data(price)
                .retry(Duration.ofSeconds(5))
                .build());
    }

    @GetMapping(value = "/heartbeat", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<String>> heartbeat() {
        return Flux.interval(Duration.ofSeconds(30))
            .map(seq -> ServerSentEvent.<String>builder()
                .event("heartbeat")
                .data("ping")
                .build());
    }
}

const eventSource = new EventSource('/api/stream/prices?symbols=AAPL,GOOG,MSFT')

eventSource.addEventListener('price-update', (event) => {
  const price = JSON.parse(event.data)
  updateUI(price)
})

eventSource.addEventListener('heartbeat', () => {
  console.log('Connection alive')
})

eventSource.onerror = (error) => {
  console.error('SSE error:', error)
  // Browser automatically reconnects
}

@Configuration
public class WebSocketConfig {

    @Bean
    public HandlerMapping webSocketMapping(StockWebSocketHandler handler) {
        Map<String, WebSocketHandler> map = new HashMap<>();
        map.put("/ws/stocks", handler);
        SimpleUrlHandlerMapping mapping = new SimpleUrlHandlerMapping();
        mapping.setUrlMap(map);
        mapping.setOrder(-1);
        return mapping;
    }

    @Bean
    public WebSocketHandlerAdapter handlerAdapter() {
        return new WebSocketHandlerAdapter();
    }
}

@Component
@RequiredArgsConstructor
public class StockWebSocketHandler implements WebSocketHandler {

    private final StockPriceService stockPriceService;
    private final ObjectMapper objectMapper;

    @Override
    public Mono<Void> handle(WebSocketSession session) {
        // Incoming messages (subscriptions)
        Flux<String> incoming = session.receive()
            .map(WebSocketMessage::getPayloadAsText)
            .doOnNext(msg -> log.info("Received subscription: {}", msg));

        // Outgoing price updates
        Flux<WebSocketMessage> outgoing = incoming
            .flatMap(msg -> {
                SubscriptionRequest req = parseRequest(msg);
                return stockPriceService.getPriceStream(req.getSymbols());
            })
            .map(price -> {
                try {
                    String json = objectMapper.writeValueAsString(price);
                    return session.textMessage(json);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            });

        return session.send(outgoing);
    }
}

const ws = new WebSocket('ws://localhost:8080/ws/stocks')

ws.onopen = () => {
  ws.send(
    JSON.stringify({
      action: 'subscribe',
      symbols: ['AAPL', 'GOOG', 'MSFT'],
    })
  )
}

ws.onmessage = (event) => {
  const price = JSON.parse(event.data)
  updateUI(price)
}

ws.onclose = (event) => {
  console.log('WebSocket closed:', event.code, event.reason)
  // Manual reconnection needed
  setTimeout(reconnect, 3000)
}

syntax = "proto3";

package stocks;

service StockService {
    // Unary
    rpc GetPrice (PriceRequest) returns (StockPrice);

    // Server streaming
    rpc StreamPrices (PriceStreamRequest) returns (stream StockPrice);

    // Client streaming
    rpc RecordTrades (stream TradeRecord) returns (TradeSummary);

    // Bidirectional streaming
    rpc LiveTrading (stream TradeOrder) returns (stream TradeConfirmation);
}

message PriceStreamRequest {
    repeated string symbols = 1;
    int32 interval_ms = 2;
}

message StockPrice {
    string symbol = 1;
    double price = 2;
    double change = 3;
    int64 timestamp = 4;
    int64 volume = 5;
}

@GrpcService
public class StockGrpcService extends StockServiceGrpc.StockServiceImplBase {

    private final StockPriceService stockPriceService;

    @Override
    public void streamPrices(PriceStreamRequest request,
                             StreamObserver<StockPrice> responseObserver) {
        stockPriceService.getPriceStream(request.getSymbolsList())
            .subscribe(
                price -> responseObserver.onNext(toProto(price)),
                error -> responseObserver.onError(
                    Status.INTERNAL.withDescription(error.getMessage()).asException()),
                responseObserver::onCompleted
            );
    }

    @Override
    public StreamObserver<TradeOrder> liveTrading(
            StreamObserver<TradeConfirmation> responseObserver) {
        return new StreamObserver<TradeOrder>() {
            @Override
            public void onNext(TradeOrder order) {
                TradeConfirmation confirmation = executeTrade(order);
                responseObserver.onNext(confirmation);
            }

            @Override
            public void onError(Throwable t) {
                log.error("Client error in live trading", t);
            }

            @Override
            public void onCompleted() {
                responseObserver.onCompleted();
            }
        };
    }
}

@Configuration
public class ReactorKafkaConfig {

    @Bean
    public ReceiverOptions<String, String> receiverOptions() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "reactive-consumer-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

        return ReceiverOptions.<String, String>create(props)
            .subscription(Collections.singleton("events"))
            .addAssignListener(partitions ->
                log.info("Assigned: {}", partitions))
            .addRevokeListener(partitions ->
                log.info("Revoked: {}", partitions))
            .commitInterval(Duration.ofSeconds(5))
            .commitBatchSize(100);
    }

    @Bean
    public SenderOptions<String, String> senderOptions() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

        return SenderOptions.create(props);
    }

    @Bean
    public KafkaReceiver<String, String> kafkaReceiver() {
        return KafkaReceiver.create(receiverOptions());
    }

    @Bean
    public KafkaSender<String, String> kafkaSender() {
        return KafkaSender.create(senderOptions());
    }
}

@Service
@RequiredArgsConstructor
public class ReactiveEventProcessor {

    private final KafkaReceiver<String, String> kafkaReceiver;
    private final EventService eventService;

    @PostConstruct
    public void startConsuming() {
        kafkaReceiver.receive()
            .groupBy(record -> record.receiverOffset().topicPartition())
            .flatMap(partitionFlux -> partitionFlux
                .publishOn(Schedulers.boundedElastic())
                .concatMap(record -> processRecord(record)
                    .doOnSuccess(v -> record.receiverOffset().acknowledge())
                    .onErrorResume(e -> {
                        log.error("Error processing record: {}", record.key(), e);
                        record.receiverOffset().acknowledge();
                        return Mono.empty();
                    })
                ))
            .subscribe();
    }

    private Mono<Void> processRecord(ReceiverRecord<String, String> record) {
        return eventService.process(record.value())
            .doOnNext(result ->
                log.debug("Processed: partition={} offset={} key={}",
                    record.partition(), record.offset(), record.key()));
    }
}

@Service
@RequiredArgsConstructor
public class ReactiveEventProducer {

    private final KafkaSender<String, String> kafkaSender;
    private final ObjectMapper objectMapper;

    public Mono<Void> sendEvent(String topic, String key, Object event) {
        return Mono.fromCallable(() -> objectMapper.writeValueAsString(event))
            .flatMap(json -> {
                SenderRecord<String, String, String> record = SenderRecord.create(
                    new ProducerRecord<>(topic, key, json), key);
                return kafkaSender.send(Mono.just(record))
                    .next()
                    .doOnNext(result -> log.info("Sent to {}-{} offset {}",
                        result.recordMetadata().topic(),
                        result.recordMetadata().partition(),
                        result.recordMetadata().offset()))
                    .then();
            });
    }

    public Flux<SenderResult<String>> sendBatch(String topic, List<Event> events) {
        Flux<SenderRecord<String, String, String>> records = Flux.fromIterable(events)
            .map(event -> {
                String json = objectMapper.writeValueAsString(event);
                return SenderRecord.create(
                    new ProducerRecord<>(topic, event.getId(), json), event.getId());
            });

        return kafkaSender.send(records)
            .doOnError(e -> log.error("Batch send failed", e));
    }
}

┌─────────────────────────────────────────────────────────┐
│                   CQRS + Event Sourcing                 │
│                                                         │
│  Command Side                    Query Side             │
│  ┌──────────┐                    ┌──────────┐           │
│  │ WebFlux  │                    │ WebFlux  │           │
│  │ Command  │                    │ Query    │           │
│  │ Handler  │                    │ Handler  │           │
│  └────┬─────┘                    └────▲─────┘           │
│       │                               │                 │
│  ┌────▼─────┐                    ┌────┴─────┐           │
│  │ Domain   │                    │ Read     │           │
│  │ Aggregate│                    │ Model    │           │
│  └────┬─────┘                    │ (R2DBC)  │           │
│       │                          └────▲─────┘           │
│  ┌────▼─────────────────────────────┐ │                 │
│  │         Kafka (Event Store)      ├─┘                 │
│  │  orders.events topic             │                   │
│  └──────────────────────────────────┘                   │
└─────────────────────────────────────────────────────────┘

@RestController
@RequestMapping("/api/orders")
@RequiredArgsConstructor
public class OrderCommandController {

    private final OrderCommandService commandService;

    @PostMapping
    @ResponseStatus(HttpStatus.ACCEPTED)
    public Mono<OrderId> createOrder(@RequestBody Mono<CreateOrderCommand> command) {
        return command.flatMap(commandService::handle);
    }

    @PostMapping("/{orderId}/confirm")
    public Mono<ResponseEntity<Void>> confirmOrder(@PathVariable String orderId) {
        return commandService.handle(new ConfirmOrderCommand(orderId))
            .then(Mono.just(ResponseEntity.accepted().build()));
    }
}

@Service
@RequiredArgsConstructor
public class OrderCommandService {

    private final KafkaSender<String, String> kafkaSender;
    private final ObjectMapper objectMapper;

    public Mono<OrderId> handle(CreateOrderCommand command) {
        // Validate command
        OrderAggregate aggregate = OrderAggregate.create(command);
        List<DomainEvent> events = aggregate.getUncommittedEvents();

        // Publish events to Kafka
        return Flux.fromIterable(events)
            .flatMap(event -> publishEvent("order-events", aggregate.getId(), event))
            .then(Mono.just(new OrderId(aggregate.getId())));
    }

    private Mono<Void> publishEvent(String topic, String key, DomainEvent event) {
        return Mono.fromCallable(() -> objectMapper.writeValueAsString(event))
            .flatMap(json -> {
                ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, json);
                record.headers().add("event-type", event.getType().getBytes());
                return kafkaSender.send(Mono.just(SenderRecord.create(record, key)))
                    .next()
                    .then();
            });
    }
}

@Service
@RequiredArgsConstructor
public class OrderProjectionService {

    private final KafkaReceiver<String, String> kafkaReceiver;
    private final OrderReadRepository readRepository;
    private final ObjectMapper objectMapper;

    @PostConstruct
    public void startProjection() {
        kafkaReceiver.receive()
            .concatMap(record -> {
                String eventType = new String(
                    record.headers().lastHeader("event-type").value());
                return processEvent(eventType, record.value())
                    .doOnSuccess(v -> record.receiverOffset().acknowledge());
            })
            .subscribe();
    }

    private Mono<Void> processEvent(String eventType, String payload) {
        return switch (eventType) {
            case "OrderCreated" -> {
                OrderCreatedEvent event = objectMapper.readValue(payload, OrderCreatedEvent.class);
                yield readRepository.save(OrderReadModel.from(event)).then();
            }
            case "OrderConfirmed" -> {
                OrderConfirmedEvent event = objectMapper.readValue(payload, OrderConfirmedEvent.class);
                yield readRepository.findById(event.getOrderId())
                    .map(model -> model.withStatus("CONFIRMED"))
                    .flatMap(readRepository::save)
                    .then();
            }
            default -> Mono.empty();
        };
    }
}

┌──────────────────────────────────────────────────────────────────┐
│                 Real-Time Stock Ticker System                    │
│                                                                  │
│  ┌─────────────┐     ┌──────────────┐     ┌─────────────────┐   │
│  │ Market Data │     │   Kafka      │     │   WebFlux       │   │
│  │ Providers   │────▶│   Cluster    │────▶│   API Gateway   │   │
│  │ (WebSocket) │     │              │     │                 │   │
│  └─────────────┘     │ raw-prices   │     │  SSE /prices    │   │
│                      │ processed    │     │  WS  /ws/trade  │   │
│  ┌─────────────┐     │ alerts       │     │  gRPC /internal │   │
│  │ News Feed   │────▶│              │     └───────┬─────────┘   │
│  │ (REST API)  │     └──────┬───────┘             │             │
│  └─────────────┘            │              ┌──────▼─────────┐   │
│                      ┌──────▼───────┐      │   Clients      │   │
│                      │ Kafka Streams│      │  Web Dashboard │   │
│                      │ Processing   │      │  Mobile App    │   │
│                      │              │      │  Trading Bot   │   │
│                      │ - VWAP calc  │      └────────────────┘   │
│                      │ - Anomaly    │                           │
│                      │ - Aggregate  │      ┌────────────────┐   │
│                      └──────┬───────┘      │   PostgreSQL   │   │
│                             │              │   (R2DBC)      │   │
│                             └─────────────▶│   + Redis      │   │
│                                            └────────────────┘   │
└──────────────────────────────────────────────────────────────────┘

@Service
@RequiredArgsConstructor
@Slf4j
public class MarketDataIngestionService {

    private final KafkaSender<String, String> kafkaSender;
    private final ObjectMapper objectMapper;
    private final WebSocketClient webSocketClient;

    @PostConstruct
    public void connect() {
        URI uri = URI.create("wss://market-data-provider.example.com/stream");

        webSocketClient.execute(uri, session -> {
            // Subscribe to symbols
            Mono<Void> subscribe = session.send(
                Mono.just(session.textMessage(
                    "{\"action\":\"subscribe\",\"symbols\":[\"AAPL\",\"GOOG\",\"MSFT\",\"AMZN\"]}"
                ))
            );

            // Process incoming market data
            Flux<Void> receive = session.receive()
                .map(WebSocketMessage::getPayloadAsText)
                .flatMap(this::publishToKafka)
                .doOnError(e -> log.error("Market data stream error", e));

            return subscribe.thenMany(receive).then();
        })
        .retryWhen(Retry.backoff(Long.MAX_VALUE, Duration.ofSeconds(1))
            .maxBackoff(Duration.ofMinutes(5)))
        .subscribe();
    }

    private Mono<Void> publishToKafka(String rawData) {
        try {
            MarketTick tick = objectMapper.readValue(rawData, MarketTick.class);
            ProducerRecord<String, String> record = new ProducerRecord<>(
                "raw-prices", tick.getSymbol(), rawData);
            record.headers()
                .add("source", "market-data-provider".getBytes())
                .add("timestamp", String.valueOf(tick.getTimestamp()).getBytes());

            return kafkaSender.send(Mono.just(SenderRecord.create(record, tick.getSymbol())))
                .next()
                .then();
        } catch (Exception e) {
            log.error("Failed to parse market data: {}", rawData, e);
            return Mono.empty();
        }
    }
}

@Component
public class PriceProcessingTopology {

    @Autowired
    public void buildPipeline(StreamsBuilder builder) {
        KStream<String, String> rawPrices = builder.stream("raw-prices");

        // 1. Parse and validate
        KStream<String, StockPrice> validPrices = rawPrices
            .mapValues(this::parsePrice)
            .filter((symbol, price) -> price != null && price.getPrice() > 0);

        // 2. Calculate VWAP (Volume Weighted Average Price) per 1-minute window
        KTable<Windowed<String>, VWAPResult> vwap = validPrices
            .groupByKey()
            .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1)))
            .aggregate(
                VWAPAccumulator::new,
                (symbol, price, acc) -> acc.add(price.getPrice(), price.getVolume()),
                Materialized.<String, VWAPAccumulator, WindowStore<Bytes, byte[]>>
                    as("vwap-store")
                    .withValueSerde(vwapAccumulatorSerde())
            )
            .mapValues(VWAPAccumulator::result);

        // 3. Anomaly detection (price change > 5% in 30 seconds)
        validPrices
            .groupByKey()
            .windowedBy(SlidingWindows.ofTimeDifferenceWithNoGrace(Duration.ofSeconds(30)))
            .aggregate(
                PriceRange::new,
                (symbol, price, range) -> range.update(price.getPrice()),
                Materialized.with(Serdes.String(), priceRangeSerde())
            )
            .toStream()
            .filter((key, range) -> range.percentChange() > 5.0)
            .mapValues((key, range) -> new PriceAlert(
                key.key(), range.percentChange(), "ANOMALY"))
            .to("alerts", Produced.with(windowedSerde(), priceAlertSerde()));

        // 4. Output processed prices
        validPrices
            .mapValues(price -> new ProcessedPrice(
                price.getSymbol(), price.getPrice(), price.getVolume(),
                price.getTimestamp(), System.currentTimeMillis()))
            .to("processed-prices", Produced.with(Serdes.String(), processedPriceSerde()));
    }
}