Working with Message Headers
1. Understanding Headers Concept
| Aspect | Description | Detail |
|---|---|---|
| Header | Key (String) + value (byte[]) pair | Outside payload |
| Multiple | Repeated keys allowed | Iterable |
| Use | Metadata without touching value | Cross-cutting |
Example: Headers interface
Headers headers = record.headers();
for (Header h : headers) System.out.println(h.key());2. Adding Headers to Producer
| Method | Purpose | Detail |
|---|---|---|
| add(key, val) | Append a header | byte[] value |
| Constructor | Pass header list to record | At creation |
| Encoding | App serializes value | UTF-8 common |
Example: Add producer headers
ProducerRecord<String, String> r = new ProducerRecord<>("orders", "k", "v");
r.headers().add("event-type", "created".getBytes(StandardCharsets.UTF_8));
producer.send(r);3. Reading Headers in Consumer
| Method | Returns | Detail |
|---|---|---|
| headers() | All headers | Iterable |
| lastHeader(key) | Most recent value | Or null |
| headers(key) | All values for key | Iterable |
Example: Read a header
Header h = record.headers().lastHeader("event-type");
String type = h != null ? new String(h.value(), StandardCharsets.UTF_8) : null;4. Setting Custom Headers
| Header | Purpose | Example |
|---|---|---|
| correlation-id | Request correlation | UUID |
| content-type | Payload format | application/json |
| schema-version | Payload version | v2 |
Example: Multiple custom headers
r.headers()
.add("content-type", "application/json".getBytes())
.add("schema-version", "v2".getBytes());5. Using Headers for Routing
| Aspect | Description | Detail |
|---|---|---|
| Routing Key | Header drives downstream branch | No payload parse |
| Connect SMT | Route by header transform | Declarative |
| Filter | Skip records by header | Cheap check |
Example: Route by header
String region = new String(record.headers().lastHeader("region").value());
if ("eu".equals(region)) forwardToEu(record);6. Storing Metadata in Headers
| Metadata | Use | Detail |
|---|---|---|
| source-app | Origin service | Audit |
| timestamp | Producer time | Latency calc |
| retry-count | Attempt number | DLQ logic |
Example: Retry count header
int attempts = headerInt(record, "retry-count", 0);
r.headers().add("retry-count", String.valueOf(attempts + 1).getBytes());7. Filtering by Headers
| Aspect | Description | Detail |
|---|---|---|
| Pre-filter | Check header before processing | Skip early |
| No deserialize | Avoid value parse cost | Efficient |
| Streams | filter() on header predicate | Declarative |
Example: Filter by header
for (var r : records) {
if (r.headers().lastHeader("skip") != null) continue;
process(r);
}8. Understanding Header Serialization
| Aspect | Description | Detail |
|---|---|---|
| Value Type | Always byte[] | App encodes |
| No Schema | Headers are schemaless | Convention-based |
| Size | Counted in message size | Keep small |
Example: Encode/decode header value
byte[] val = ByteBuffer.allocate(8).putLong(System.currentTimeMillis()).array();
r.headers().add("ts", val);
long ts = ByteBuffer.wrap(r.headers().lastHeader("ts").value()).getLong();9. Propagating Headers in Streams
| Aspect | Description | Detail |
|---|---|---|
| Default | Headers carried to output | map preserves |
| Processor | Access via context().headers() | Read/modify |
| Joins | Header propagation varies | Verify behavior |
Example: Headers in processor
public void process(Record<String, String> record) {
record.headers().add("processed-by", "agg".getBytes());
context.forward(record);
}10. Adding Trace IDs
| Aspect | Description | Detail |
|---|---|---|
| traceparent | W3C trace context | OpenTelemetry |
| Propagation | Carried producer→consumer | End-to-end trace |
| Auto-instrument | OTel Kafka interceptors | Zero-code |