Repository: apache/kafka
File: storage/src/test/java/org/apache/kafka/storage/internals/log/LogSegmentTest.java (Java)

Code before:
        assertEquals(2, segment.offsetIndex().entry(1).offset);

        assertEquals(2, segment.timeIndex().entries());
        assertEquals(new TimestampOffset(1, 0), segment.timeIndex().entry(0));
        assertEquals(new TimestampOffset(2, 2), segment.timeIndex().entry(1));
    }

    @Test
    @Timeout(30)
    public void testConcurrentAccessToMaxTimestampSoFar() throws Exception {
        int numThreads = 16;
        ExecutorService executor = Executors.newFixedThreadPool(numThreads);
        TimeIndex mockTimeIndex = mock(TimeIndex.class);
        when(mockTimeIndex.lastEntry()).thenReturn(new TimestampOffset(RecordBatch.NO_TIMESTAMP, 0L));

        try {
            // to reproduce race, we iterate test for certain duration
            long remainingDurationNanos = Duration.ofSeconds(1).toNanos();
            while (remainingDurationNanos > 0) {
                long t0 = System.nanoTime();
                clearInvocations(mockTimeIndex);
                try (LogSegment seg = spy(LogTestUtils.createSegment(0, logDir, 10, Time.SYSTEM))) {
                    when(seg.timeIndex()).thenReturn(mockTimeIndex);
                    List<Future<?>> futures = new ArrayList<>();
                    for (int i = 0; i < numThreads; i++) {
                        futures.add(executor.submit(() -> {
                            try {
                                seg.maxTimestampSoFar();
                            } catch (Exception e) {
                                throw new RuntimeException(e);
                            }
                        }));
                    }
                    for (Future<?> future : futures) {
                        future.get();
                    }
                    // timeIndex.lastEntry should be called once if no race
                    verify(mockTimeIndex, times(1)).lastEntry();

                    long elapsedNanos = System.nanoTime() - t0;
                    remainingDurationNanos -= elapsedNanos;
                }
            }
        } finally {
            executor.shutdown();
            executor.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
        }
    }

    private ProducerStateManager newProducerStateManager() throws IOException {
        return new ProducerStateManager(

Review comment (suggestion):
```java
futures.add(executor.submit(() -> assertDoesNotThrow(seg::maxTimestampSoFar)));
```

Code after:
        assertEquals(2, segment.offsetIndex().entry(1).offset);

        assertEquals(2, segment.timeIndex().entries());
        assertEquals(new TimestampOffset(1, 0), segment.timeIndex().entry(0));
        assertEquals(new TimestampOffset(2, 2), segment.timeIndex().entry(1));
    }

    @Test
    @Timeout(30)
    public void testConcurrentAccessToMaxTimestampSoFar() throws Exception {
        int numThreads = 16;
        ExecutorService executor = Executors.newFixedThreadPool(numThreads);
        TimeIndex mockTimeIndex = mock(TimeIndex.class);
        when(mockTimeIndex.lastEntry()).thenReturn(new TimestampOffset(RecordBatch.NO_TIMESTAMP, 0L));

        try {
            // to reproduce race, we iterate test for certain duration
            long remainingDurationNanos = Duration.ofSeconds(1).toNanos();
            while (remainingDurationNanos > 0) {
                long t0 = System.nanoTime();
                clearInvocations(mockTimeIndex);
                try (LogSegment seg = spy(LogTestUtils.createSegment(0, logDir, 10, Time.SYSTEM))) {
                    when(seg.timeIndex()).thenReturn(mockTimeIndex);
                    List<Future<?>> futures = new ArrayList<>();
                    for (int i = 0; i < numThreads; i++) {
                        futures.add(executor.submit(() -> assertDoesNotThrow(seg::maxTimestampSoFar)));
                    }
                    for (Future<?> future : futures) {
                        future.get();
                    }
                    // timeIndex.lastEntry should be called once if no race
                    verify(mockTimeIndex, times(1)).lastEntry();

                    long elapsedNanos = System.nanoTime() - t0;
                    remainingDurationNanos -= elapsedNanos;
                }
            }
        } finally {
            executor.shutdown();
            executor.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
        }
    }

    private ProducerStateManager newProducerStateManager() throws IOException {
        return new ProducerStateManager(
            topicPartition,
            logDir,
            (int) (Duration.ofMinutes(5).toMillis()),
            new ProducerStateManagerConfig(86400000, false),
            new MockTime()
        );