...

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

Describe the problems you are trying to solve.

Public Interfaces

Briefly list any new interfaces that will be introduced as part of this proposal or any existing interfaces that will be removed or changed. The purpose of this section is to concisely call out the public contract that will come along with this feature.

A public interface is any change to the following:

• Binary log format

• The network protocol and api behavior

• Any class in the public packages under clientsConfiguration, especially client configuration

  • org/apache/kafka/common/serialization

  • org/apache/kafka/common

  • org/apache/kafka/common/errors

  • org/apache/kafka/clients/producer

  • org/apache/kafka/clients/consumer (eventually, once stable)

• Monitoring

• Command line tools and arguments

• Anything else that will likely break existing users in some way when they upgrade

Proposed Changes

Describe the new thing you want to do in appropriate detail. This may be fairly extensive and have large subsections of its own. Or it may be a few sentences. Use judgement based on the scope of the change.

Compatibility, Deprecation, and Migration Plan

• What impact (if any) will there be on existing users?
• If we are changing behavior how will we phase out the older behavior?
• If we need special migration tools, describe them here.
• When will we remove the existing behavior?

Test Plan

Describe in few sentences how the KIP will be tested. We are mostly interested in system tests (since unit-tests are specific to implementation details). How will we know that the implementation works as expected? How will we know nothing broke?

Rejected Alternatives

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, which can lead to unexpected behavior.
So far, three concurrency-related issues (KAFKA-12999, KAFKA-17725, KAFKA-18470) have been reported in connection with RecordHeader data access.
Making RecordHeader thread-safe would help avoid user confusion and prevent similar issues in the future.

Public Interfaces

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

Proposed Changes

Use double-checked locking and volatile to make RecordHeader thread-safe.
This ensures that synchronization happens only during initialization, and once initialized, subsequent accesses do not acquire any locks.

Before

public class RecordHeader implements Header {
    private ByteBuffer keyBuffer;
    private String key;
    private ByteBuffer valueBuffer;
    private byte[] value;

    public synchronized String key() {
        if (key == null) {
            key = Utils.utf8(keyBuffer, keyBuffer.remaining());
            keyBuffer = null;
        }
        return key;
    }

    public synchronized byte[] value() {
        if (value == null && valueBuffer != null) {
            value = Utils.toArray(valueBuffer);
            valueBuffer = null;
        }
        return value;
    }
}

After

public class RecordHeader implements Header {
    private ByteBuffer keyBuffer;
    private volatile String key;
    private ByteBuffer valueBuffer;
    private volatile byte[] value;

    public String key() {
        if (key == null) {
            synchronized (this) {
                if (key == null) {
                    key = Utils.utf8(keyBuffer, keyBuffer.remaining());
                    keyBuffer = null;
                }
            }
        }
        return key;
    }

    public byte[] value() {
        if (value == null && valueBuffer != null) {
            synchronized (this) {
                if (value == null && valueBuffer != null) {
                    value = Utils.toArray(valueBuffer);
                    valueBuffer = null;
                }
            }
        }
        return value;
    }
}

JMH Benchmark: Double-Checked Locking vs. Full-Method Synchronization

Full-Method Synchronization means that the entire key()/value() method is synchronized.

Benchmark code

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

    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
    @Threads(8)
    public String benchmarkKey() {
        return header.key();
    }

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

Result

The benchmark was executed on an Apple M4 Max system with 46 GB RAM.

Double-Checked Locking is significantly faster than full-method synchronization.

Double-Checked Locking

Benchmark                             Mode  Cnt  Score   Error  Units
RecordHeaderBenchmark.benchmarkKey    avgt   15  0.854 ± 0.011  ns/op
RecordHeaderBenchmark.benchmarkValue  avgt   15  0.846 ± 0.005  ns/op

Full-Method Synchronization

Benchmark                             Mode  Cnt    Score    Error  Units
RecordHeaderBenchmark.benchmarkKey    avgt   15  244.625 ± 36.088  ns/op
RecordHeaderBenchmark.benchmarkValue  avgt   15  233.566 ± 49.321  ns/op

Compatibility, Deprecation, and Migration Plan

Making RecordHeader thread-safe does not break any compatibility.

Test Plan

Testing will be carried out using the JMH benchmark described above.
If the benchmark completes successfully without any NullPointerException, it demonstrates that RecordHeader is thread-safe.
Furthermore, it confirms that double-checked locking performs significantly better than full-method synchronization.

Rejected Alternatives

Full-Method Synchronization

Although it seems simple, it introduces significant overhead to key()andvalue() on every method invocationIf there are alternative ways of accomplishing the same thing, what were they? The purpose of this section is to motivate why the design is the way it is and not some other way.