Status

Current state: "Under Discussion"Accepted

Discussion thread: https://lists.apache.org/thread/dlrkd9qnd2p4kysy26fygrh885mdm0lo

Vote thread: https://lists.apache.org/thread/0nvz4td8xvqrqkno9vlf4l6nf8xcvqz1Discussion thread: here

JIRA:

...

Lazy initialization for RecordHeader was introduced in KAFKA-10438, improving performance but also creating unexpected side effects.
Since the Consumer is not thread-safe, the same assumption naturally extends to ConsumerRecord. However, users often assume that read-only access across threads is safe.
With lazy initialization, this assumption no longer holds, and users may encounter NullPointerException.
So far, three concurrency-related issues (KAFKA-12999, KAFKA-17725, KAFKA-18470) have been reported in connection with respect to RecordHeader data access.
We can ensure that users have thread-safety guarantees when accessing RecordHeader in a read-only manner, eliminating the risk of NullPointerException.
(Note that because the Consumer is not thread-safe, if other threads can modify the RecordHeader through the Consumer, its value field in RecordHeader (a mutable byte[]), thread-safety can no longer be is not guaranteed.)

Public Interfaces

org.apache.kafka.common.header.internals.RecordHeader class will be updated to be thread-safe.

...

The benchmark was executed on an Apple M4 Max system with 48 GB RAM.
The thread-safe version adds a little overhead during the first initialization (key() 0.457 → 0.770 ns/op, value() 0.451 → 0.761 ns/op).
After the initial lazy initialization, subsequent accesses do not incur any locking or additional cost, so the steady-state performance remains essentially identical to the non-thread-safe version.

...

@State(Scope.Benchmark)
@Fork(value = 1)
@Warmup(iterations = 5)
@Measurement(iterations = 15)
@BenchmarkMode(Mode.AverageTime)
@OutputTimeUnit(TimeUnit.NANOSECONDS)
public class RecordHeaderSingleThreadBenchmark {

    private RecordHeader header;

    @Setup(Level.Iteration)
    public void setup() {
        byte[] valueBytes = new byte[1000];
        ByteBuffer keyBuffer = ByteBuffer.wrap("key".getBytes());
        ByteBuffer valueBuffer = ByteBuffer.wrap(valueBytes);
        header = new RecordHeader(keyBuffer, valueBuffer);
    }

    @Benchmark
    public String benchmarkKey() {
        return header.key();
    }

    @Benchmark
    public byte[] benchmarkValue() {
        return header.value();
    }
}

Result

The thread-safe double-checked locking implementation introduces negligible overhead (~0.3 ns) only during the first initialization of key or value

Current Implementation (non-thread-safe)

...

Making RecordHeader thread-safe does not break any compatibility.

Test Plan

...

• Unit tests will be written to verify that no NullPointerException occurs.

• The benchmark described above

...

• will also be included. Successful completion of the benchmark

...

• without any NullPointerException

...

• will demonstrate that RecordHeader is thread-safe.

...

• Additionally, the benchmark will confirm that the double-checked locking implementation performs significantly better than full-method synchronization.

Rejected Alternatives

Full-Method Synchronization

...