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

Code before:
    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) throws Exception {
        int threadCount = 32;
        CountDownLatch startLatch = new CountDownLatch(1);
        AtomicBoolean raceDetected = new AtomicBoolean(false);

        var futures = IntStream.range(0, threadCount)
            .mapToObj(i -> CompletableFuture.runAsync(() -> {
                try {
                    startLatch.await();
                    header.key();
                    header.value();
                } catch (NullPointerException e) {
                    raceDetected.set(true);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    throw new RuntimeException(e);
                }
            })).collect(Collectors.toUnmodifiableList());

        startLatch.countDown();
        futures.forEach(CompletableFuture::join);

        assertFalse(raceDetected.get());
    }

    @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)),

Review comment (suggestion):
we don't need `raceDetected`, since `join` could show the error.

Code after:
        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
        );
        assertRecordHeaderReadThreadSafe(header);
    }
}