Working with Message Headers

1. Understanding Headers Concept

Property	Description	Usage
key	Unique identifier for the header	Used to retrieve or set a specific header value
value	Value associated with the key	Can be any byte array or string representation

Example: Setting and Retrieving Headers

ProducerRecord<String, String> record = new ProducerRecord<>("topic", "key", "value");
record.headers().add("headerKey", "headerValue".getBytes(StandardCharsets.UTF_8));

Note: Headers are not part of the message payload but are sent alongside it.

2. Adding Headers to Producer

Property	Description	Example
key	Unique identifier for the message	`"user-123"`
value	Message payload (can be string or binary)	`"Hello, Kafka!"`
headers	Map of key-value pairs for metadata	`{ "contentType": "application/json" }`

Example: Adding Headers to Producer Record

const { Kafka } = require('kafkajs');

const kafka = new Kafka({ brokers: ['localhost:9092'] });
const producer = kafka.producer();

await producer.connect();

const message = {
  key: 'user-123',
  value: JSON.stringify({ name: 'John', age: 30 }),
  headers: {
    contentType: 'application/json',
    timestamp: Date.now().toString(),
    source: 'user-service'
  }
};

await producer.send({
  topic: 'user-events',
  messages: [message]
});

await producer.disconnect();

Note: Headers are stored as key-value pairs where both keys and values are strings or binary data. They are useful for carrying metadata such as content type, timestamps, or routing information without affecting the message payload.

3. Reading Headers in Consumer

Header Key	Value Type	Description
message.id	`string`	Unique identifier for the message
correlation.id	`string`	Identifier used to correlate requests and responses
timestamp	`long`	Timestamp when the message was created

Example: Reading Headers in a Kafka Consumer

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Header;

// Assuming a consumer is already configured
ConsumerRecord<String, String> record = consumer.poll(Duration.ofMillis(1000)).iterator().next();

// Read headers from the record
for (Header header : record.headers()) {
    String key = header.key();
    byte[] value = header.value();
    System.out.println(key + ": " + new String(value));
}

4. Setting Custom Headers

Property	Description	Example
key	Header key (string)	`"user-id"`
value	Header value (byte array or string)	`"12345"`

Example: Setting Custom Headers in Producer Record

const { Kafka } = require('kafkajs');

const kafka = new Kafka({ clientId: 'my-app', brokers: ['localhost:9092'] });
const producer = kafka.producer();

const message = {
  topic: 'my-topic',
  messages: [
    {
      value: 'Hello Kafka',
      headers: {
        'user-id': '12345',
        'timestamp': new Date().toISOString()
      }
    }
  ]
};

await producer.send(message);

Note: Header values must be serializable to bytes. Strings are automatically encoded using UTF-8.

5. Using Headers for Routing

Header Key	Value Type	Description	Usage Example
routing.key	`string`	Used to determine the partition for a message based on a hash of this key.	`headers.put("routing.key", "user-123");`
message.type	`string`	Indicates the type of message for routing logic in consumers.	`headers.put("message.type", "order.created");`

Example: Setting Routing Headers in Producer

ProducerRecord<String, String> record = new ProducerRecord<>(
    "orders-topic",
    "user-123",
    "{\"id\": \"order-456\", \"status\": \"created\"}"
);
record.headers().add("routing.key", "user-123".getBytes());
record.headers().add("message.type", "order.created".getBytes());

Example: Consuming with Header-Based Routing

ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, String> record : records) {
    String messageType = record.headers().lastHeader("message.type").value();
    if (messageType != null && messageType.equals("order.created")) {
        // Route to order processing logic
    }
}

6. Storing Metadata in Headers

Header Key	Value Type	Description
messageId	`string`	Unique identifier for the message
sourceSystem	`string`	System that originated the message
timestamp	`long`	Unix timestamp when the message was created
contentType	`string`	MIME type of the message payload (e.g., `application/json`)

Example: Setting Headers in Producer

ProducerRecord<String, String> record = new ProducerRecord<>(
    "my-topic",
    "key",
    "message-body"
);
record.headers().add("messageId", "msg-12345".getBytes());
record.headers().add("sourceSystem", "payment-service".getBytes());
record.headers().add("timestamp", String.valueOf(System.currentTimeMillis()).getBytes());
record.headers().add("contentType", "application/json".getBytes());

Example: Reading Headers in Consumer

consumer.subscribe(Arrays.asList("my-topic"));
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));

for (ConsumerRecord<String, String> record : records) {
    Headers headers = record.headers();
    Header messageIdHeader = headers.lastHeader("messageId");
    Header sourceSystemHeader = headers.lastHeader("sourceSystem");
    
    if (messageIdHeader != null) {
        String messageId = new String(messageIdHeader.value());
        System.out.println("Message ID: " + messageId);
    }
}

7. Filtering by Headers

Header Key	Value Type	Description	Usage Example
messageType	`string`	Indicates the type of message being sent	`{ "messageType": "order.created" }`
correlationId	`string`	Used to correlate messages across systems	`{ "correlationId": "req-12345" }`
sourceSystem	`string`	Identifies the system that originated the message	`{ "sourceSystem": "payment-service" }`

Example: Filtering Messages by Header

import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.clients.consumer.ConsumerRecord;

// Example filter logic in a consumer
public boolean shouldProcessMessage(ConsumerRecord<String, String> record) {
    Headers headers = record.headers();
    Header messageTypeHeader = headers.lastHeader("messageType");
    
    if (messageTypeHeader != null) {
        String messageType = new String(messageTypeHeader.value());
        return "order.created".equals(messageType);
    }
    
    return false;
}

Note: Header filtering is typically done at the consumer level, not within Kafka itself.

Header Filtering Best Practices

Use consistent naming conventions for header keys
Filter headers early in the processing pipeline to reduce unnecessary computation
Validate header values before using them in business logic

8. Understanding Header Serialization

Header Type	Description	Serialization Format	Notes
String	Text-based header values	UTF-8 encoded bytes	Most common for metadata like user IDs or session tokens
Integer	Numeric header values (32-bit)	Big-endian 4-byte binary	Used for counters or status codes
Long	Numeric header values (64-bit)	Big-endian 8-byte binary	Useful for timestamps or large IDs
Bytes	Arbitrary binary data	Raw byte array	For complex structured data or encrypted content

Example: Setting and Reading Headers in Java

import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.clients.producer.ProducerRecord;

// Create a producer record with headers
ProducerRecord<String, String> record = new ProducerRecord<>(
    "my-topic", 
    "key", 
    "message-value"
);

Headers headers = record.headers();
headers.add("user-id", "12345".getBytes(StandardCharsets.UTF_8));
headers.add("timestamp", String.valueOf(System.currentTimeMillis()).getBytes(StandardCharsets.UTF_8));

// Reading headers
for (Header header : record.headers()) {
    System.out.println(header.key() + ": " + new String(header.value()));
}

Note: Header keys must be unique within a record. Duplicate keys will overwrite previous values.

Warning: Headers are not part of the message payload and do not participate in compression or encryption. Use them carefully for sensitive data.

9. Propagating Headers in Streams

Property	Description	Example
headers	Collection of key-value pairs attached to each message	`headers.set('user-id', '12345')`
header.key	Unique identifier for a header	`headers.get('content-type')`
header.value	Value associated with a header key	`headers.get('correlation-id')`

Example: Setting Headers in a Stream Processor

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> sourceStream = builder.stream("input-topic");

KStream<String, String> processedStream = sourceStream
    .mapValues(value -> {
        // Create a new header with metadata
        Headers headers = new RecordHeaders();
        headers.add("processed-by", "stream-processor-v1".getBytes());
        headers.add("timestamp", String.valueOf(System.currentTimeMillis()).getBytes());

        // Attach headers to the record (in a real scenario, this would be part of the processor)
        return value;
    });

processedStream.to("output-topic");

Note: Headers are propagated through Kafka Streams operations but not automatically forwarded to downstream consumers unless explicitly handled.

Warning: Header keys must be unique within a message; duplicate keys may overwrite previous values.

Key Points

Headers are metadata attached to Kafka messages.
They can be used for routing, filtering, or logging purposes.
In Kafka Streams, headers can be added and accessed during processing.
Header propagation depends on the specific operation and configuration used in the stream topology.

10. Adding Trace IDs

Property	Description	Example
traceId	Unique identifier for tracking messages across services	`"traceId": "12345678-1234-1234-1234-123456789012"`

Example: Adding Trace ID to Kafka Message

import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.clients.producer.ProducerRecord;

ProducerRecord<String, String> record = new ProducerRecord<>(
    "my-topic", 
    "key", 
    "message-value"
);

Headers headers = record.headers();
headers.add("traceId", "12345678-1234-1234-1234-123456789012".getBytes());

// Send record with trace ID header
producer.send(record);

Note: Trace IDs should be globally unique to ensure proper correlation across distributed systems.