Repository: apache/kafka
File: clients/src/test/java/org/apache/kafka/common/header/internals/RecordHeadersTest.java (Java)

Code before:
        assertThrows(NullPointerException.class, () -> recordHeaders.add(null));
    }

    @Test
    public void shouldThrowNpeWhenAddingCollectionWithNullHeader() {
        assertThrows(NullPointerException.class, () -> new RecordHeaders(new Header[1]));
    }

    private int getCount(Headers headers) {
        return headers.toArray().length;
    }

    static void assertHeader(String key, String value, Header actual) {
        assertEquals(key, actual.key());
        assertArrayEquals(value.getBytes(), actual.value());
    }

    @RepeatedTest(100)
    public void testRecordHeaderIsReadThreadSafe() throws Exception {
        int threads = 8;
        RecordHeader header = new RecordHeader(
            ByteBuffer.wrap("key".getBytes(StandardCharsets.UTF_8)),
            ByteBuffer.wrap("value".getBytes(StandardCharsets.UTF_8))
        );

        ExecutorService pool = Executors.newFixedThreadPool(threads);
        CountDownLatch startLatch = new CountDownLatch(1);
        AtomicBoolean raceDetected = new AtomicBoolean(false);

        try {
            Runnable task = () -> {
                try {
                    startLatch.await();
                    header.key();
                    header.value();
                } catch (NullPointerException e) {
                    raceDetected.set(true);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    throw new RuntimeException(e);
                }
            };

            for (int i = 0; i < threads; i++) pool.submit(task);

            startLatch.countDown();

            pool.shutdown();
            pool.awaitTermination(5, TimeUnit.SECONDS);

            assertFalse(raceDetected.get(), "Read race condition detected in RecordHeader!");

Review comment (suggestion):
Or we could leverage `CompletableFuture`

```java
        CountDownLatch startLatch = new CountDownLatch(1);
        var fs = IntStream.range(0, threads).mapToObj(__ -> CompletableFuture.supplyAsync(() -> {
            try {
                startLatch.await();
                header.key();
                header.value();
                return null;
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        })).collect(Collectors.toUnmodifiableList());
        startLatch.countDown();
        fs.forEach(CompletableFuture::join);
```

Code after:
    }

    @Test
    public void shouldThrowNpeWhenAddingCollectionWithNullHeader() {
        assertThrows(NullPointerException.class, () -> new RecordHeaders(new Header[1]));
    }

    private int getCount(Headers headers) {
        return headers.toArray().length;
    }

    static void assertHeader(String key, String value, Header actual) {
        assertEquals(key, actual.key());
        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
        );