KAFKA-19407 Fix potential IllegalStateException when appending to timeIndex

[ocadaruma]

Summary

• Fix potential race condition in
  LogSegment#readMaxTimestampAndOffsetSoFar(), which may result in
  non-monotonic offsets and causes replication to stop.
• See https://issues.apache.org/jira/browse/KAFKA-19407 for the details
  how it happen.

Reviewers: Vincent PÉRICART [email protected], Jun Rao
[email protected], Chia-Ping Tsai [email protected]

[ocadaruma]

This PR is supposed to introduce no any new lock contention, because (potentially blocking) lazy timeIndex materialization has already exclusive control by lock in LazyIndex#get

[mauhiz]

Should this volatile field become an Atomic instead?

[ocadaruma]

Using Atomic doesn't make the thing simple, because we need double-check locking anyways to ensure:

• initialize the value only once at first time
• always return initialized value

[mauhiz]

I thought updateAndGet could work to this effect

maxTimestampAndOffsetSoFar.updateAndGet(t -> if (t == TimestampOffset.UNKNOWN) timeIndex().lastEntry() else t)

But I have not fully considered whether it would end up being slower on any microbenchmark. It just looked simpler.

[chia7712]

@ocadaruma nice find!

[chia7712]

The monitor lock of LogSegment may be used by external code. Perhaps it would be safer to use the monitor lock of an internal Object instead?

private final Object maxTimestampAndOffsetLock = new Object();

synchronized (maxTimestampAndOffsetLock) {
}

[ocadaruma]

fair enough. fixed.

[chia7712]

futures.add(executor.submit(() -> assertDoesNotThrow(seg::maxTimestampSoFar)));

[chia7712]

or we can use CompletableFuture to streamline the code.

        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);
                var futures = IntStream.range(0, numThreads).mapToObj(ignored -> CompletableFuture.runAsync(() -> assertDoesNotThrow(seg::maxTimestampSoFar))).toList();
                futures.forEach(CompletableFuture::join);
                // timeIndex.lastEntry should be called once if no race
                verify(mockTimeIndex, times(1)).lastEntry();

                long elapsedNanos = System.nanoTime() - t0;
                remainingDurationNanos -= elapsedNanos;
            }
        }

[ocadaruma]

thanks, fixed to use assertDoesNotThrow

[chia7712]

Could you please add comments for this lock? By default, LogSegment is not thread-safe so all modification should be executed within UnifiedLog lock. However, there is an exceptional path that could change the maxTimestampAndOffsetSoFar without UnifiedLog lock, which causes this concurrent issue ...

[ocadaruma]

Sounds good. Added a comment

[junrao]

@ocadaruma : Thanks for the PR. LGTM