Forem: Roberto de Vargas Neto

The Ecosystem Conductor: Orchestrating Market Orders with Kafka, RabbitMQ and Spring Boot

Roberto de Vargas Neto — Sat, 02 May 2026 00:02:45 +0000

Hey folks!

We've reached one of the most important posts in the My Broker B3 series. Up until now we built each piece of the ecosystem separately — the price synchronizer, the matching engine, the financial custody, the asset catalog. Now it's time to build the service that connects all of it: the Broker Order API.

This is the conductor of the ecosystem. When a user wants to buy or sell a stock, this is where everything begins — and where the results come back to.

🏦 What is the Order API?

The trading-broker-order is the entry point for the broker's transactional flow. It has a clear responsibility: orchestrate the complete lifecycle of an order, from the user's intent to the final execution confirmation.

To do this, it communicates with four other ecosystem services:

Asset API (REST/Feign) — validates whether the ticker exists and is active
Wallet API (REST/Feign) — validates whether the user has sufficient balance
B3 Matching Engine (RabbitMQ) — sends the order for execution
Broker Wallet (Kafka) — notifies the ecosystem about each status change

🗺️ The Complete Order Flow

Before diving into the code, here's the end-to-end flow:

[User]
    │
    │  POST /api/v1/orders
    │  { userId, ticker, quantity, price, side: "BUY" }
    ▼
[trading-broker-order]
    │
    ├─ 1. Validate ticker  → GET trading-asset/assets/{ticker} (Feign/REST)
    │
    ├─ 2. Validate balance → GET trading-wallet/summary (Feign/REST)
    │
    ├─ 3. Persist order → MySQL (status: PENDING)
    │
    ├─ 4. Send to B3 → RabbitMQ (mq-broker-to-b3)
    │
    ├─ 5. Notify ecosystem → Kafka (order-events-v1, status: PENDING)
    │
    └─ Returns created order → HTTP 201

         ─ ─ ─ (B3 processes the order) ─ ─ ─

[b3-matching-engine]
    │
    │  RabbitMQ (mq-b3-to-broker)
    │  { orderId, status: FILLED/REJECTED, executedPrice }
    ▼
[trading-broker-order]
    │
    ├─ 6. Update status → MySQL (FILLED or REJECTED)
    │
    └─ 7. Notify ecosystem → Kafka (order-events-v1, final status)
              │
              ├─ trading-broker-wallet consumes → blocks/settles/refunds balance
              └─ other future consumers

Two distinct communication channels, two Kafka publish moments, and the wallet reacting to each state change. Let's see how each part was implemented.

🛠️ Tech Stack

Technology	Usage
Java 21 + Spring Boot 3.3.5	Service core
MySQL + Flyway	Order persistence
Spring Kafka	Lifecycle event publishing
Spring AMQP	Integration with Matching Engine via RabbitMQ
Spring Cloud OpenFeign	REST calls to Asset API and Wallet API
SpringDoc OpenAPI	Swagger UI documentation

🏗️ Implementation Pillars

1. The Domain Model — Order

The Order entity represents the state of an order at any point in its lifecycle:

@Entity
@Table(name = "orders")
public class Order {

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    private String userId;
    private String ticker;

    @Column(nullable = false, precision = 19, scale = 4)
    private BigDecimal quantity;

    @Column(nullable = false, precision = 19, scale = 4)
    private BigDecimal price;

    // Null while PENDING — only set after B3 feedback
    @Column(precision = 19, scale = 4)
    private BigDecimal executedPrice;

    @Enumerated(EnumType.STRING)
    private OrderSide side;   // BUY | SELL

    @Enumerated(EnumType.STRING)
    private OrderStatus status; // PENDING | FILLED | REJECTED | CANCELLED

    @CreationTimestamp
    private LocalDateTime createdAt;

    @UpdateTimestamp
    private LocalDateTime updatedAt;
}

An important design detail: executedPrice is intentionally nullable. When the order is created, we don't yet know the execution price — that only comes as a response from the Matching Engine. Forcing nullable = false here would be a domain modeling error.

The entity also carries the mapToEvent() method that serializes the current state into a Kafka event:

public OrderEventDTO mapToEvent() {
    return OrderEventDTO.builder()
            .orderId(this.getId())
            .userId(this.getUserId())
            .ticker(this.getTicker())
            .quantity(this.getQuantity())
            .price(this.getPrice())
            .executedPrice(this.getExecutedPrice())
            .side(this.getSide().name())
            .status(this.getStatus().name())
            .eventTimestamp(LocalDateTime.now())
            .build();
}

2. Ticker Validation — Asset Feign Client

Before validating balance or persisting anything, we need to ensure the requested ticker actually exists and is available for trading. A user submitting an order for XPTO999 should get a clear error immediately — not a silent rejection from B3 two hops later.

@FeignClient(name = "trading-broker-asset", url = "${app.services.asset-url}")
public interface AssetClient {

    @GetMapping("/api/v1/assets/{ticker}")
    AssetDTO getByTicker(@PathVariable("ticker") String ticker);
}

The validation in OrderService:

private void validateTicker(String ticker) {
    try {
        var asset = assetClient.getByTicker(ticker.toUpperCase());
        if (!"ACTIVE".equalsIgnoreCase(asset.status())) {
            throw new RuntimeException("Asset is not available for trading: " + ticker);
        }
        log.info("Ticker {} validated — status: {}", ticker, asset.status());
    } catch (FeignException.NotFound e) {
        throw new RuntimeException("Asset not found or inactive: " + ticker);
    } catch (FeignException e) {
        log.error("Asset service unavailable during ticker validation: {}", e.getMessage());
        throw new RuntimeException("Asset service unavailable. Please try again later.");
    }
}

Why catch FeignException separately from FeignException.NotFound? A 404 from the asset service means the ticker doesn't exist — a business error that should fail fast. Any other FeignException (connection refused, timeout, 5xx) means the asset service is unavailable — a different kind of failure that warrants a different message to the user.

3. Balance Validation — Wallet Feign Client

For buy orders (BUY), we need to ensure the user has available balance. We use Spring Cloud OpenFeign to query the Wallet API declaratively:

@FeignClient(name = "trading-broker-wallet", url = "${app.services.wallet-url}")
public interface WalletClient {

    @GetMapping("/api/v1/wallet/{userId}/summary")
    WalletSummaryDTO getWalletSummary(@PathVariable String userId);
}

The validation happens after ticker validation:

private void validateBalance(OrderRequestDTO request) {
    BigDecimal totalOrderValue = request.getPrice().multiply(request.getQuantity());
    var walletSummary = walletClient.getWalletSummary(request.getUserId());

    if (walletSummary.getAvailableBalance().compareTo(totalOrderValue) < 0) {
        throw new RuntimeException("Insufficient balance. Available: "
                + walletSummary.getAvailableBalance()
                + " | Required: " + totalOrderValue);
    }
}

Why Feign and not RestTemplate? OpenFeign eliminates HTTP boilerplate — the interface is sufficient. The URL is injected via application.yaml, making environment changes straightforward without recompiling.

4. The Creation Flow — OrderService

The placeOrder() is the heart of the service. It orchestrates five steps in sequence within a transaction:

@Transactional
public Order placeOrder(OrderRequestDTO request) {
    log.info("Processing new {} order for user: {}", request.getSide(), request.getUserId());

    // 1. Validate ticker exists and is ACTIVE in asset catalog
    validateTicker(request.getTicker());

    // 2. Validate balance for BUY orders via Wallet API
    if (request.getSide() == OrderSide.BUY) {
        validateBalance(request);
    }

    // 3. Persist order with PENDING status
    Order order = Order.builder()
            .userId(request.getUserId())
            .ticker(request.getTicker().toUpperCase())
            .quantity(request.getQuantity())
            .price(request.getPrice())
            .side(request.getSide())
            .status(OrderStatus.PENDING)
            .build();

    order = orderRepository.save(order);

    // 4. Send to B3 via RabbitMQ
    b3MessageProducer.sendToB3(mapToB3Message(order));

    // 5. Publish PENDING event to Kafka for wallet and other consumers
    orderEventProducer.sendOrderEvent(order.mapToEvent());

    return order;
}

Why validate ticker before balance? It's cheaper to fail fast on a non-existent ticker (one network call) than to validate balance first and then discover the ticker is invalid. Order of validations matters for UX and for minimizing unnecessary load on downstream services.

Design decision: The Kafka event is published with PENDING status right after creation. This allows trading-broker-wallet to immediately block the balance, without waiting for B3's result. This is the essence of eventual consistency — each service reacts to the event in its own time.

5. B3 Integration — RabbitMQ

Communication with the Matching Engine uses RabbitMQ for a specific reason: delivery guarantee. Even if the Matching Engine is temporarily unavailable, the message stays in the queue and is processed when it comes back.

Infrastructure configuration

All RabbitMQ infrastructure is declared as Spring beans — ensuring queues, exchanges and bindings exist before any message flows:

@Bean
public DirectExchange brokerExchange() {
    return new DirectExchange(exchangeName);
}

@Bean
public Queue orderOutQueue() {
    return QueueBuilder.durable(queueIn)
            .withArgument("x-dead-letter-exchange", "dlx.broker.orders")
            .withArgument("x-dead-letter-routing-key", "dlq-broker-to-b3")
            .build();
}

@Bean
public Binding orderOutBinding() {
    return BindingBuilder.bind(orderOutQueue())
            .to(brokerExchange())
            .with(routingKey);
}

The dead letter withArgument ensures messages that fail are redirected to the DLQ instead of simply being discarded.

Producer — sending to B3

public void sendToB3(B3OrderRequestDTO message) {
    log.info("Sending order {} to B3 via exchange: {} | routing-key: {}",
            message.getOrderId(), exchange, routingKey);
    rabbitTemplate.convertAndSend(exchange, routingKey, message);
}

Important decision: the message is sent to the exchange with the routing key, not directly to the queue. This respects the RabbitMQ routing model and allows adding new consumers without changing the producer.

Consumer — receiving B3 feedback

@RabbitListener(queues = "${app.rabbitmq.queue-out}")
public void receiveFeedback(B3OrderResponseDTO response) {
    log.info("Feedback received from B3 for order: {} | Status: {}",
            response.getOrderId(), response.getStatus());
    try {
        orderStatusService.updateOrderStatus(response);
    } catch (Exception e) {
        log.error("Failed to update status for order {}: {}",
                response.getOrderId(), e.getMessage(), e);
        throw e; // RabbitMQ applies retry/DLQ policy
    }
}

The throw e is intentional: rethrowing the exception allows RabbitMQ to apply its retry policy instead of silently discarding the message.

6. Status Update — OrderStatusService

When B3 feedback arrives, OrderStatusService updates the order state and publishes the second Kafka event:

@Transactional
public void updateOrderStatus(B3OrderResponseDTO response) {
    Order order = orderRepository.findById(Long.valueOf(response.getOrderId()))
            .orElseThrow(() -> new RuntimeException("Order not found: " + response.getOrderId()));

    order.setStatus(OrderStatus.valueOf(response.getStatus()));
    order.setExecutedPrice(response.getExecutedPrice());

    orderRepository.save(order);

    // Notify ecosystem with final status — wallet settles or refunds
    orderEventProducer.sendOrderEvent(order.mapToEvent());
}

This second Kafka event closes the loop: trading-broker-wallet consumes it and executes the settlement (FILLED) or refund (REJECTED) of the balance blocked in the previous step.

🌐 REST API

Method	Endpoint	Description
POST	`/api/v1/orders`	Place a new buy or sell order
GET	`/api/v1/orders/{id}`	Get order details by ID
GET	`/api/v1/orders/user/{userId}`	List all orders for a user

📄 Swagger UI: http://localhost:8088/swagger-ui.html

✅ Validating the Execution

With the application running locally:

✅ Flyway validated the orders table schema
✅ Tomcat started on port 8088
✅ RabbitMQ connected with exchange, queues and DLQ declared
✅ Swagger UI with 3 documented endpoints
✅ AssetClient validating ticker before any persistence
✅ WalletClient validating balance for BUY orders

🔗 The Complete Ecosystem

With trading-broker-order ready, we have the first end-to-end flow working:

[User] → [Order API] → [Asset API]  (ticker validation)
                    → [Wallet API]  (balance validation)
                    → [B3 Matching Engine] (execution)
                    → [Kafka] → [Wallet API] (custody)

From the user's click to the updated wallet balance, through ticker and balance validation, RabbitMQ routing, price matching against Redis, and eventual consistency via Kafka.

Got any questions about the design decisions or the inter-service orchestration? Drop them in the comments!

🔎 About the Series

⬅️ Previous Post: Financial Custody: Managing Balance and Portfolio with Eventual Consistency

📘 Series Index: Series Roadmap

Links:

O Maestro do Ecossistema: Orquestrando Ordens de Mercado com Kafka, RabbitMQ e Spring Boot

Roberto de Vargas Neto — Fri, 01 May 2026 23:59:54 +0000

Olá, pessoal!

Chegamos em um dos posts mais importantes da série My Broker B3. Até aqui construímos cada peça do ecossistema separadamente — o sincronizador de preços, o matching engine, a custódia financeira, o catálogo de ativos. Agora é hora de construir o serviço que conecta tudo isso: o Broker Order API.

Este é o maestro do ecossistema. Quando um usuário quer comprar ou vender uma ação, é aqui que tudo começa — e é aqui que os resultados chegam de volta.

🏦 O que é a Order API?

A trading-broker-order é o ponto de entrada para o fluxo transacional da corretora. Ela tem uma responsabilidade clara: orquestrar o ciclo de vida completo de uma ordem, desde a intenção do usuário até a confirmação final da execução.

Para isso, ela se comunica com quatro outros serviços do ecossistema:

Asset API (REST/Feign) — valida se o ticker existe e está ativo
Wallet API (REST/Feign) — valida se o usuário tem saldo suficiente
B3 Matching Engine (RabbitMQ) — envia a ordem para execução
Broker Wallet (Kafka) — notifica o ecossistema sobre cada mudança de status

🗺️ O Fluxo Completo de uma Ordem

Antes de mergulhar no código, veja o fluxo ponta a ponta:

[Usuário]
    │
    │  POST /api/v1/orders
    │  { userId, ticker, quantity, price, side: "BUY" }
    ▼
[trading-broker-order]
    │
    ├─ 1. Valida ticker  → GET trading-asset/assets/{ticker} (Feign/REST)
    │
    ├─ 2. Valida saldo   → GET trading-wallet/summary (Feign/REST)
    │
    ├─ 3. Persiste ordem → MySQL (status: PENDING)
    │
    ├─ 4. Envia para B3  → RabbitMQ (mq-broker-to-b3)
    │
    ├─ 5. Notifica ecossistema → Kafka (order-events-v1, status: PENDING)
    │
    └─ Retorna ordem criada → HTTP 201

         ─ ─ ─ (B3 processa a ordem) ─ ─ ─

[b3-matching-engine]
    │
    │  RabbitMQ (mq-b3-to-broker)
    │  { orderId, status: FILLED/REJECTED, executedPrice }
    ▼
[trading-broker-order]
    │
    ├─ 6. Atualiza status → MySQL (FILLED ou REJECTED)
    │
    └─ 7. Notifica ecossistema → Kafka (order-events-v1, status final)
              │
              ├─ trading-broker-wallet consome → bloqueia/liquida/estorna saldo
              └─ outros consumers futuros

Dois canais de comunicação distintos, dois momentos de publicação Kafka, e a wallet reagindo a cada mudança de estado. Vamos ver como cada parte foi implementada.

🛠️ Stack Tecnológica

Tecnologia	Uso
Java 21 + Spring Boot 3.3.5	Core do serviço
MySQL + Flyway	Persistência de ordens
Spring Kafka	Publicação de eventos de ciclo de vida
Spring AMQP	Integração com o Matching Engine via RabbitMQ
Spring Cloud OpenFeign	Chamadas REST à Asset API e Wallet API
SpringDoc OpenAPI	Documentação via Swagger UI

🏗️ Os Pilares da Implementação

1. O Modelo de Domínio — Order

A entidade Order representa o estado de uma ordem em qualquer ponto do ciclo de vida:

@Entity
@Table(name = "orders")
public class Order {

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    private String userId;
    private String ticker;

    @Column(nullable = false, precision = 19, scale = 4)
    private BigDecimal quantity;

    @Column(nullable = false, precision = 19, scale = 4)
    private BigDecimal price;

    // Null enquanto PENDING — preenchido apenas após feedback da B3
    @Column(precision = 19, scale = 4)
    private BigDecimal executedPrice;

    @Enumerated(EnumType.STRING)
    private OrderSide side;   // BUY | SELL

    @Enumerated(EnumType.STRING)
    private OrderStatus status; // PENDING | FILLED | REJECTED | CANCELLED

    @CreationTimestamp
    private LocalDateTime createdAt;

    @UpdateTimestamp
    private LocalDateTime updatedAt;
}

Um detalhe de design importante: executedPrice é nullable intencionalmente. Quando a ordem é criada, ainda não sabemos o preço de execução — isso só vem como resposta do Matching Engine. Forçar nullable = false aqui seria um erro de modelo de domínio.

A entidade também carrega o método mapToEvent() que serializa o estado atual para o evento Kafka:

public OrderEventDTO mapToEvent() {
    return OrderEventDTO.builder()
            .orderId(this.getId())
            .userId(this.getUserId())
            .ticker(this.getTicker())
            .quantity(this.getQuantity())
            .price(this.getPrice())
            .executedPrice(this.getExecutedPrice())
            .side(this.getSide().name())
            .status(this.getStatus().name())
            .eventTimestamp(LocalDateTime.now())
            .build();
}

2. Validação de Ticker — Asset Feign Client

Antes de validar saldo ou persistir qualquer coisa, precisamos garantir que o ticker solicitado existe e está disponível para negociação. Um usuário submetendo uma ordem para XPTO999 deve receber um erro claro imediatamente — não uma rejeição silenciosa da B3 duas chamadas depois.

@FeignClient(name = "trading-broker-asset", url = "${app.services.asset-url}")
public interface AssetClient {

    @GetMapping("/api/v1/assets/{ticker}")
    AssetDTO getByTicker(@PathVariable("ticker") String ticker);
}

A validação no OrderService:

private void validateTicker(String ticker) {
    try {
        var asset = assetClient.getByTicker(ticker.toUpperCase());
        if (!"ACTIVE".equalsIgnoreCase(asset.status())) {
            throw new RuntimeException("Asset is not available for trading: " + ticker);
        }
        log.info("Ticker {} validated — status: {}", ticker, asset.status());
    } catch (FeignException.NotFound e) {
        throw new RuntimeException("Asset not found or inactive: " + ticker);
    } catch (FeignException e) {
        log.error("Asset service unavailable during ticker validation: {}", e.getMessage());
        throw new RuntimeException("Asset service unavailable. Please try again later.");
    }
}

Por que capturar FeignException.NotFound separadamente? Um 404 da asset service significa que o ticker não existe — erro de negócio que deve falhar rápido. Qualquer outro FeignException (conexão recusada, timeout, 5xx) significa que o serviço está indisponível — um tipo diferente de falha que merece uma mensagem diferente para o usuário.

3. Validação de Saldo — Wallet Feign Client

Para ordens de compra (BUY), precisamos garantir que o usuário tem saldo disponível. Usamos o Spring Cloud OpenFeign para consultar a Wallet API de forma declarativa:

@FeignClient(name = "trading-broker-wallet", url = "${app.services.wallet-url}")
public interface WalletClient {

    @GetMapping("/api/v1/wallet/{userId}/summary")
    WalletSummaryDTO getWalletSummary(@PathVariable String userId);
}

A validação acontece após a validação do ticker:

private void validateBalance(OrderRequestDTO request) {
    BigDecimal totalOrderValue = request.getPrice().multiply(request.getQuantity());
    var walletSummary = walletClient.getWalletSummary(request.getUserId());

    if (walletSummary.getAvailableBalance().compareTo(totalOrderValue) < 0) {
        throw new RuntimeException("Insufficient balance. Available: "
                + walletSummary.getAvailableBalance()
                + " | Required: " + totalOrderValue);
    }
}

Por que Feign e não RestTemplate? O OpenFeign elimina o boilerplate de HTTP — a interface é suficiente. A URL é injetada via application.yaml, facilitando a mudança de ambiente sem recompilar.

4. O Fluxo de Criação — OrderService

O placeOrder() é o coração do serviço. Ele orquestra cinco etapas em sequência dentro de uma transação:

@Transactional
public Order placeOrder(OrderRequestDTO request) {
    log.info("Processing new {} order for user: {}", request.getSide(), request.getUserId());

    // 1. Valida ticker existe e está ACTIVE no catálogo de ativos
    validateTicker(request.getTicker());

    // 2. Valida saldo para ordens BUY via Wallet API
    if (request.getSide() == OrderSide.BUY) {
        validateBalance(request);
    }

    // 3. Persiste ordem com status PENDING
    Order order = Order.builder()
            .userId(request.getUserId())
            .ticker(request.getTicker().toUpperCase())
            .quantity(request.getQuantity())
            .price(request.getPrice())
            .side(request.getSide())
            .status(OrderStatus.PENDING)
            .build();

    order = orderRepository.save(order);

    // 4. Envia para a B3 via RabbitMQ
    b3MessageProducer.sendToB3(mapToB3Message(order));

    // 5. Publica evento PENDING no Kafka para a wallet e outros consumers
    orderEventProducer.sendOrderEvent(order.mapToEvent());

    return order;
}

Por que validar o ticker antes do saldo? É mais barato falhar rápido num ticker inexistente (uma chamada de rede) do que validar saldo primeiro e só então descobrir que o ticker é inválido. A ordem das validações importa para a UX e para minimizar carga desnecessária nos serviços downstream.

Decisão de design: O evento Kafka é publicado com status PENDING logo após a criação. Isso permite que a trading-broker-wallet já bloqueie o saldo imediatamente, sem esperar o resultado da B3. Essa é a essência da consistência eventual — cada serviço reage ao evento no seu próprio tempo.

5. A Integração com a B3 — RabbitMQ

A comunicação com o Matching Engine usa RabbitMQ por uma razão específica: garantia de entrega. Mesmo que o Matching Engine esteja temporariamente indisponível, a mensagem fica na fila e é processada quando ele voltar.

Configuração de infraestrutura

Toda a infraestrutura RabbitMQ é declarada como beans Spring — garantindo que filas, exchanges e bindings existam antes de qualquer mensagem trafegar:

@Bean
public DirectExchange brokerExchange() {
    return new DirectExchange(exchangeName);
}

@Bean
public Queue orderOutQueue() {
    return QueueBuilder.durable(queueIn)
            .withArgument("x-dead-letter-exchange", "dlx.broker.orders")
            .withArgument("x-dead-letter-routing-key", "dlq-broker-to-b3")
            .build();
}

@Bean
public Binding orderOutBinding() {
    return BindingBuilder.bind(orderOutQueue())
            .to(brokerExchange())
            .with(routingKey);
}

O withArgument de dead letter garante que mensagens que falham são redirecionadas para a DLQ em vez de simplesmente descartadas.

Producer — enviando para a B3

public void sendToB3(B3OrderRequestDTO message) {
    log.info("Sending order {} to B3 via exchange: {} | routing-key: {}",
            message.getOrderId(), exchange, routingKey);
    rabbitTemplate.convertAndSend(exchange, routingKey, message);
}

Decisão importante: a mensagem é enviada para o exchange com a routing key, não diretamente para a fila. Isso respeita o modelo de roteamento do RabbitMQ e permite adicionar novos consumers sem alterar o producer.

Consumer — recebendo o feedback da B3

@RabbitListener(queues = "${app.rabbitmq.queue-out}")
public void receiveFeedback(B3OrderResponseDTO response) {
    log.info("Feedback received from B3 for order: {} | Status: {}",
            response.getOrderId(), response.getStatus());
    try {
        orderStatusService.updateOrderStatus(response);
    } catch (Exception e) {
        log.error("Failed to update status for order {}: {}",
                response.getOrderId(), e.getMessage(), e);
        throw e; // RabbitMQ applies retry/DLQ policy
    }
}

O throw e é intencional: relançar a exceção permite que o RabbitMQ aplique sua política de retry em vez de descartar a mensagem silenciosamente.

6. Atualização de Status — OrderStatusService

Quando o feedback da B3 chega, o OrderStatusService atualiza o estado da ordem e publica o segundo evento Kafka:

@Transactional
public void updateOrderStatus(B3OrderResponseDTO response) {
    Order order = orderRepository.findById(Long.valueOf(response.getOrderId()))
            .orElseThrow(() -> new RuntimeException("Order not found: " + response.getOrderId()));

    order.setStatus(OrderStatus.valueOf(response.getStatus()));
    order.setExecutedPrice(response.getExecutedPrice());

    orderRepository.save(order);

    // Notifica ecossistema com status final — wallet liquida ou estorna
    orderEventProducer.sendOrderEvent(order.mapToEvent());
}

Este segundo evento Kafka fecha o loop: a trading-broker-wallet o consome e executa a liquidação (FILLED) ou estorno (REJECTED) do saldo bloqueado na etapa anterior.

🌐 API REST

Método	Endpoint	Descrição
POST	`/api/v1/orders`	Submeter nova ordem de compra ou venda
GET	`/api/v1/orders/{id}`	Consultar detalhes de uma ordem por ID
GET	`/api/v1/orders/user/{userId}`	Listar todas as ordens de um usuário

📄 Swagger UI: http://localhost:8088/swagger-ui.html

✅ Validando a Execução

Com a aplicação rodando localmente:

✅ Flyway validou o schema da tabela orders
✅ Tomcat subiu na porta 8088
✅ RabbitMQ conectado com exchange, filas e DLQ declarados
✅ Swagger UI com 3 endpoints documentados
✅ AssetClient validando ticker antes de qualquer persistência
✅ WalletClient validando saldo para ordens BUY

🔗 O Ecossistema Completo

Com a trading-broker-order pronta, temos o primeiro fluxo ponta a ponta funcionando:

[Usuário] → [Order API] → [Asset API]  (validação de ticker)
                       → [Wallet API]  (validação de saldo)
                       → [B3 Matching Engine] (execução)
                       → [Kafka] → [Wallet API] (custódia)

Do clique do usuário até o saldo atualizado na carteira, passando por validação de ticker e saldo, roteamento RabbitMQ, matching de preço no Redis e consistência eventual via Kafka.

Ficou com alguma dúvida sobre as decisões de design ou sobre a orquestração entre os serviços? Deixe nos comentários!

🔎 Sobre a série

⬅️ Post Anterior: Custódia Financeira: Gerenciando Saldo e Carteira com Consistência Eventual

📘 Índice da Série: Guia da Série

Links:

Financial Custody: Managing Balance and Portfolio with Eventual Consistency

Roberto de Vargas Neto — Fri, 01 May 2026 13:44:38 +0000

Hey folks!

Continuing the My Broker B3 series, we've reached one of the most business-logic-rich services in the ecosystem: the Broker Wallet API.

In previous posts we built the B3 infrastructure (market price sync and matching engine). Now we enter the financial core of the broker. This service is the guardian of the investor's money and assets — it needs to be correct above all else.

🏦 What is the Wallet API?

The trading-broker-wallet is the financial custody service of the ecosystem. Its responsibility is to react to order lifecycle events and ensure the user's money and assets are managed correctly at each step.

It doesn't receive direct buy commands — it listens to Kafka events and acts according to each order's status:

Kafka: order-events-v1
         │
    ┌────┴────────────────────────┐
    │                             │
 PENDING                       FILLED / REJECTED
    │                             │
 reserveBalance()          settleOrder() / refundBalance()
    │                             │
 Blocks balance            Settles or refunds

🎯 Why is the Wallet so Critical?

Imagine the following scenario without proper custody:

User has R$ 1,000 available
Places a buy order for R$ 800
Before the order is processed, places another order for R$ 500
Both orders are executed → user ends up with negative balance

The wallet solves this with the concept of blocked balance: the moment an order enters as PENDING, the value is immediately reserved. The available balance is always balance - blockedBalance.

🛠️ Tech Stack

Technology	Usage
Java 21 + Spring Boot 3.5.11	Service core
MySQL + Flyway	Operational persistence and versioning
Spring Kafka	Order event consumption
Spring Data Redis	Real-time prices for portfolio valuation
SpringDoc OpenAPI	Swagger UI documentation

🏗️ The Data Model

Before diving into the logic, it's important to understand the three entities that support the service:

Account
├── userId (UNIQUE)
├── balance          ← total balance (includes blocked amount)
├── blockedBalance   ← amount reserved for PENDING orders
└── currency

Account (1) ──▶ (N) Position
                     ├── ticker
                     ├── quantity
                     └── averagePrice

WalletTransaction (audit log)
├── orderId
├── userId
├── transactionType  ← RESERVE | SETTLEMENT | REFUND
└── amount

The blockedBalance is the central piece of custody. The balance never goes below zero — only blockedBalance goes up and down as orders progress through their lifecycle.

🔄 The Order Lifecycle in the Wallet

Step 1 — PENDING: Reserving the Balance

When an order is created, it arrives as PENDING. The wallet must immediately block the maximum amount that could be debited (quantity × limit price):

@Transactional
public void reserveBalance(OrderEventDTO event) {
    BigDecimal amountToReserve = event.getPrice().multiply(event.getQuantity());

    Account account = accountRepository.findByUserId(event.getUserId())
            .orElseThrow(() -> new RuntimeException("Account not found for user: " + event.getUserId()));

    // Available = balance - blockedBalance
    BigDecimal availableBalance = account.getAvailableBalance();

    if (availableBalance.compareTo(amountToReserve) < 0) {
        throw new RuntimeException("Insufficient balance to reserve funds.");
    }

    // Only blockedBalance increases — total balance stays intact
    account.setBlockedBalance(account.getBlockedBalance().add(amountToReserve));
    accountRepository.save(account);

    // Audit record
    walletTransactionRepository.save(WalletTransaction.builder()
            .userId(event.getUserId())
            .orderId(event.getOrderId())
            .transactionType(TransactionType.RESERVE)
            .amount(amountToReserve)
            .build());
}

Important point: the total balance doesn't change at this step. Only blockedBalance increases. The money is still in the account — it's just reserved.

Step 2a — FILLED: Settling the Order

The order was executed by the Matching Engine. Now the money actually leaves. But here's a subtlety: the executed price may differ from the reserved price.

Example: user placed a buy order with a limit of R$ 30.00, but the market executed at R$ 29.50. The reserved amount was R$ 300 (10 shares × R$ 30), but the actual executed amount was R$ 295 (10 × R$ 29.50). The R$ 5.00 difference automatically becomes available again.

@Transactional
public void settleOrder(OrderEventDTO event) {
    BigDecimal totalReserved = event.getPrice().multiply(event.getQuantity());
    BigDecimal totalExecuted = event.getExecutedPrice().multiply(event.getQuantity());

    Account account = accountRepository.findByUserId(event.getUserId())
            .orElseThrow(() -> new RuntimeException("Account not found"));

    // Release the initial block
    BigDecimal newBlockedBalance = account.getBlockedBalance().subtract(totalReserved);
    account.setBlockedBalance(newBlockedBalance.compareTo(BigDecimal.ZERO) < 0
            ? BigDecimal.ZERO
            : newBlockedBalance);

    // Debit the actual executed amount from total balance
    account.setBalance(account.getBalance().subtract(totalExecuted));
    accountRepository.save(account);

    // Update asset position
    updatePosition(event, account);
}

Step 2b — REJECTED: Refunding the Balance

The order was rejected (price outside market range, ticker not found in Redis, etc.). The blocked money becomes available again:

@Transactional
public void refundBalance(OrderEventDTO event) {
    BigDecimal amountToRefund = event.getPrice().multiply(event.getQuantity());

    Account account = accountRepository.findByUserId(event.getUserId())
            .orElseThrow(() -> new RuntimeException("Account not found for refund"));

    // Only subtract from blockedBalance — total balance was never touched
    if (account.getBlockedBalance().compareTo(amountToRefund) >= 0) {
        account.setBlockedBalance(account.getBlockedBalance().subtract(amountToRefund));
    } else {
        log.warn("Refund amount exceeds blocked balance for order {}", event.getOrderId());
        account.setBlockedBalance(BigDecimal.ZERO);
    }

    accountRepository.save(account);
}

📊 Weighted Average Price Calculation

When a purchase is settled, we need to update the user's position in that asset. The weighted average price ensures that purchases made at different times are correctly reflected:

New Average Price = (Current Cost + New Cost) / (Current Qty + New Qty)

private void updatePosition(OrderEventDTO event, Account account) {
    Position position = positionRepository
            .findByAccountIdAndTicker(account.getId(), event.getTicker())
            .orElse(Position.builder()
                    .account(account)
                    .ticker(event.getTicker())
                    .quantity(BigDecimal.ZERO)
                    .averagePrice(BigDecimal.ZERO)
                    .build());

    boolean isSell = "SELL".equalsIgnoreCase(event.getSide());

    if (isSell) {
        BigDecimal newQuantity = position.getQuantity().subtract(event.getQuantity());
        if (newQuantity.compareTo(BigDecimal.ZERO) <= 0) {
            positionRepository.delete(position);
            return;
        }
        position.setQuantity(newQuantity);
        // Average price stays the same on sell
    } else {
        // BUY: recalculate weighted average price
        BigDecimal currentCost = position.getAveragePrice().multiply(position.getQuantity());
        BigDecimal newCost = event.getExecutedPrice().multiply(event.getQuantity());
        BigDecimal totalQuantity = position.getQuantity().add(event.getQuantity());
        BigDecimal newAveragePrice = currentCost.add(newCost)
                .divide(totalQuantity, 4, RoundingMode.HALF_UP);

        position.setQuantity(totalQuantity);
        position.setAveragePrice(newAveragePrice);
    }

    positionRepository.save(position);
}

Practical example:

Purchase 1: 10 PETR4 shares at R$ 30.00 → average price = R$ 30.00
Purchase 2: 5 PETR4 shares at R$ 33.00 → average price = (300 + 165) / 15 = R$ 31.00

🐛 Critical Bugs Fixed

During the code review, I identified issues that would cause silent failures in production:

1. Guaranteed NPE on first deposit
New account created without initializing blockedBalance. The first call to getAvailableBalance() would throw NullPointerException.

// ❌ Before
Account.builder()
    .balance(BigDecimal.ZERO)
    // blockedBalance missing → NPE

// ✅ After
Account.builder()
    .balance(BigDecimal.ZERO)
    .blockedBalance(BigDecimal.ZERO)

2. Withdraw validated total balance, not available balance
Allowed withdrawing money that was blocked in pending orders:

// ❌ Before — allows withdrawing reserved funds
if (account.getBalance().compareTo(amount) < 0)

// ✅ After — validates only what's available
if (account.getAvailableBalance().compareTo(amount) < 0)

3. SELL increased position instead of decreasing it
The updatePosition() method always added quantity, completely ignoring the side field. Selling 10 shares would add 10 to the position.

4. Kafka consumer silently swallowed exceptions
Errors were logged but the offset was committed — the message was discarded without retry. Now the consumer rethrows the exception so Kafka can apply its retry policy correctly.

🔒 Idempotency

To prevent duplicates in case of Kafka reprocessing, we added a database constraint:

-- V4 migration
ALTER TABLE wallet_transactions
    ADD CONSTRAINT uq_order_transaction UNIQUE (order_id, transaction_type);

If the same PENDING event arrives twice, the second insert will fail with a constraint violation — preventing the balance from being blocked twice for the same order.

🌐 REST API

Method	Endpoint	Description
POST	`/api/v1/wallet/{userId}/deposit`	Deposit funds
POST	`/api/v1/wallet/{userId}/withdraw`	Withdraw available funds
GET	`/api/v1/wallet/{userId}/summary`	Summary: balance + positions + total equity
GET	`/api/v1/wallet/{userId}/positions`	List asset positions
GET	`/api/v1/wallet/{userId}/transactions`	Transaction history

📄 Swagger UI: http://localhost:8085/swagger-ui.html

✅ Validating the Execution

With the application running locally:

✅ Flyway applied all 4 migrations successfully
✅ Hibernate validated the schema
✅ Kafka consumer connected and subscribed to order-events-v1
✅ Consumer group trading-broker-wallet synced
✅ Swagger UI with 5 documented endpoints at http://localhost:8085/swagger-ui.html

🚀 What's Next?

With the Wallet ready to react to order events, the next step is building the trading-broker-order — the orchestrator that will manage the complete order lifecycle:

Receive buy/sell intent via REST
Validate the ticker against trading-broker-asset
Save the order as PENDING and publish to Kafka
Send the order to the Matching Engine via RabbitMQ
Consume the result and update the final status

Once that service is ready, we'll have the first complete end-to-end flow: from the user's click all the way to the updated wallet balance.

Got any questions about the custody logic or the PENDING → FILLED → REJECTED cycle? Drop them in the comments!

🔎 About the Series

⬅️ Previous Post: The Heart of B3: Building the Matching Engine with RabbitMQ and Redis

📘 Series Index: Series Roadmap

Links:

Custódia Financeira: Gerenciando Saldo e Carteira com Consistência Eventual

Roberto de Vargas Neto — Fri, 01 May 2026 13:43:34 +0000

Olá, pessoal!

Continuando a série My Broker B3, chegamos em um dos serviços mais ricos em regras de negócio do ecossistema: o Broker Wallet API.

Se nos posts anteriores construímos a infraestrutura da B3 (sync de preços e matching engine), agora entramos no coração financeiro da corretora. Este serviço é o guardião do dinheiro e dos ativos do investidor — ele precisa ser correto acima de tudo.

🏦 O que é a Wallet API?

A trading-broker-wallet é o serviço de custódia financeira do ecossistema. Sua responsabilidade é reagir aos eventos do ciclo de vida de uma ordem e garantir que o dinheiro e os ativos do usuário sejam gerenciados corretamente em cada etapa.

Ela não recebe comandos diretos de compra — ela escuta eventos Kafka e age de acordo com o status de cada ordem:

Kafka: order-events-v1
         │
    ┌────┴────────────────────────┐
    │                             │
 PENDING                       FILLED / REJECTED
    │                             │
 reserveBalance()          settleOrder() / refundBalance()
    │                             │
 Bloqueia saldo            Liquida ou estorna

🎯 Por que a Wallet é tão crítica?

Imagine o seguinte cenário sem uma custódia bem implementada:

Usuário tem R$ 1.000 disponíveis
Coloca uma ordem de compra de R$ 800
Antes da ordem ser processada, coloca outra ordem de R$ 500
Ambas as ordens são executadas → usuário fica com saldo negativo

A wallet resolve isso com o conceito de saldo bloqueado: no momento em que uma ordem entra como PENDING, o valor é imediatamente reservado. O saldo disponível é sempre balance - blockedBalance.

🛠️ Stack Tecnológica

Tecnologia	Uso
Java 21 + Spring Boot 3.5.11	Core do serviço
MySQL + Flyway	Persistência operacional e versionamento
Spring Kafka	Consumo de eventos de ordem
Spring Data Redis	Preços em tempo real para valorização da carteira
SpringDoc OpenAPI	Documentação via Swagger UI

🏗️ O Modelo de Dados

Antes de entrar na lógica, é importante entender as três entidades que sustentam o serviço:

Account
├── userId (UNIQUE)
├── balance          ← saldo total (inclui o bloqueado)
├── blockedBalance   ← valor reservado para ordens PENDING
└── currency

Account (1) ──▶ (N) Position
                     ├── ticker
                     ├── quantity
                     └── averagePrice

WalletTransaction (auditoria)
├── orderId
├── userId
├── transactionType  ← RESERVE | SETTLEMENT | REFUND
└── amount

O blockedBalance é a peça central da custódia. O balance nunca desce abaixo de zero — apenas o blockedBalance sobe e desce conforme as ordens avançam no ciclo de vida.

🔄 O Ciclo de Vida de uma Ordem na Wallet

Etapa 1 — PENDING: Reservando o Saldo

Quando uma ordem é criada, ela chega como PENDING. A wallet precisa bloquear imediatamente o valor máximo que pode ser debitado (quantidade × preço limite):

@Transactional
public void reserveBalance(OrderEventDTO event) {
    BigDecimal amountToReserve = event.getPrice().multiply(event.getQuantity());

    Account account = accountRepository.findByUserId(event.getUserId())
            .orElseThrow(() -> new RuntimeException("Account not found for user: " + event.getUserId()));

    // Available = balance - blockedBalance
    BigDecimal availableBalance = account.getAvailableBalance();

    if (availableBalance.compareTo(amountToReserve) < 0) {
        throw new RuntimeException("Insufficient balance to reserve funds.");
    }

    // Only blockedBalance increases — total balance stays intact
    account.setBlockedBalance(account.getBlockedBalance().add(amountToReserve));
    accountRepository.save(account);

    // Audit record
    walletTransactionRepository.save(WalletTransaction.builder()
            .userId(event.getUserId())
            .orderId(event.getOrderId())
            .transactionType(TransactionType.RESERVE)
            .amount(amountToReserve)
            .build());
}

Ponto importante: o balance total não muda nesta etapa. Apenas o blockedBalance aumenta. O dinheiro ainda está na conta — só está reservado.

Etapa 2a — FILLED: Liquidando a Ordem

A ordem foi executada no Matching Engine. Agora sim o dinheiro sai de verdade. Mas aqui temos uma sutileza: o preço executado pode ser diferente do preço reservado.

Exemplo: usuário colocou ordem de compra com limite de R$ 30,00, mas o mercado executou a R$ 29,50. O valor reservado foi R$ 300 (10 ações × R$ 30), mas o real executado foi R$ 295 (10 × R$ 29,50). Os R$ 5,00 de diferença ficam disponíveis automaticamente.

@Transactional
public void settleOrder(OrderEventDTO event) {
    BigDecimal totalReserved = event.getPrice().multiply(event.getQuantity());
    BigDecimal totalExecuted = event.getExecutedPrice().multiply(event.getQuantity());

    Account account = accountRepository.findByUserId(event.getUserId())
            .orElseThrow(() -> new RuntimeException("Account not found"));

    // Release the initial block
    BigDecimal newBlockedBalance = account.getBlockedBalance().subtract(totalReserved);
    account.setBlockedBalance(newBlockedBalance.compareTo(BigDecimal.ZERO) < 0
            ? BigDecimal.ZERO
            : newBlockedBalance);

    // Debit the actual executed amount from total balance
    account.setBalance(account.getBalance().subtract(totalExecuted));
    accountRepository.save(account);

    // Update asset position
    updatePosition(event, account);
}

Etapa 2b — REJECTED: Estornando o Saldo

A ordem foi rejeitada (preço fora do mercado, ticker não encontrado no Redis, etc.). O dinheiro bloqueado volta a ficar disponível:

@Transactional
public void refundBalance(OrderEventDTO event) {
    BigDecimal amountToRefund = event.getPrice().multiply(event.getQuantity());

    Account account = accountRepository.findByUserId(event.getUserId())
            .orElseThrow(() -> new RuntimeException("Account not found for refund"));

    // Only subtract from blockedBalance — total balance was never touched
    if (account.getBlockedBalance().compareTo(amountToRefund) >= 0) {
        account.setBlockedBalance(account.getBlockedBalance().subtract(amountToRefund));
    } else {
        log.warn("Refund amount exceeds blocked balance for order {}", event.getOrderId());
        account.setBlockedBalance(BigDecimal.ZERO);
    }

    accountRepository.save(account);
}

📊 Cálculo de Preço Médio Ponderado

Quando uma compra é liquidada, precisamos atualizar a posição do usuário naquele ativo. O preço médio ponderado garante que compras em momentos diferentes sejam refletidas corretamente:

Novo Preço Médio = (Custo Atual + Custo Novo) / (Qtd Atual + Qtd Nova)

private void updatePosition(OrderEventDTO event, Account account) {
    Position position = positionRepository
            .findByAccountIdAndTicker(account.getId(), event.getTicker())
            .orElse(Position.builder()
                    .account(account)
                    .ticker(event.getTicker())
                    .quantity(BigDecimal.ZERO)
                    .averagePrice(BigDecimal.ZERO)
                    .build());

    boolean isSell = "SELL".equalsIgnoreCase(event.getSide());

    if (isSell) {
        BigDecimal newQuantity = position.getQuantity().subtract(event.getQuantity());
        if (newQuantity.compareTo(BigDecimal.ZERO) <= 0) {
            positionRepository.delete(position);
            return;
        }
        position.setQuantity(newQuantity);
        // Average price stays the same on sell
    } else {
        // BUY: recalculate weighted average price
        BigDecimal currentCost = position.getAveragePrice().multiply(position.getQuantity());
        BigDecimal newCost = event.getExecutedPrice().multiply(event.getQuantity());
        BigDecimal totalQuantity = position.getQuantity().add(event.getQuantity());
        BigDecimal newAveragePrice = currentCost.add(newCost)
                .divide(totalQuantity, 4, RoundingMode.HALF_UP);

        position.setQuantity(totalQuantity);
        position.setAveragePrice(newAveragePrice);
    }

    positionRepository.save(position);
}

Exemplo prático:

Compra 1: 10 ações de PETR4 a R$ 30,00 → preço médio = R$ 30,00
Compra 2: 5 ações de PETR4 a R$ 33,00 → preço médio = (300 + 165) / 15 = R$ 31,00

🐛 Bugs Críticos Corrigidos

Durante a revisão do código, identifiquei problemas que causariam falhas silenciosas em produção:

1. NPE garantido no primeiro depósito
Conta nova criada sem inicializar blockedBalance. A primeira chamada a getAvailableBalance() explodiria com NullPointerException.

// ❌ Antes
Account.builder()
    .balance(BigDecimal.ZERO)
    // blockedBalance ausente → NPE

// ✅ Depois
Account.builder()
    .balance(BigDecimal.ZERO)
    .blockedBalance(BigDecimal.ZERO)

2. Saque validava saldo total, não disponível
Permitia sacar dinheiro bloqueado em ordens pendentes:

// ❌ Antes — permite sacar fundos reservados
if (account.getBalance().compareTo(amount) < 0)

// ✅ Depois — valida apenas o disponível
if (account.getAvailableBalance().compareTo(amount) < 0)

3. SELL aumentava posição em vez de diminuir
O método updatePosition() sempre somava quantidade, ignorando completamente o campo side. Uma venda de 10 ações adicionava 10 à posição.

4. Consumer Kafka engolia exceções silenciosamente
Erros eram logados mas o offset era confirmado — a mensagem era descartada sem retry. Agora o consumer relança a exceção para o Kafka aplicar a política de retry corretamente.

🔒 Idempotência

Para evitar duplicidade em caso de reprocessamento Kafka, adicionamos uma constraint de banco:

-- V4 migration
ALTER TABLE wallet_transactions
    ADD CONSTRAINT uq_order_transaction UNIQUE (order_id, transaction_type);

Se o mesmo evento PENDING chegar duas vezes, o segundo insert vai falhar com violação de constraint — evitando que o saldo seja bloqueado duas vezes para a mesma ordem.

🌐 API REST

Método	Endpoint	Descrição
POST	`/api/v1/wallet/{userId}/deposit`	Depositar fundos
POST	`/api/v1/wallet/{userId}/withdraw`	Sacar fundos disponíveis
GET	`/api/v1/wallet/{userId}/summary`	Resumo: saldo + posições + patrimônio total
GET	`/api/v1/wallet/{userId}/positions`	Listar posições em ativos
GET	`/api/v1/wallet/{userId}/transactions`	Histórico de transações

📄 Swagger UI: http://localhost:8085/swagger-ui.html

✅ Validando a Execução

Com a aplicação rodando localmente:

✅ Flyway aplicou as 4 migrations com sucesso
✅ Hibernate validou o schema
✅ Kafka consumer conectado e subscrito ao tópico order-events-v1
✅ Consumer group trading-broker-wallet sincronizado
✅ Swagger UI com 5 endpoints documentados em http://localhost:8085/swagger-ui.html

🚀 O que vem a seguir?

Com a Wallet pronta para reagir aos eventos de ordem, o próximo passo é construir o trading-broker-order — o maestro que vai orquestrar o ciclo de vida completo de uma ordem:

Receber a intenção de compra/venda via REST
Validar o ticker na trading-broker-asset
Salvar a ordem como PENDING e publicar no Kafka
Enviar a ordem ao Matching Engine via RabbitMQ
Consumir o resultado e atualizar o status final

Quando esse serviço estiver pronto, teremos o primeiro fluxo ponta a ponta funcionando: do clique do usuário até o saldo atualizado na carteira.

Ficou com alguma dúvida sobre a lógica de custódia ou sobre o ciclo PENDING → FILLED → REJECTED? Deixe nos comentários!

🔎 Sobre a série

⬅️ Post Anterior: O Coração da B3: Construindo o Matching Engine com RabbitMQ e Redis

📘 Índice da Série: Guia da Série

Links:

The Heart of B3: Building the Matching Engine with RabbitMQ, Redis and Spring Boot

Roberto de Vargas Neto — Sun, 26 Apr 2026 23:31:27 +0000

Hello everyone!

Continuing the My Broker B3 series, we've reached one of the most anticipated components of the ecosystem: the B3 Matching Engine API.

In the previous post, we built b3-market-sync-api which synchronizes real prices from brapi.dev and stores them in Redis. Now it's time to use that data — this is the service that simulates the stock exchange itself, deciding whether a buy or sell order will be executed or rejected based on the prices we just put in cache.

🏗️ What is the Matching Engine?

In the real world, when you place a buy order through your broker, it gets sent to B3, which checks whether there's a counterpart willing to trade at that price. This process is called matching.

In our simulator, the b3-matching-engine-api reproduces this logic in a simplified way:

Intake: Receives the order from the broker via RabbitMQ (queue mq-broker-to-b3)
Price: Fetches the current asset price from Redis (market:price:{TICKER}) — data injected by the Market Sync we built previously
Decision: Applies the matching rule
Persistence: Records the result in PostgreSQL
Notification: Returns the result to the broker via RabbitMQ (queue mq-b3-to-broker)

[Broker] ──RabbitMQ──▶ [mq-broker-to-b3] ──▶ [Matching Engine]
                                                      │
                                             Redis: market:price:{TICKER}
                                             (fed by Market Sync ⬆️)
                                                      │
                                         ┌────────────┴────────────┐
                                      FILLED                   REJECTED
                                         │                         │
                                   PostgreSQL                 PostgreSQL
                                         │                         │
                               [mq-b3-to-broker] ◀────────────────┘
                                         │
                                      [Broker]

🎯 MVP Focus

Before diving into the code, same disclaimer as previous posts: we're building the foundation. The goal is to have an end-to-end flow working with enough robustness to validate the POC.

In this phase, I prioritized:

End-to-end flow working correctly
Proper failure handling (no orders silently disappearing)
Dead Letter Queue for messages that fail during processing
REST API for querying execution history
Swagger documentation

🛠️ Tech Stack

Technology	Usage
Java 21 + Spring Boot 3.5	Service core
Spring RabbitMQ	Order intake and result notification
Spring Data Redis	Real-time price lookup
Spring Data JPA + PostgreSQL	Execution persistence
Flyway	Database schema versioning
SpringDoc OpenAPI	Swagger UI documentation

🏗️ Implementation Pillars

1. RabbitMQ Configuration

A critical lesson learned here: never rely on queues already existing in the broker. If the application starts before RabbitMQ has the queues created, the consumer fails on startup.

The solution is to declare all infrastructure beans directly in Spring — it ensures queues, exchanges and bindings exist before any message is processed:

@Bean
public DirectExchange exchange() {
    return new DirectExchange(exchangeName);
}

@Bean
public Queue queueIn() {
    return QueueBuilder.durable(queueIn)
            .withArgument("x-dead-letter-exchange", dlxName)
            .withArgument("x-dead-letter-routing-key", dlqName)
            .build();
}

@Bean
public Binding bindingQueueIn(Queue queueIn, DirectExchange exchange) {
    return BindingBuilder.bind(queueIn).to(exchange).with(routingKey);
}

Note the withArgument calls — they configure the Dead Letter Queue directly in the main queue declaration. Any message that fails processing is automatically redirected.

2. The Matching Logic

The heart of the service. The rule is straightforward:

BUY: If the price the user accepts to pay >= market price → FILLED
SELL: If the price the user wants to receive <= market price → FILLED
Otherwise → REJECTED

private static boolean isCanExecute(OrderEventDTO order, BigDecimal marketPrice) {
    if (SideStatus.BUY.name().equalsIgnoreCase(order.getSide())) {
        return order.getPrice().compareTo(marketPrice) >= 0;
    } else if (SideStatus.SELL.name().equalsIgnoreCase(order.getSide())) {
        return order.getPrice().compareTo(marketPrice) <= 0;
    }
    return false;
}

3. Failure Handling — No Silent Order Loss

A silent bug I identified: when the ticker wasn't in Redis, the method simply returned without doing anything. The broker would wait for a response that never came.

The fix was simple but critical — any failure must notify the broker:

if (marketDataOpt.isEmpty()) {
    log.warn("Price for ticker {} not found in Redis. Rejecting order {}",
             order.getTicker(), order.getOrderId());
    OrderResponseEvent response = new OrderResponseEvent(
        order.getOrderId(), ExecutionStatus.REJECTED.name(), BigDecimal.ZERO);
    orderProducer.sendToBroker(response);
    return;
}

💡 This is exactly why we built b3-market-sync-api first. If Redis doesn't have the price, the order is immediately rejected — correct and predictable behavior.

4. Transactional Guarantee

Another critical point: persisting to PostgreSQL and publishing to RabbitMQ are two distinct operations. If the database saves but the publish fails, the execution is recorded but the broker is never notified — silent inconsistency.

The solution was adding @Transactional to the method that orchestrates both operations:

@Transactional
private void saveAndNotify(OrderEventDTO order, BigDecimal price, ExecutionStatus status) {
    // 1. Persists in PostgreSQL
    OrderExecution execution = OrderExecution.builder()
            .orderId(order.getOrderId())
            .ticker(order.getTicker())
            .side(SideStatus.valueOf(order.getSide()))
            .quantity(order.getQuantity())
            .executedPrice(price)
            .status(status)
            .build();
    repository.save(execution);

    // 2. Notifies the Broker
    OrderResponseEvent response = new OrderResponseEvent(
        order.getOrderId(), status.name(), price);
    orderProducer.sendToBroker(response);
}

5. Dead Letter Queue

For messages that fail in the consumer (unexpected processing error), we configured a full DLQ setup:

@Bean
public DirectExchange deadLetterExchange() {
    return new DirectExchange(dlxName);
}

@Bean
public Queue deadLetterQueue() {
    return QueueBuilder.durable(dlqName).build();
}

@Bean
public Binding bindingDlq(Queue deadLetterQueue, DirectExchange deadLetterExchange) {
    return BindingBuilder.bind(deadLetterQueue)
            .to(deadLetterExchange)
            .with(dlqName);
}

In the consumer, we rethrow the exception so RabbitMQ knows the message failed and routes it to the DLQ:

@RabbitListener(queues = "${app.rabbitmq.queue-in}")
public void receiveOrder(OrderEventDTO event) {
    log.info("New order received: ID {} | Ticker {} | Side {}",
            event.getOrderId(), event.getTicker(), event.getSide());
    try {
        matchingService.process(event);
    } catch (Exception e) {
        log.error("Failed to process order {}: {}", event.getOrderId(), e.getMessage(), e);
        throw e; // RabbitMQ routes to DLQ
    }
}

6. REST API + Swagger

Since the service exposes query endpoints, I configured Swagger UI from the start:

@Bean
public OpenAPI customOpenAPI() {
    return new OpenAPI()
            .info(new Info()
                    .title("B3 Matching Engine API")
                    .version("1.0.0")
                    .description("Simulates the B3 stock exchange matching engine..."));
}

Available endpoints:

Method	Endpoint	Description
GET	`/api/v1/executions`	List all executions
GET	`/api/v1/executions/order/{orderId}`	Find by order ID
GET	`/api/v1/executions/ticker/{ticker}`	Find by ticker
GET	`/api/v1/executions/status/{status}`	Find by status

✅ Validating the Execution

With the application running locally, all components started correctly:

✅ Flyway applied the order_executions table migration
✅ Hibernate validated the schema
✅ RabbitMQ connected with queues and DLQ declared
✅ Tomcat running on port 8091
✅ Swagger UI accessible at http://localhost:8091/swagger-ui.html

🚀 What's Next?

With b3-market-sync-api feeding Redis and b3-matching-engine-api ready to process orders, the entire B3 side is operational.

The next step is building the broker-order-api — the orchestrator on the broker side that will manage the complete order lifecycle:

Receive the buy/sell intent from the user
Validate the ticker via REST against broker-asset-api
Save the order as PENDING
Publish the event to Kafka
Send the order to the Matching Engine via RabbitMQ
Consume the feedback and update the final status

Once that service is ready, we'll have the first complete end-to-end flow running in the ecosystem.

Got any questions about the matching logic or RabbitMQ DLQ configuration? Drop them in the comments!

🔎 About the Series

⬅️ Previous Post: Syncing the Real Market: Consuming Brapi and Feeding Redis with Spring Boot

📘 Series Index: Series Roadmap

Links:

O Coração da B3: Construindo o Matching Engine com RabbitMQ, Redis e Spring Boot

Roberto de Vargas Neto — Sun, 26 Apr 2026 23:28:47 +0000

Olá, pessoal!

Dando continuidade à série My Broker B3, chegamos em um dos componentes mais aguardados do ecossistema: o B3 Matching Engine API.

No post anterior, construímos o b3-market-sync-api que sincroniza preços reais da brapi.dev e os armazena no Redis. Agora é hora de usar esses dados — este é o serviço que simula a própria Bolsa de Valores, decidindo se uma ordem de compra ou venda será executada ou rejeitada com base nos preços que acabamos de colocar no cache.

🏗️ O que é o Matching Engine?

No mundo real, quando você envia uma ordem de compra pela sua corretora, ela vai para a B3, que verifica se existe uma contraparte disposta a negociar naquele preço. Esse processo é chamado de matching.

No nosso simulador, o b3-matching-engine-api reproduz essa lógica de forma simplificada:

Consumo: Recebe a ordem da corretora via RabbitMQ (fila mq-broker-to-b3)
Preço: Busca o preço atual do ativo no Redis (market:price:{TICKER}) — dados injetados pelo Market Sync que construímos antes
Decisão: Aplica a regra de matching
Persistência: Grava o resultado no PostgreSQL
Notificação: Devolve o resultado para a corretora via RabbitMQ (fila mq-b3-to-broker)

[Broker] ──RabbitMQ──▶ [mq-broker-to-b3] ──▶ [Matching Engine]
                                                      │
                                             Redis: market:price:{TICKER}
                                             (alimentado pelo Market Sync ⬆️)
                                                      │
                                         ┌────────────┴────────────┐
                                      FILLED                   REJECTED
                                         │                         │
                                   PostgreSQL                 PostgreSQL
                                         │                         │
                               [mq-b3-to-broker] ◀────────────────┘
                                         │
                                      [Broker]

🎯 Foco no MVP

Antes de mergulhar no código, vale o mesmo disclaimer dos posts anteriores: estamos construindo a base. O objetivo é ter o fluxo ponta a ponta funcionando com robustez suficiente para validar a POC.

Nesta fase, priorizei:

Fluxo principal funcionando de ponta a ponta
Tratamento correto de falhas (sem ordens "sumindo" silenciosamente)
Dead Letter Queue para mensagens que falham no processamento
API REST para consulta do histórico de execuções
Documentação via Swagger

🛠️ Stack Tecnológica

Tecnologia	Uso
Java 21 + Spring Boot 3.5	Core do serviço
Spring RabbitMQ	Consumo de ordens e envio de resultados
Spring Data Redis	Consulta de preços em tempo real
Spring Data JPA + PostgreSQL	Persistência das execuções
Flyway	Versionamento do schema do banco
SpringDoc OpenAPI	Documentação via Swagger UI

🏗️ Os Pilares da Implementação

1. Configuração do RabbitMQ

Um ponto crítico que aprendi neste serviço: nunca dependa que as filas já existam no broker. Se a aplicação subir antes do RabbitMQ ter as filas criadas, o consumer falha na inicialização.

A solução é declarar todos os beans de infraestrutura diretamente no Spring — ele garante que filas, exchanges e bindings existam antes de qualquer mensagem trafegar:

@Bean
public DirectExchange exchange() {
    return new DirectExchange(exchangeName);
}

@Bean
public Queue queueIn() {
    return QueueBuilder.durable(queueIn)
            .withArgument("x-dead-letter-exchange", dlxName)
            .withArgument("x-dead-letter-routing-key", dlqName)
            .build();
}

@Bean
public Binding bindingQueueIn(Queue queueIn, DirectExchange exchange) {
    return BindingBuilder.bind(queueIn).to(exchange).with(routingKey);
}

Note o withArgument — ele já configura a Dead Letter Queue diretamente na declaração da fila principal. Qualquer mensagem que falhe no processamento é automaticamente redirecionada.

2. A Lógica de Matching

O coração do serviço. A regra é simples e direta:

COMPRA (BUY): Se o preço que o usuário aceita pagar >= preço de mercado → FILLED
VENDA (SELL): Se o preço que o usuário quer receber <= preço de mercado → FILLED
Caso contrário → REJECTED

private static boolean isCanExecute(OrderEventDTO order, BigDecimal marketPrice) {
    if (SideStatus.BUY.name().equalsIgnoreCase(order.getSide())) {
        return order.getPrice().compareTo(marketPrice) >= 0;
    } else if (SideStatus.SELL.name().equalsIgnoreCase(order.getSide())) {
        return order.getPrice().compareTo(marketPrice) <= 0;
    }
    return false;
}

3. Tratamento de Falhas — Sem Ordens Sumindo

Um bug silencioso que identifiquei: quando o ticker não estava no Redis, o método simplesmente retornava sem fazer nada. A corretora ficava esperando uma resposta que nunca chegava.

A correção foi simples mas fundamental — qualquer falha deve notificar o broker:

if (marketDataOpt.isEmpty()) {
    log.warn("Price for ticker {} not found in Redis. Rejecting order {}",
             order.getTicker(), order.getOrderId());
    OrderResponseEvent response = new OrderResponseEvent(
        order.getOrderId(), ExecutionStatus.REJECTED.name(), BigDecimal.ZERO);
    orderProducer.sendToBroker(response);
    return;
}

💡 Este é exatamente o motivo pelo qual construímos o b3-market-sync-api primeiro. Se o Redis não tiver o preço, a ordem é imediatamente rejeitada — comportamento correto e previsível.

4. Garantia Transacional

Outro ponto crítico: persistir no PostgreSQL e publicar no RabbitMQ são duas operações distintas. Se o banco salvar mas o publish falhar, a execução fica gravada mas o broker nunca é notificado — inconsistência silenciosa.

A solução foi adicionar @Transactional no método que orquestra as duas operações:

@Transactional
private void saveAndNotify(OrderEventDTO order, BigDecimal price, ExecutionStatus status) {
    // 1. Persists in PostgreSQL
    OrderExecution execution = OrderExecution.builder()
            .orderId(order.getOrderId())
            .ticker(order.getTicker())
            .side(SideStatus.valueOf(order.getSide()))
            .quantity(order.getQuantity())
            .executedPrice(price)
            .status(status)
            .build();
    repository.save(execution);

    // 2. Notifies the Broker
    OrderResponseEvent response = new OrderResponseEvent(
        order.getOrderId(), status.name(), price);
    orderProducer.sendToBroker(response);
}

5. Dead Letter Queue

Para mensagens que falham no consumer (erro inesperado no processamento), configuramos uma DLQ completa:

@Bean
public DirectExchange deadLetterExchange() {
    return new DirectExchange(dlxName);
}

@Bean
public Queue deadLetterQueue() {
    return QueueBuilder.durable(dlqName).build();
}

@Bean
public Binding bindingDlq(Queue deadLetterQueue, DirectExchange deadLetterExchange) {
    return BindingBuilder.bind(deadLetterQueue)
            .to(deadLetterExchange)
            .with(dlqName);
}

No consumer, relançamos a exceção para que o RabbitMQ saiba que a mensagem falhou e a redirecione:

@RabbitListener(queues = "${app.rabbitmq.queue-in}")
public void receiveOrder(OrderEventDTO event) {
    log.info("New order received: ID {} | Ticker {} | Side {}",
            event.getOrderId(), event.getTicker(), event.getSide());
    try {
        matchingService.process(event);
    } catch (Exception e) {
        log.error("Failed to process order {}: {}", event.getOrderId(), e.getMessage(), e);
        throw e; // RabbitMQ routes to DLQ
    }
}

6. API REST + Swagger

Como o serviço expõe endpoints de consulta, aproveitei para configurar o Swagger UI desde o início:

@Bean
public OpenAPI customOpenAPI() {
    return new OpenAPI()
            .info(new Info()
                    .title("B3 Matching Engine API")
                    .version("1.0.0")
                    .description("Simulates the B3 stock exchange matching engine..."));
}

Os endpoints disponíveis:

Método	Endpoint	Descrição
GET	`/api/v1/executions`	Lista todas as execuções
GET	`/api/v1/executions/order/{orderId}`	Busca por ID de ordem
GET	`/api/v1/executions/ticker/{ticker}`	Busca por ticker
GET	`/api/v1/executions/status/{status}`	Busca por status

✅ Validando a Execução

Com a aplicação rodando localmente, todos os componentes subiram corretamente:

✅ Flyway aplicou a migration da tabela order_executions
✅ Hibernate validou o schema
✅ RabbitMQ conectado com filas e DLQ declaradas
✅ Tomcat rodando na porta 8091
✅ Swagger UI acessível em http://localhost:8091/swagger-ui.html

🚀 O que vem a seguir?

Com o b3-market-sync-api alimentando o Redis e o b3-matching-engine-api pronto para processar ordens, toda a parte da B3 está operacional.

O próximo passo é construir o broker-order-api — o maestro que vai orquestrar todo o ciclo de vida de uma ordem do lado da corretora:

Receber a intenção de compra/venda do usuário
Validar o ticker via REST na broker-asset-api
Salvar a ordem como PENDING
Publicar o evento no Kafka
Enviar a ordem para o Matching Engine via RabbitMQ
Consumir o feedback e atualizar o status final

Quando esse serviço estiver pronto, teremos o primeiro fluxo completo ponta a ponta rodando no ecossistema.

Ficou com alguma dúvida sobre a lógica de matching ou sobre a configuração do RabbitMQ com DLQ? Deixe nos comentários!

🔎 Sobre a série

⬅️ Post Anterior: Sincronizando o Mercado Real: Consumindo a Brapi e Alimentando o Redis com Spring Boot

📘 Índice da Série: Guia da Série

Links:

Syncing the Real Market: Consuming Brapi and Feeding Redis with Spring Boot

Roberto de Vargas Neto — Sun, 26 Apr 2026 23:09:07 +0000

Hello everyone!

Continuing the My Broker B3 series, today we'll talk about an essential component on the B3 side of our simulator: the B3 Market Sync API.

Before building the engine that decides whether an order gets executed or rejected, we need to ensure it has access to real market prices. That's exactly what b3-market-sync-api does: fetches real asset prices from brapi.dev and stores them in Redis, ready for real-time consumption.

🏗️ What is the Market Sync?

This microservice has a single, well-defined responsibility: to be the market data producer of the ecosystem. It operates completely asynchronously and independently — no other service needs to call it directly.

The flow is simple and elegant:

[brapi.dev] ◀── Feign Client ── [MarketSyncScheduler]
                                         │
                              market:price:{TICKER}
                                         │
                                      [Redis]
                                         │
                              [B3 Matching Engine] (next post)

🎯 MVP Focus

The goal here is clear: have real prices available in Redis with low latency. The Matching Engine we'll build in the next post won't query a database or make HTTP calls to get prices — it goes straight to Redis. This ensures the matching decision happens in microseconds.

In this phase, I prioritized:

Scheduled sync with market hours guard
Redis cache with TTL to prevent stale data
External API rate limit protection
Critical bug fixes found during code review
REST API for querying cached prices
Swagger documentation

🛠️ Tech Stack

Technology	Usage
Java 21 + Spring Boot 3.5.11	Service core
Spring Cloud OpenFeign	Declarative HTTP client for brapi.dev
Spring Data Redis	High-performance cache
Spring Scheduling	Periodic synchronization
Jackson JSR310	Java 8 Date/Time serialization support
SpringDoc OpenAPI	Swagger UI documentation

🏗️ Implementation Pillars

1. Feign Client — Consuming Brapi

Spring Cloud OpenFeign lets you declare an HTTP client as a simple interface, with no boilerplate code:

@FeignClient(name = "brapiClient", url = "${app.brapi.url}")
public interface BrapiClient {

    @GetMapping("/quote/{ticker}")
    BrapiResponseDTO getQuote(
        @PathVariable String ticker,
        @RequestParam String token
    );
}

The URL and token are injected via application.yaml, keeping the code clean and environment-configurable.

2. Scheduler with Market Hours Guard

The heart of the service is a scheduled job that fires every 30 minutes. But before any API call, it checks whether the market is open:

@Scheduled(fixedRateString = "${app.sync.interval:1800000}")
public void sync() {
    if (!isMarketOpen()) {
        log.info("Sync aborted: market is closed (outside trading hours or weekend).");
        return;
    }
    // ... fetch and store prices
}

private boolean isMarketOpen() {
    ZonedDateTime now = ZonedDateTime.now(ZoneId.of("America/Sao_Paulo"));
    DayOfWeek day = now.getDayOfWeek();
    int hour = now.getHour();

    if (day == DayOfWeek.SATURDAY || day == DayOfWeek.SUNDAY) {
        return false;
    }

    // B3 trading hours: 10:00 to 18:00 (Sao Paulo time)
    return hour >= 10 && hour < 18;
}

Important detail: we use ZonedDateTime with America/Sao_Paulo explicitly. Inside a Docker container, the JVM may default to UTC — which would make the market hours guard behave incorrectly. We also pass -Duser.timezone=America/Sao_Paulo in the Dockerfile ENTRYPOINT to ensure the JVM honors the correct timezone:

ENTRYPOINT ["java", "-Duser.timezone=America/Sao_Paulo", "-jar", "app.jar"]

3. Redis Cache with Correct TTL

The MarketDataService stores each quote in Redis with a standardized key and 5-minute TTL:

private static final Duration CACHE_TTL = Duration.ofMinutes(5);

public void saveToCache(BrapiResultDTO quote) {
    String key = CACHE_KEY_PREFIX + quote.getTicker();
    // TTL of 5 minutes — if sync stops, stale prices expire quickly
    // preventing the Matching Engine from using outdated data
    redisTemplate.opsForValue().set(key, quote, CACHE_TTL);
    log.info("Ticker {} updated in cache: R$ {}", quote.getTicker(), quote.getRegularMarketPrice());
}

The TTL is fundamental: if the service goes down, prices expire in 5 minutes. The Matching Engine then fails to find the price in Redis and rejects orders — correct and predictable behavior, much better than using hours-old prices.

4. Epoch Timestamp Deserialization

A subtle bug I found: brapi.dev returns the regularMarketTime field as a Unix epoch (integer), but Jackson was trying to deserialize it as LocalDateTime (ISO string). This causes silent deserialization failure.

The solution was a custom deserializer:

public class EpochToLocalDateTimeDeserializer extends StdDeserializer<LocalDateTime> {

    private static final ZoneId SAO_PAULO = ZoneId.of("America/Sao_Paulo");

    @Override
    public LocalDateTime deserialize(JsonParser parser, DeserializationContext context)
            throws IOException {
        long epoch = parser.getLongValue();
        return LocalDateTime.ofInstant(Instant.ofEpochSecond(epoch), SAO_PAULO);
    }
}

And annotating it on the DTO:

@JsonDeserialize(using = EpochToLocalDateTimeDeserializer.class)
private LocalDateTime regularMarketTime;

5. Rate Limit Protection

Since we use the brapi.dev free plan, we added a delay between calls to avoid throttling:

for (String ticker : brapiProperties.getTickers()) {
    try {
        BrapiResponseDTO response = brapiClient.getQuote(ticker.trim(), brapiProperties.getToken());
        if (nonNull(response.getResults()) && !response.getResults().isEmpty()) {
            marketDataService.saveToCache(response.getResults().getFirst());
        }
        // Small delay between calls to avoid hitting rate limit on free plan
        Thread.sleep(200);
    } catch (Exception e) {
        log.error("Failed to sync ticker {}: {}", ticker, e.getMessage());
    }
}

The per-ticker try/catch is essential — if one asset fails, the loop continues for the rest.

6. REST API + Swagger

Although the Matching Engine consumes prices directly from Redis, we added REST endpoints to make debugging and observability easier:

Method	Endpoint	Description
GET	`/api/v1/quotes`	List prices for all configured tickers
GET	`/api/v1/quotes/{ticker}`	Get price for a specific ticker

Documented via Swagger UI at http://localhost:8096/swagger-ui.html.

🔒 Security — Token Out of Code

A critical issue I fixed: the brapi.dev token was hardcoded as a default value in application.yaml:

# ❌ Before — token exposed in public repository
token: ${BRAPI_TOKEN:mFk3HMLijAtxsP5y1ZjhsY}

# ✅ After — no fallback, variable is required
token: ${BRAPI_TOKEN}

Never expose credentials as default values in public repositories. The application now fails on startup if the variable is not provided — correct behavior.

✅ Validating the Execution

With the application running locally:

✅ Tomcat started on port 8096
✅ Application initialized in under 3 seconds
✅ Scheduler fired and logged: Sync aborted: market is closed (outside trading hours or weekend)
✅ Swagger UI accessible at http://localhost:8096/swagger-ui.html

🚀 What's Next?

With b3-market-sync-api up and feeding Redis with real prices, we have the foundation the next service needs to work. In the next post we'll build the B3 Matching Engine — the engine that will consume these prices and decide whether the broker's orders are executed or rejected.

Got any questions about the Redis integration or the market hours guard? Drop them in the comments!

🔎 About the Series

⬅️ Previous Post: From Stream to Database: Processing Market Data with Spring Boot, Redis and Flyway

➡️ Next Post: The Heart of B3: Building the Matching Engine with RabbitMQ and Redis (coming soon)

📘 Series Index: Series Roadmap

Links:

Sincronizando o Mercado Real: Consumindo a Brapi e Alimentando o Redis com Spring Boot

Roberto de Vargas Neto — Sun, 26 Apr 2026 23:04:02 +0000

Olá, pessoal!

Dando continuidade à série My Broker B3, hoje vamos falar sobre um componente essencial para o lado da B3 no nosso simulador: o B3 Market Sync API.

Antes de construirmos o motor que decide se uma ordem é executada ou rejeitada, precisamos garantir que ele tenha acesso a preços reais do mercado. É exatamente isso que o b3-market-sync-api faz: busca cotações reais de ativos brasileiros via brapi.dev e as armazena no Redis, pronto para consumo em tempo real.

🏗️ O que é o Market Sync?

Este microserviço tem uma responsabilidade única e bem definida: ser o produtor de dados de mercado do ecossistema. Ele opera de forma completamente assíncrona e independente — nenhum outro serviço precisa chamá-lo diretamente.

O fluxo é simples e elegante:

[brapi.dev] ◀── Feign Client ── [MarketSyncScheduler]
                                         │
                              market:price:{TICKER}
                                         │
                                      [Redis]
                                         │
                              [B3 Matching Engine] (próximo post)

🎯 Foco no MVP

Neste serviço o objetivo é claro: ter preços reais disponíveis no Redis com baixa latência. O Matching Engine que construiremos no próximo post não vai consultar banco de dados nem fazer chamadas HTTP para obter preços — ele vai direto no Redis. Isso garante que a decisão de match aconteça em microssegundos.

Nesta fase, priorizei:

Sincronização agendada com guard de horário de pregão
Cache Redis com TTL para evitar dados obsoletos
Proteção de rate limit da API externa
Correção de bugs críticos encontrados na revisão
API REST para consulta dos preços em cache
Documentação via Swagger

🛠️ Stack Tecnológica

Tecnologia	Uso
Java 21 + Spring Boot 3.5.11	Core do serviço
Spring Cloud OpenFeign	Cliente HTTP declarativo para brapi.dev
Spring Data Redis	Cache de alta performance
Spring Scheduling	Sincronização periódica
Jackson JSR310	Suporte a tipos de data Java 8
SpringDoc OpenAPI	Documentação via Swagger UI

🏗️ Os Pilares da Implementação

1. Feign Client — Consumindo a Brapi

O Spring Cloud OpenFeign permite declarar um cliente HTTP como uma simples interface, sem código boilerplate:

@FeignClient(name = "brapiClient", url = "${app.brapi.url}")
public interface BrapiClient {

    @GetMapping("/quote/{ticker}")
    BrapiResponseDTO getQuote(
        @PathVariable String ticker,
        @RequestParam String token
    );
}

A URL e o token são injetados via application.yaml, mantendo o código limpo e configurável por ambiente.

2. Scheduler com Guard de Pregão

O coração do serviço é um job agendado que roda a cada 30 minutos. Mas antes de qualquer chamada à API, ele verifica se o mercado está aberto:

@Scheduled(fixedRateString = "${app.sync.interval:1800000}")
public void sync() {
    if (!isMarketOpen()) {
        log.info("Sync aborted: market is closed (outside trading hours or weekend).");
        return;
    }
    // ... busca e salva os preços
}

private boolean isMarketOpen() {
    ZonedDateTime now = ZonedDateTime.now(ZoneId.of("America/Sao_Paulo"));
    DayOfWeek day = now.getDayOfWeek();
    int hour = now.getHour();

    if (day == DayOfWeek.SATURDAY || day == DayOfWeek.SUNDAY) {
        return false;
    }

    // B3 trading hours: 10:00 to 18:00 (Sao Paulo time)
    return hour >= 10 && hour < 18;
}

Detalhe importante: usamos ZonedDateTime com America/Sao_Paulo explicitamente. Dentro de um container Docker, a JVM pode usar UTC por padrão — o que faria o guard de horário funcionar errado. Além disso, passamos -Duser.timezone=America/Sao_Paulo no ENTRYPOINT do Dockerfile para garantir que a JVM também respeite o timezone correto:

ENTRYPOINT ["java", "-Duser.timezone=America/Sao_Paulo", "-jar", "app.jar"]

3. Cache Redis com TTL Correto

O MarketDataService salva cada cotação no Redis com uma chave padronizada e TTL de 5 minutos:

private static final Duration CACHE_TTL = Duration.ofMinutes(5);

public void saveToCache(BrapiResultDTO quote) {
    String key = CACHE_KEY_PREFIX + quote.getTicker();
    // TTL of 5 minutes — if sync stops, stale prices expire quickly
    // preventing the Matching Engine from using outdated data
    redisTemplate.opsForValue().set(key, quote, CACHE_TTL);
    log.info("Ticker {} updated in cache: R$ {}", quote.getTicker(), quote.getRegularMarketPrice());
}

O TTL é fundamental: se o serviço cair, os preços expiram em 5 minutos. O Matching Engine passa a não encontrar o preço no Redis e rejeita as ordens — comportamento correto e previsível, muito melhor do que usar preços de horas atrás.

4. Deserialização do Timestamp Epoch

Um bug sutil que encontrei: a brapi.dev retorna o campo regularMarketTime como epoch Unix (número inteiro), mas o Jackson tentava desserializar como LocalDateTime (string ISO). Isso causa falha silenciosa na desserialização.

A solução foi criar um deserializer customizado:

public class EpochToLocalDateTimeDeserializer extends StdDeserializer<LocalDateTime> {

    private static final ZoneId SAO_PAULO = ZoneId.of("America/Sao_Paulo");

    @Override
    public LocalDateTime deserialize(JsonParser parser, DeserializationContext context)
            throws IOException {
        long epoch = parser.getLongValue();
        return LocalDateTime.ofInstant(Instant.ofEpochSecond(epoch), SAO_PAULO);
    }
}

E anotá-lo no DTO:

@JsonDeserialize(using = EpochToLocalDateTimeDeserializer.class)
private LocalDateTime regularMarketTime;

5. Proteção de Rate Limit

Como utilizamos o plano gratuito da brapi.dev, implementamos um delay entre as chamadas para evitar throttling:

for (String ticker : brapiProperties.getTickers()) {
    try {
        BrapiResponseDTO response = brapiClient.getQuote(ticker.trim(), brapiProperties.getToken());
        if (nonNull(response.getResults()) && !response.getResults().isEmpty()) {
            marketDataService.saveToCache(response.getResults().getFirst());
        }
        // Small delay between calls to avoid hitting rate limit on free plan
        Thread.sleep(200);
    } catch (Exception e) {
        log.error("Failed to sync ticker {}: {}", ticker, e.getMessage());
    }
}

O try/catch por ticker é essencial — se um ativo falhar, o loop continua para os próximos.

6. API REST + Swagger

Embora o Matching Engine consuma os preços diretamente do Redis, adicionamos endpoints REST para facilitar debug e observabilidade:

Método	Endpoint	Descrição
GET	`/api/v1/quotes`	Lista preços de todos os tickers configurados
GET	`/api/v1/quotes/{ticker}`	Retorna o preço de um ticker específico

Documentados via Swagger UI em http://localhost:8096/swagger-ui.html.

🔒 Segurança — Token Fora do Código

Um problema crítico que corrigi: o token da brapi.dev estava hardcoded como valor padrão no application.yaml:

# ❌ Antes — token exposto no repositório público
token: ${BRAPI_TOKEN:mFk3HMLijAtxsP5y1ZjhsY}

# ✅ Depois — sem fallback, variável obrigatória
token: ${BRAPI_TOKEN}

Nunca exponha credenciais com valores default em repositórios públicos. A aplicação agora falha na inicialização se a variável não for fornecida — comportamento correto.

✅ Validando a Execução

Com a aplicação rodando localmente:

✅ Tomcat subiu na porta 8096
✅ Aplicação inicializou em menos de 3 segundos
✅ Scheduler disparou e exibiu: Sync aborted: market is closed (outside trading hours or weekend)
✅ Swagger UI acessível em http://localhost:8096/swagger-ui.html

🚀 O que vem a seguir?

Com o b3-market-sync-api pronto e alimentando o Redis com preços reais, temos a base que o próximo serviço precisa para funcionar. No próximo post vamos construir o B3 Matching Engine — o motor que vai consumir esses preços e decidir se as ordens da corretora serão executadas ou rejeitadas.

Ficou com alguma dúvida sobre a integração com Redis ou sobre o guard de horário de pregão? Deixe nos comentários!

🔎 Sobre a série

⬅️ Post Anterior: Do Stream para o Banco: Processando Market Data com Spring Boot, Redis e Flyway

➡️ Próximo Post: O Coração da B3: Construindo o Matching Engine com RabbitMQ e Redis (em breve)

📘 Índice da Série: Guia da Série

Links:

From Stream to Database: Processing Market Data with Spring Boot, Redis, and Flyway

Roberto de Vargas Neto — Wed, 01 Apr 2026 20:34:52 +0000

Hey folks!

Continuing the My Broker B3 series, today we build trading-broker-asset — the Java service that consumes the quotes published to Kafka by trading-broker-market-data and transforms them into structured data in MySQL, with high-performance caching in Redis.

If the Python service is the collector, this one is the processor.

🏗️ This Service's Role in the Ecosystem

[trading-broker-market-data]
         │
         │ Kafka: trading-assets-market-data-v1
         ▼
[trading-broker-asset]
         │
    ┌────┴────┐
    ▼         ▼
  MySQL     Redis
(assets)  market:price:{TICKER}
              │
    [b3-matching-engine]   (reads from here)
    [trading-broker-order] (validates ticker from here)

Two critical consumers depend on the data this service maintains: the Matching Engine (for price decisions) and the Order API (to validate whether a ticker exists and is active).

🎯 MVP Focus

In this phase, I prioritized:

Robust Kafka consumer with error handling
Asset upsert: creates on first event, updates on subsequent ones
Redis cache with 10-minute TTL
REST API for catalog queries
Status filter: only ACTIVE assets are returned

🛠️ Tech Stack

Technology	Usage
Java 21 + Spring Boot 3.5.11	Service core
Spring Kafka	Quote event consumption
MySQL + Flyway	Versioned asset catalog
Spring Data Redis	Real-time price cache
SpringDoc OpenAPI	Swagger UI documentation

🏗️ Implementation Pillars

1. The Database Schema (Flyway)

CREATE TABLE assets (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    ticker VARCHAR(20) NOT NULL UNIQUE,
    name VARCHAR(200),
    current_price DECIMAL(19, 4),
    last_update DATETIME,
    status ENUM('ACTIVE', 'INACTIVE') NOT NULL DEFAULT 'ACTIVE',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

Why Flyway? Schema versioning ensures that any environment (local, CI, production) has exactly the same database structure, without surprises.

2. The Kafka Consumer

@KafkaListener(
        topics = "trading-assets-market-data-v1",
        groupId = "trading-broker-asset"
)
public void consume(AssetMarketDataDTO dto) {
    log.info("Market data received for ticker: {}", dto.getTicker());
    try {
        assetService.updateAsset(dto);
    } catch (Exception e) {
        log.error("Failed to process market data for ticker {}: {}",
                dto.getTicker(), e.getMessage(), e);
        // Rethrow so Kafka can apply retry policy instead of silently discarding
        throw e;
    }
}

Why rethrow the exception? If processing fails silently, the offset is committed and the message is discarded — the asset is never updated. Rethrowing lets Kafka apply its retry policy.

3. The Asset Upsert

@Transactional
public void updateAsset(AssetMarketDataDTO dto) {
    Asset asset = assetRepository.findByTicker(dto.getTicker())
            .orElse(Asset.builder()
                    .ticker(dto.getTicker())
                    .status(AssetStatus.ACTIVE)
                    .build());

    asset.setName(dto.getName());
    asset.setCurrentPrice(dto.getPrice());
    // Uses real market timestamp from event — not LocalDateTime.now()
    asset.setLastUpdate(dto.getUpdatedAt() != null ? dto.getUpdatedAt() : LocalDateTime.now());

    assetRepository.save(asset);

    // Update Redis cache after persisting
    marketPriceCacheService.updatePrice(dto.getTicker(), dto.getPrice());
}

Important decision: we use the real updatedAt from the Kafka event, not LocalDateTime.now(). This ensures lastUpdate in the database reflects when the price was generated by the market, not when it was processed by the service.

4. Redis Cache with TTL

private static final Duration CACHE_TTL = Duration.ofMinutes(10);

public void updatePrice(String ticker, BigDecimal price) {
    String key = "market:price:" + ticker;
    // TTL of 10 minutes — prevents stale prices for inactive or delisted assets
    redisTemplate.opsForValue().set(key, price.toString(), CACHE_TTL);
}

Why TTL? If an asset is deactivated or delisted, its price will expire from Redis in 10 minutes. Without TTL, the Matching Engine could use prices for assets that no longer exist.

5. Status Filter in Queries

public List<AssetDTO> findAllActive() {
    // Query directly by status — not findAll() + filter in memory
    return assetRepository.findAllByStatus(AssetStatus.ACTIVE).stream()
            .map(AssetDTO::fromEntity)
            .toList();
}

public Optional<AssetDTO> findByTicker(String ticker) {
    // Only returns ACTIVE assets — inactive returns 404
    return assetRepository.findByTickerAndStatus(ticker.toUpperCase(), AssetStatus.ACTIVE)
            .map(AssetDTO::fromEntity);
}

When trading-broker-order validates a ticker before creating an order, it calls GET /api/v1/assets/{ticker}. If the asset is inactive, it gets a 404 and the order is rejected. Correct and predictable.

🌐 REST API

Method	Endpoint	Description
GET	`/api/v1/assets`	List all active assets
GET	`/api/v1/assets/{ticker}`	Find asset by ticker (404 if inactive)

📄 Swagger UI: http://localhost:8083/swagger-ui.html

✅ Validating the Execution

With the application running and trading-broker-market-data publishing to Kafka:

✅ Consumer connected to the trading-assets-market-data-v1 topic
✅ Assets being created/updated in MySQL in real time
✅ Redis being updated with TTL after each event
✅ Logs showing Updating ticker PETR4 with price R$ 47.37

🚀 What's Next?

With the asset catalog working, the next step is b3-market-sync-api — which synchronizes prices directly to the B3's Redis — followed by b3-matching-engine-api, which uses those prices to execute orders.

🔎 About the Series

⬅️ Previous Post: Tooling Tips: MongoDB and Kafka

➡️ Next Post: Syncing the Real Market: Consuming Brapi and Feeding Redis with Spring Boot

📘 Series Index: Series Roadmap

Links:

Do Stream para o Banco: Processando Market Data com Spring Boot, Redis e Flyway

Roberto de Vargas Neto — Sat, 28 Mar 2026 14:44:22 +0000

Olá, pessoal!

Continuando a série My Broker B3, hoje construímos o trading-broker-asset — o serviço Java que consome as cotações publicadas no Kafka pelo trading-broker-market-data e as transforma em dados estruturados no MySQL, com cache de alta performance no Redis.

Se o serviço Python é o coletor, este é o processador.

🏗️ O Papel deste Serviço no Ecossistema

[trading-broker-market-data]
         │
         │ Kafka: trading-assets-market-data-v1
         ▼
[trading-broker-asset]
         │
    ┌────┴────┐
    ▼         ▼
  MySQL     Redis
(assets)  market:price:{TICKER}
              │
    [b3-matching-engine]   (lê daqui)
    [trading-broker-order] (valida ticker daqui)

Dois consumers críticos dependem dos dados que este serviço mantém: o Matching Engine (para decisões de preço) e a Order API (para validar se um ticker existe e está ativo).

🎯 Foco no MVP

Nesta fase, priorizei:

Consumer Kafka robusto com tratamento de erros
Upsert de ativos: cria na primeira vez, atualiza nas seguintes
Cache Redis com TTL de 10 minutos
API REST para consulta do catálogo
Filtro de status: apenas ativos ACTIVE são retornados

🛠️ Stack Tecnológica

Tecnologia	Uso
Java 21 + Spring Boot 3.5.11	Core do serviço
Spring Kafka	Consumo de eventos de cotação
MySQL + Flyway	Catálogo de ativos versionado
Spring Data Redis	Cache de preços em tempo real
SpringDoc OpenAPI	Documentação via Swagger UI

🏗️ Os Pilares da Implementação

1. O Schema do Banco (Flyway)

CREATE TABLE assets (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    ticker VARCHAR(20) NOT NULL UNIQUE,
    name VARCHAR(200),
    current_price DECIMAL(19, 4),
    last_update DATETIME,
    status ENUM('ACTIVE', 'INACTIVE') NOT NULL DEFAULT 'ACTIVE',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

Por que Flyway? Versionamento do schema garante que qualquer ambiente (local, CI, produção) tenha exatamente a mesma estrutura de banco, sem surpresas.

2. O Consumer Kafka

@KafkaListener(
        topics = "trading-assets-market-data-v1",
        groupId = "trading-broker-asset"
)
public void consume(AssetMarketDataDTO dto) {
    log.info("Market data received for ticker: {}", dto.getTicker());
    try {
        assetService.updateAsset(dto);
    } catch (Exception e) {
        log.error("Failed to process market data for ticker {}: {}",
                dto.getTicker(), e.getMessage(), e);
        // Rethrow so Kafka can apply retry policy instead of silently discarding
        throw e;
    }
}

Por que relançar a exceção? Se o processamento falhar silenciosamente, o offset é confirmado e a mensagem é descartada — o ativo nunca é atualizado. Relançar permite que o Kafka aplique sua política de retry.

3. O Upsert de Ativos

@Transactional
public void updateAsset(AssetMarketDataDTO dto) {
    Asset asset = assetRepository.findByTicker(dto.getTicker())
            .orElse(Asset.builder()
                    .ticker(dto.getTicker())
                    .status(AssetStatus.ACTIVE)
                    .build());

    asset.setName(dto.getName());
    asset.setCurrentPrice(dto.getPrice());
    // Uses real market timestamp from event — not LocalDateTime.now()
    asset.setLastUpdate(dto.getUpdatedAt() != null ? dto.getUpdatedAt() : LocalDateTime.now());

    assetRepository.save(asset);

    // Update Redis cache after persisting
    marketPriceCacheService.updatePrice(dto.getTicker(), dto.getPrice());
}

Decisão importante: usamos o updatedAt real do evento Kafka, não LocalDateTime.now(). Isso garante que o lastUpdate no banco reflete quando o preço foi gerado pelo mercado, não quando foi processado pelo serviço.

4. O Cache Redis com TTL

private static final Duration CACHE_TTL = Duration.ofMinutes(10);

public void updatePrice(String ticker, BigDecimal price) {
    String key = "market:price:" + ticker;
    // TTL of 10 minutes — prevents stale prices for inactive or delisted assets
    redisTemplate.opsForValue().set(key, price.toString(), CACHE_TTL);
}

Por que TTL? Se um ativo for inativado ou deslistado, seu preço vai expirar do Redis em 10 minutos. Sem TTL, o Matching Engine poderia usar preços de ativos que não existem mais.

5. Filtro de Status nas Queries

public List<AssetDTO> findAllActive() {
    // Query directly by status — not findAll() + filter in memory
    return assetRepository.findAllByStatus(AssetStatus.ACTIVE).stream()
            .map(AssetDTO::fromEntity)
            .toList();
}

public Optional<AssetDTO> findByTicker(String ticker) {
    // Only returns ACTIVE assets — inactive returns 404
    return assetRepository.findByTickerAndStatus(ticker.toUpperCase(), AssetStatus.ACTIVE)
            .map(AssetDTO::fromEntity);
}

Quando a trading-broker-order valida um ticker antes de criar uma ordem, ela chama GET /api/v1/assets/{ticker}. Se o ativo estiver inativo, recebe 404 e rejeita a ordem. Correto e previsível.

🌐 API REST

Método	Endpoint	Descrição
GET	`/api/v1/assets`	Lista todos os ativos ativos
GET	`/api/v1/assets/{ticker}`	Busca ativo por ticker (404 se inativo)

📄 Swagger UI: http://localhost:8083/swagger-ui.html

✅ Validando a Execução

Com a aplicação rodando e o trading-broker-market-data publicando no Kafka:

✅ Consumer conectado ao tópico trading-assets-market-data-v1
✅ Ativos sendo criados/atualizados no MySQL em tempo real
✅ Redis sendo atualizado com TTL após cada evento
✅ Logs mostrando Updating ticker PETR4 with price R$ 47.37

🚀 O que vem a seguir?

Com o catálogo de ativos funcionando, o próximo passo é o b3-market-sync-api — que sincroniza os preços diretamente para o Redis da B3 — seguido do b3-matching-engine-api, que usa esses preços para executar as ordens.

🔎 Sobre a série

⬅️ Post Anterior: Dica de Ferramentas: MongoDB e Kafka

➡️ Próximo Post: Sincronizando o Mercado Real: Consumindo a Brapi e Alimentando o Redis com Spring Boot

📘 Índice da Série: Guia da Série

Links:

Tooling Tips: Visualizing Your Data in MongoDB and Kafka

Roberto de Vargas Neto — Sun, 15 Mar 2026 12:36:10 +0000

Hey folks!

Quick tooling post today. Before moving forward to the next services, I want to share the tools I use to visualize and inspect data in MongoDB and Kafka during development.

Without these tools, debugging an asynchronous data pipeline would be much harder — you're essentially "blind" to what's happening inside the containers.

🍃 MongoDB — MongoDB Compass

MongoDB Compass is MongoDB's official GUI. Free and available for Windows, Mac and Linux.

What you can do:

Browse collections and documents
Run queries with visual filters
See indexes created by Flyway/code
Inspect document schemas
Monitor performance metrics

How to connect to the local container:

URI: mongodb://root:password@localhost:27017/?authSource=admin

After connecting, you'll see the market_data_db database with the price_history collection being populated by broker-market-data-api on each ingestion round.

Download: mongodb.com/products/tools/compass

📨 Kafka — Offset Explorer (formerly Kafka Tool)

Offset Explorer is a GUI for inspecting Kafka clusters. It lets you see topics, partitions, messages and offsets without needing the command line.

What you can do:

List all topics in the cluster
See messages from each partition in real time
Inspect the content (key + value) of each message
Monitor consumer group offsets
Check whether consumers have lag

How to connect to the local container:

Bootstrap Servers: localhost:9092

After connecting, you'll see the trading-assets-market-data-v1 topic with messages published by broker-market-data-api. Each message will have the ticker as key (e.g. PETR4) and the quote JSON as value.

Download: kafkatool.com

🐰 RabbitMQ — Management UI (already included)

RabbitMQ comes with a built-in web interface. In our docker-compose.yml, port 15672 is exposed:

URL: http://localhost:15672
User: admin
Pass: admin_pass

What you can do:

View queues, exchanges and bindings
Inspect messages in queues
Manually publish messages to test consumers
Monitor message rates and lag
See the DLQ (Dead Letter Queue) when messages fail

This is the most useful tool for debugging the flow between trading-broker-order and b3-matching-engine.

💡 Workflow Tip

During development, I use all three tools in parallel:

Offset Explorer to confirm broker-market-data-api is publishing messages to Kafka
MongoDB Compass to confirm the history is being persisted
RabbitMQ Management to confirm orders are arriving in the B3 queues

This turns an opaque asynchronous pipeline into something completely observable.

🚀 What's Next?

With the tools set up, in the next post we build b3-market-sync — the Java service that consumes quotes from Kafka and synchronizes them to Redis, feeding the Matching Engine with real-time prices.

🔎 About the Series

⬅️ Previous Post: Market Data Integrator

➡️ Next Post: Syncing the Real Market: Consuming Brapi and Feeding Redis with Spring Boot

📘 Series Index: Series Roadmap

Links:

Dica de Ferramentas: Como Visualizar seus Dados no MongoDB e Kafka

Roberto de Vargas Neto — Sun, 15 Mar 2026 12:02:53 +0000

Olá, pessoal!

Post rápido de ferramentas hoje. Antes de avançarmos para os próximos serviços, quero compartilhar as ferramentas que uso para visualizar e inspecionar dados no MongoDB e no Kafka durante o desenvolvimento.

Sem essas ferramentas, depurar um pipeline de dados assíncrono seria muito mais difícil — você fica "cego" para o que está acontecendo dentro dos containers.

🍃 MongoDB — MongoDB Compass

O MongoDB Compass é a interface gráfica oficial do MongoDB. Gratuito e disponível para Windows, Mac e Linux.

O que você consegue fazer:

Navegar pelas collections e documentos
Executar queries com filtros visuais
Ver índices criados pelo Flyway/código
Inspecionar o schema dos documentos
Monitorar métricas de performance

Como conectar ao container local:

URI: mongodb://root:password@localhost:27017/?authSource=admin

Após conectar, você verá o banco market_data_db com a collection price_history sendo populada pelo trading-broker-market-data a cada rodada de ingestão.

Download: mongodb.com/products/tools/compass

📨 Kafka — Offset Explorer (antigo Kafka Tool)

O Offset Explorer é uma interface gráfica para inspecionar clusters Kafka. Permite ver tópicos, partições, mensagens e offsets sem precisar usar a linha de comando.

O que você consegue fazer:

Listar todos os tópicos do cluster
Ver as mensagens de cada partição em tempo real
Inspecionar o conteúdo (key + value) de cada mensagem
Monitorar offsets dos consumer groups
Verificar se os consumers estão com lag

Como conectar ao container local:

Bootstrap Servers: localhost:9092

Após conectar, você verá o tópico trading-assets-market-data-v1 com as mensagens publicadas pelo broker-market-data-api. Cada mensagem terá como key o ticker (ex: PETR4) e como value o JSON com a cotação.

Download: kafkatool.com

🐰 RabbitMQ — Management UI (já incluso)

O RabbitMQ já vem com uma interface web embutida. No nosso docker-compose.yml, a porta 15672 está exposta:

URL: http://localhost:15672
User: admin
Pass: admin_pass

O que você consegue fazer:

Ver filas, exchanges e bindings
Inspecionar mensagens nas filas
Publicar mensagens manualmente para testar consumers
Monitorar taxa de mensagens e lag
Ver a DLQ (Dead Letter Queue) quando mensagens falham

Esta é a ferramenta mais útil para debugar o fluxo entre o broker-order-api e o b3-matching-engine-api.

💡 Dica de Workflow

Durante o desenvolvimento, eu uso as três ferramentas em paralelo:

Offset Explorer para confirmar que o trading-broker-market-data está publicando mensagens no Kafka
MongoDB Compass para confirmar que o histórico está sendo persistido
RabbitMQ Management para confirmar que as ordens estão chegando nas filas da B3

Isso transforma um pipeline assíncrono opaco em algo completamente observável.

🚀 O que vem a seguir?

Com as ferramentas configuradas, no próximo post construímos o b3-market-sync — o serviço Java que consome as cotações do Kafka e as sincroniza para o Redis, alimentando o Matching Engine com preços em tempo real.

🔎 Sobre a série

⬅️ Post Anterior: Integrador de Market Data

➡️ Próximo Post: Sincronizando o Mercado Real: Consumindo a Brapi e Alimentando o Redis com Spring Boot

📘 Índice da Série: Guia da Série

Links: