Current state: "Under Discussion"
Discussion thread: TBD
JIRA:
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
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.
org.apache.kafka.common.header.internals.RecordHeader class will be updated to be thread-safe.
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.
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;
}
}
|
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;
}
} |
Full-Method Synchronization means that the entire key()/value() method is synchronized.
@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();
}
} |
The benchmark was executed on an Apple M4 Max system with 46 GB RAM.
Double-Checked Locking is significantly faster than full-method synchronization.
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 |
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 |
Making RecordHeader thread-safe does not break any compatibility.
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.
Although it seems simple, it introduces significant overhead to key() and value() on every method invocation.