How would you implement the propagation of business headers (x-session-id, x-client-id) across the microservices system via REST and Kafka?

To propagate business headers (for example, x-session-id, x-client-id) across a microservices system via REST and Kafka, the following approach can be used:

1. REST:

• Extract the necessary headers from the HTTP request at the entry point of each microservice (for example, in filters or middleware).
• Store them in a context (for example, ThreadLocal or a specialized request context).
• When calling downstream services via REST, add these headers to the outgoing HTTP requests.

2. Kafka:

• When publishing messages to Kafka, include these headers in the message metadata (for example, in Kafka ProducerRecord headers).
• When consuming messages, extract headers from Kafka headers and place them into the processing context.
• If necessary, pass them further in other messages or REST calls.

3. Implementation in Java:

• Use an interceptor or filter for REST (for example, Spring's HandlerInterceptor or OncePerRequestFilter) to automatically handle headers.
• For Kafka, use ProducerInterceptor and ConsumerInterceptor or wrappers around the producer and consumer to manage headers.

Example for REST (Spring):

public class BusinessHeadersInterceptor implements HandlerInterceptor {
    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) {
        String sessionId = request.getHeader("x-session-id");
        String clientId = request.getHeader("x-client-id");
        BusinessContext.setSessionId(sessionId);
        BusinessContext.setClientId(clientId);
        return true;
    }
}

// When calling downstream
HttpHeaders headers = new HttpHeaders();
headers.add("x-session-id", BusinessContext.getSessionId());
headers.add("x-client-id", BusinessContext.getClientId());
// ... use headers in RestTemplate or WebClient

Example for Kafka:

ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
record.headers().add("x-session-id", BusinessContext.getSessionId().getBytes(StandardCharsets.UTF_8));
record.headers().add("x-client-id", BusinessContext.getClientId().getBytes(StandardCharsets.UTF_8));
producer.send(record);

Thus, business headers pass through the entire call and message chain, ensuring traceability and context.