KAFKA-12999 Make RecordHeader reads thread-safe (KIP-1205) (#20751)

This patch implements

[KIP-1205](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1205%3A+Improve+RecordHeader+to+be+Thread-Safe)
to address concurrency issues (NPE) related to the lazy initialization
of `RecordHeader` fields.

The added test verified that concurrent access to `RecordHeader.key()`
and `RecordHeader.value()` no longer throws `NullPointerException`

Reviewers: TaiJuWu <[email protected]>, Chia-Ping Tsai
 <[email protected]>

Author: mingyen066

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRecord.java

@@ -45,6 +45,11 @@
  * states. It is the responsibility of the user to ensure that multi-threaded access is properly synchronized.
  *
  * <p>
+ * However, each individual {@link org.apache.kafka.common.header.Header} instance
+ * is <b>read thread-safe</b>; that is, it is safe for multiple threads to read the same header's key or value concurrently
+ * as long as no thread modifies it.
+ *
+ * <p>
  * Refer to the {@link KafkaConsumer} documentation for more details on multi-threaded consumption and processing strategies.
  */
 public class ConsumerRecord<K, V> {

clients/src/main/java/org/apache/kafka/common/header/internals/RecordHeader.java

@@ -25,9 +25,9 @@
 
 public class RecordHeader implements Header {
     private ByteBuffer keyBuffer;
-    private String key;
-    private ByteBuffer valueBuffer;
-    private byte[] value;
+    private volatile String key;
+    private volatile ByteBuffer valueBuffer;
+    private volatile byte[] value;
 
     public RecordHeader(String key, byte[] value) {
         Objects.requireNonNull(key, "Null header keys are not permitted");
@@ -42,16 +42,24 @@ public RecordHeader(ByteBuffer keyBuffer, ByteBuffer valueBuffer) {
 
     public String key() {
         if (key == null) {
-            key = Utils.utf8(keyBuffer, keyBuffer.remaining());
-            keyBuffer = null;
+            synchronized (this) {
+                if (key == null) {
+                    key = Utils.utf8(keyBuffer, keyBuffer.remaining());
+                    keyBuffer = null;
+                }
+            }
         }
         return key;
     }
 
     public byte[] value() {
         if (value == null && valueBuffer != null) {
-            value = Utils.toArray(valueBuffer);
-            valueBuffer = null;
+            synchronized (this) {
+                if (value == null && valueBuffer != null) {
+                    value = Utils.toArray(valueBuffer);
+                    valueBuffer = null;
+                }
+            }
         }
         return value;
     }

clients/src/test/java/org/apache/kafka/common/header/internals/RecordHeadersTest.java

@@ -19,10 +19,17 @@
 import org.apache.kafka.common.header.Header;
 import org.apache.kafka.common.header.Headers;
 
+import org.junit.jupiter.api.RepeatedTest;
 import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
 import java.util.Iterator;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -265,4 +272,41 @@ static void assertHeader(String key, String value, Header actual) {
         assertArrayEquals(value.getBytes(), actual.value());
     }
 
+    private void assertRecordHeaderReadThreadSafe(RecordHeader header) {
+        int threadCount = 16;
+        CountDownLatch startLatch = new CountDownLatch(1);
+
+        var futures = IntStream.range(0, threadCount)
+            .mapToObj(i -> CompletableFuture.runAsync(() -> {
+                try {
+                    startLatch.await();
+                    header.key();
+                    header.value();
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    throw new RuntimeException(e);
+                }
+            })).collect(Collectors.toUnmodifiableList());
+
+        startLatch.countDown();
+        futures.forEach(CompletableFuture::join);
+    }
+
+    @RepeatedTest(100)
+    public void testRecordHeaderIsReadThreadSafe() throws Exception {
+        RecordHeader header = new RecordHeader(
+            ByteBuffer.wrap("key".getBytes(StandardCharsets.UTF_8)),
+            ByteBuffer.wrap("value".getBytes(StandardCharsets.UTF_8))
+        );
+        assertRecordHeaderReadThreadSafe(header);
+    }
+
+    @RepeatedTest(100)
+    public void testRecordHeaderWithNullValueIsReadThreadSafe() throws Exception {
+        RecordHeader header = new RecordHeader(
+            ByteBuffer.wrap("key".getBytes(StandardCharsets.UTF_8)),
+            null
+        );
+        assertRecordHeaderReadThreadSafe(header);
+    }
 }

docs/upgrade.html

@@ -25,6 +25,10 @@ <h5><a id="upgrade_4_2_0_from" href="#upgrade_4_2_0_from">Upgrading Servers to 4
 
 <h5><a id="upgrade_420_notable" href="#upgrade_420_notable">Notable changes in 4.2.0</a></h5>
 <ul>
+    <li>
+        The <code>org.apache.kafka.common.header.internals.RecordHeader</code> class has been updated to be read thread-safe. See <a href="https://cwiki.apache.org/confluence/x/nYmhFg">KIP-1205</a> for details.
+        In other words, each individual <code>Header</code> object within a <code>ConsumerRecord</code>'s <code>headers</code> can now be safely read from multiple threads concurrently.
+    </li>
     <li>
         The <code>org.apache.kafka.disallowed.login.modules</code> config was deprecated. Please use the <code>org.apache.kafka.allowed.login.modules</code>
         instead.