如何关联多个Kafka Topic？含业务场景说明

Nice question! Correlating distributed Kafka topics is a super common need in modern systems, and the core solution hinges on a consistent, unique business request identifier that follows your workflow from start to finish. Let’s break down exactly how to implement this for your specific topics:

1. Anchor Your Application Log Topic with a Request ID #

First, you need to embed a unique request-id into every log entry tied to the same business request:

• When an HTTP API request hits your service, generate a UUID (or another globally unique string) as the request-id, and drop it into log4j’s MDC (Mapped Diagnostic Context). This will automatically attach the ID to every log statement generated during that request’s processing.

  // Example: Set MDC at the start of your request handler
  String requestId = UUID.randomUUID().toString();
  MDC.put("request-id", requestId);
  
  // Don't forget to clear MDC when the request finishes to avoid leaks!
  try {
      // Process request logic here
  } finally {
      MDC.remove("request-id");
  }
  

• Update your log4j configuration to include the request-id in your log output. For log4j2, that might look like this:

  <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss} [%t] %-5level %c{1} - %msg | request-id: %X{request-id}%n"/>
  

Now every log entry (API req/res, warnings, exceptions) for the same request will share this ID, making it easy to group them later.

2. Link Your Command Topic to the Original Request #

Since commands can be generated minutes after the initial request, you need to ensure the request-id is preserved through your business logic:

• When your system generates a command to send to the second topic, include the original request-id as a field in the command message (either in the Kafka message header or the payload). For example, a JSON payload might look like:

  {
    "command_id": "cmd_789",
    "request_id": "req_456_uuid",
    "action": "process_payment",
    "details": {...}
  }
  

• Store the request-id alongside any business state related to the request (e.g., in your database, cache, or workflow engine) so when the command is triggered later, you can retrieve and attach the ID.

3. Extend the Pattern to Your Event Topic #

Even with limited details about the third event topic, the same rule applies:

• Any event generated as part of the original business request (whether directly from the API flow, or triggered by a command execution) must include the same request-id. This ensures events can be grouped with their corresponding logs and commands.

4. Practical Ways to Aggregate Correlated Data #

Once all topics have the request-id, you have two main paths to correlate the data:

• Real-Time Correlation: Use a stream processing framework like Kafka Streams or Apache Flink. You can create a stream job that joins all three topics using request-id as the key. For the command’s delayed arrival, set a window (e.g., 10 minutes) to wait for late commands before finalizing the aggregated view.
• Offline Correlation: If real-time isn’t a requirement, ingest all three topics into a data warehouse or analytics database (like ClickHouse or PostgreSQL). Then you can run SQL queries to group logs, commands, and events by request-id whenever you need to audit or analyze a specific request flow.

Bonus Tips #

• Make sure the request-id travels across service boundaries! If your request calls other services, pass the ID in HTTP headers (e.g., X-Request-ID) or RPC metadata so downstream services can attach it to their own logs/commands/events.
• If you can’t generate a global request-id for some edge cases, fall back to a business-specific unique identifier (like an order ID or user session ID) — just ensure it’s consistent across all related messages.

内容的提问来源于stack exchange，提问作者user432024