The tablet uses two locks with deliberately different scopes: + * + *
When both locks are needed, the only valid order is {@code kvStateLock -> mutationLock}. Lock + * upgrading and the reverse order are forbidden. In particular, a put that requests backpressure + * flushing must release both locks before requesting {@code kvStateLock}'s write lock, and + * callbacks that may enter Replica lifecycle code must run after both locks are released. + * + *
A logical flush holds both locks across every native RocksDB write and the corresponding + * {@code flushedLogOffset}/{@code rowCount} update. Consequently, a lookup can observe either the + * state before the flush or the completed state, but never an intermediate prefix of a segmented + * flush. Close holds {@code kvStateLock}'s write lock while closing RocksDB, which makes a + * point-lookup lease unnecessary. Resources that outlive one lock acquisition, such as scan + * iterators, must still retain a {@link ResourceGuard.Lease} for their full lifetime. + */ @ThreadSafe public final class KvTablet { private static final Logger LOG = LoggerFactory.getLogger(KvTablet.class); @@ -148,8 +178,18 @@ public final class KvTablet { private final KvFlushScheduler kvFlushScheduler; private final boolean closeFlushScheduler; - // A lock that guards all modifications to the kv. - private final ReadWriteLock kvLock = new ReentrantReadWriteLock(); + /** + * Guards RocksDB-visible state and the tablet lifecycle. Read holders may access a live RocksDB + * instance; write holders may atomically publish a flush, take a snapshot or close it. + */ + private final ReadWriteLock kvStateLock = new ReentrantReadWriteLock(); + + /** + * Serializes merge, auto-increment, pre-write-buffer mutation, WAL append and rollback. If + * acquired together with {@link #kvStateLock}, this lock must always be acquired second. + */ + private final Lock mutationLock = new ReentrantLock(); + private final LogFormat logFormat; private final KvFormat kvFormat; // defines how to merge rows on the same primary key @@ -174,13 +214,13 @@ public final class KvTablet { */ private volatile long flushedLogOffset = 0; - @GuardedBy("kvLock") + @GuardedBy("kvStateLock") private FlushState flushState = FlushState.IDLE; - @GuardedBy("kvLock") + @GuardedBy("kvStateLock") private long requestedFlushOffset = 0; - @GuardedBy("kvLock") + @GuardedBy("kvStateLock") private int flushRetryAttempts = 0; /** Invoked after each flush that made progress; set at construction time by the owner. */ @@ -190,7 +230,7 @@ public final class KvTablet { private volatile long rowCount; - @GuardedBy("kvLock") + @GuardedBy("kvStateLock") private volatile boolean isClosed = false; private KvTablet( @@ -465,10 +505,10 @@ public long getRowCount() { * ID range. This is used for snapshot and recovery to capture the state of the tablet at a * specific log offset. * - *
Note: this method must be called under the kvLock to ensure the consistency between the - * returned state and the log offset. + *
Note: this method must be called under the kvStateLock to ensure the consistency between
+ * the returned state and the log offset.
*/
- @GuardedBy("kvLock")
+ @GuardedBy("kvStateLock")
public TabletState getTabletState() {
return new TabletState(
flushedLogOffset,
@@ -514,103 +554,154 @@ public LogAppendInfo putAsLeader(KvRecordBatch kvRecords, @Nullable int[] target
public LogAppendInfo putAsLeader(
KvRecordBatch kvRecords, @Nullable int[] targetColumns, MergeMode mergeMode)
throws Exception {
- return inWriteLock(
- kvLock,
- () -> {
- rocksDBKv.checkIfRocksDBClosed();
-
- long pendingFlushBytesAfterWrite =
- kvPreWriteBuffer.pendingFlushBytes() + kvRecords.sizeInBytes();
-
- // Write-path admission gate: reject the request if accepting this batch would
- // push the buffered KV view beyond RocksDB's safe write budget.
- if (rocksDBKv.wouldExceedFlushBudget(pendingFlushBytesAfterWrite)) {
- requestFlushInternal(
- Math.max(requestedFlushOffset, logTablet.getHighWatermark()));
- throw new StorageBackpressureException(
- String.format(
- "Write rejected for %s: flush budget exceeded "
- + "(storage pressure or buffer size limit reached). "
- + "Retry after backoff.",
- tableBucket));
- }
-
- SchemaInfo schemaInfo = schemaGetter.getLatestSchemaInfo();
- Schema latestSchema = schemaInfo.getSchema();
- short latestSchemaId = (short) schemaInfo.getSchemaId();
- validateSchemaId(kvRecords.schemaId(), latestSchemaId);
-
- AutoIncrementUpdater currentAutoIncrementUpdater =
- autoIncrementManager.getUpdaterForSchema(kvFormat, latestSchemaId);
-
- // Validate targetColumns doesn't contain auto-increment column
- currentAutoIncrementUpdater.validateTargetColumns(targetColumns);
-
- // Determine the row merger based on mergeMode:
- // - DEFAULT: Use the configured merge engine (rowMerger)
- // - OVERWRITE: Bypass merge engine, use pre-created overwriteRowMerger
- // to directly replace values (for undo recovery scenarios)
- // We only support ADD COLUMN, so targetColumns is fine to be used directly.
- RowMerger currentMerger =
- (mergeMode == MergeMode.OVERWRITE)
- ? overwriteRowMerger.configureTargetColumns(
- targetColumns, latestSchemaId, latestSchema)
- : rowMerger.configureTargetColumns(
- targetColumns, latestSchemaId, latestSchema);
-
- RowType latestRowType = latestSchema.getRowType();
- WalBuilder walBuilder = createWalBuilder(latestSchemaId, latestRowType);
- walBuilder.setWriterState(kvRecords.writerId(), kvRecords.batchSequence());
- // we only support ADD COLUMN LAST, so the BinaryRow after RowMerger is
- // only has fewer ending columns than latest schema, so we pad nulls to
- // the end of the BinaryRow to get the latest schema row.
- PaddingRow latestSchemaRow = new PaddingRow(latestRowType.getFieldCount());
- // get offset to track the offset corresponded to the kv record
- long logEndOffsetOfPrevBatch = logTablet.localLogEndOffset();
-
- try {
- processKvRecords(
- kvRecords,
- kvRecords.schemaId(),
- currentMerger,
- currentAutoIncrementUpdater,
- walBuilder,
- latestSchemaRow,
- logEndOffsetOfPrevBatch);
-
- // There will be a situation that these batches of kvRecordBatch have not
- // generated any CDC logs, for example, when client attempts to delete
- // some non-existent keys or MergeEngineType set to FIRST_ROW. In this case,
- // we cannot simply return, as doing so would cause a
- // OutOfOrderSequenceException problem. Therefore, here we will build an
- // empty batch with lastLogOffset to 0L as the baseLogOffset is 0L. As doing
- // that, the logOffsetDelta in logRecordBatch will be set to 0L. So, we will
- // put a batch into file with recordCount 0 and offset plus 1L, it will
- // update the batchSequence corresponding to the writerId and also increment
- // the CDC log offset by 1.
- LogAppendInfo logAppendInfo = logTablet.appendAsLeader(walBuilder.build());
-
- // if the batch is duplicated, we should truncate the kvPreWriteBuffer
- // already written.
- if (logAppendInfo.duplicated()) {
- kvPreWriteBuffer.truncateTo(
- logEndOffsetOfPrevBatch, TruncateReason.DUPLICATED);
- }
- return logAppendInfo;
- } catch (Throwable t) {
- // While encounter error here, the CDC logs may fail writing to disk,
- // and the client probably will resend the batch. If we do not remove the
- // values generated by the erroneous batch from the kvPreWriteBuffer, the
- // retry-send batch will produce incorrect CDC logs.
- // TODO for some errors, the cdc logs may already be written to disk, for
- // those errors, we should not truncate the kvPreWriteBuffer.
- kvPreWriteBuffer.truncateTo(logEndOffsetOfPrevBatch, TruncateReason.ERROR);
- throw t;
- } finally {
- // deallocate the memory and arrow writer used by the wal builder
- walBuilder.deallocate();
- }
- });
+ try {
+ return inReadLock(
+ kvStateLock,
+ () ->
+ inLock(
+ mutationLock,
+ () -> {
+ checkIfClosed();
+
+ long pendingFlushBytesAfterWrite =
+ kvPreWriteBuffer.pendingFlushBytes()
+ + kvRecords.sizeInBytes();
+
+ // Write-path admission gate: reject the request if
+ // accepting this batch would
+ // push the buffered KV view beyond RocksDB's safe write
+ // budget.
+ if (rocksDBKv.wouldExceedFlushBudget(
+ pendingFlushBytesAfterWrite)) {
+ throw new StorageBackpressureException(
+ String.format(
+ "Write rejected for %s: flush budget exceeded "
+ + "(storage pressure or buffer size limit reached). "
+ + "Retry after backoff.",
+ tableBucket));
+ }
+
+ SchemaInfo schemaInfo = schemaGetter.getLatestSchemaInfo();
+ Schema latestSchema = schemaInfo.getSchema();
+ short latestSchemaId = (short) schemaInfo.getSchemaId();
+ validateSchemaId(kvRecords.schemaId(), latestSchemaId);
+
+ AutoIncrementUpdater currentAutoIncrementUpdater =
+ autoIncrementManager.getUpdaterForSchema(
+ kvFormat, latestSchemaId);
+
+ // Validate targetColumns doesn't contain auto-increment
+ // column
+ currentAutoIncrementUpdater.validateTargetColumns(
+ targetColumns);
+
+ // Determine the row merger based on mergeMode:
+ // - DEFAULT: Use the configured merge engine (rowMerger)
+ // - OVERWRITE: Bypass merge engine, use pre-created
+ // overwriteRowMerger
+ // to directly replace values (for undo recovery
+ // scenarios)
+ // We only support ADD COLUMN, so targetColumns is fine to
+ // be used directly.
+ RowMerger currentMerger =
+ (mergeMode == MergeMode.OVERWRITE)
+ ? overwriteRowMerger.configureTargetColumns(
+ targetColumns,
+ latestSchemaId,
+ latestSchema)
+ : rowMerger.configureTargetColumns(
+ targetColumns,
+ latestSchemaId,
+ latestSchema);
+
+ RowType latestRowType = latestSchema.getRowType();
+ WalBuilder walBuilder =
+ createWalBuilder(latestSchemaId, latestRowType);
+ walBuilder.setWriterState(
+ kvRecords.writerId(), kvRecords.batchSequence());
+ // we only support ADD COLUMN LAST, so the BinaryRow after
+ // RowMerger is
+ // only has fewer ending columns than latest schema, so we
+ // pad nulls to
+ // the end of the BinaryRow to get the latest schema row.
+ PaddingRow latestSchemaRow =
+ new PaddingRow(latestRowType.getFieldCount());
+ // get offset to track the offset corresponded to the kv
+ // record
+ long logEndOffsetOfPrevBatch =
+ logTablet.localLogEndOffset();
+
+ try {
+ processKvRecords(
+ kvRecords,
+ kvRecords.schemaId(),
+ currentMerger,
+ currentAutoIncrementUpdater,
+ walBuilder,
+ latestSchemaRow,
+ logEndOffsetOfPrevBatch);
+
+ // There will be a situation that these batches of
+ // kvRecordBatch have not
+ // generated any CDC logs, for example, when client
+ // attempts to delete
+ // some non-existent keys or MergeEngineType set to
+ // FIRST_ROW. In this case,
+ // we cannot simply return, as doing so would cause a
+ // OutOfOrderSequenceException problem. Therefore, here
+ // we will build an
+ // empty batch with lastLogOffset to 0L as the
+ // baseLogOffset is 0L. As doing
+ // that, the logOffsetDelta in logRecordBatch will be
+ // set to 0L. So, we will
+ // put a batch into file with recordCount 0 and offset
+ // plus 1L, it will
+ // update the batchSequence corresponding to the
+ // writerId and also increment
+ // the CDC log offset by 1.
+ LogAppendInfo logAppendInfo =
+ logTablet.appendAsLeader(walBuilder.build());
+
+ // if the batch is duplicated, we should truncate the
+ // kvPreWriteBuffer
+ // already written.
+ if (logAppendInfo.duplicated()) {
+ kvPreWriteBuffer.truncateTo(
+ logEndOffsetOfPrevBatch,
+ TruncateReason.DUPLICATED);
+ }
+ return logAppendInfo;
+ } catch (Throwable t) {
+ // While encounter error here, the CDC logs may fail
+ // writing to disk,
+ // and the client probably will resend the batch. If we
+ // do not remove the
+ // values generated by the erroneous batch from the
+ // kvPreWriteBuffer, the
+ // retry-send batch will produce incorrect CDC logs.
+ // TODO for some errors, the cdc logs may already be
+ // written to disk, for
+ // those errors, we should not truncate the
+ // kvPreWriteBuffer.
+ kvPreWriteBuffer.truncateTo(
+ logEndOffsetOfPrevBatch, TruncateReason.ERROR);
+ throw t;
+ } finally {
+ // deallocate the memory and arrow writer used by the
+ // wal builder
+ walBuilder.deallocate();
+ }
+ }));
+ } catch (StorageBackpressureException e) {
+ // Releasing the read lock before requesting a flush avoids a read-to-write lock
+ // upgrade, which ReentrantReadWriteLock does not support.
+ inWriteLock(
+ kvStateLock,
+ () ->
+ requestFlushInternal(
+ Math.max(requestedFlushOffset, logTablet.getHighWatermark())));
+ throw e;
+ }
}
private void validateSchemaId(short schemaIdOfNewData, short latestSchemaId) {
@@ -833,7 +924,7 @@ long localLogEndOffset() {
public void requestFlush(long exclusiveUpToLogOffset, FatalErrorHandler fatalErrorHandler) {
asyncFatalErrorHandler = fatalErrorHandler;
- inWriteLock(kvLock, () -> requestFlushInternal(exclusiveUpToLogOffset));
+ inWriteLock(kvStateLock, () -> requestFlushInternal(exclusiveUpToLogOffset));
}
/** Detaches or replaces the flush-complete listener set at construction time. */
@@ -848,15 +939,15 @@ public long getFlushedLogOffset() {
@VisibleForTesting
FlushState getFlushState() {
- return inReadLock(kvLock, () -> flushState);
+ return inReadLock(kvStateLock, () -> flushState);
}
@VisibleForTesting
void setFlushState(FlushState state) {
- inWriteLock(kvLock, () -> flushState = state);
+ inWriteLock(kvStateLock, () -> flushState = state);
}
- @GuardedBy("kvLock")
+ @GuardedBy("kvStateLock")
private void requestFlushInternal(long exclusiveUpToLogOffset) {
if (isClosed || exclusiveUpToLogOffset <= flushedLogOffset) {
return;
@@ -872,7 +963,7 @@ private void requestFlushInternal(long exclusiveUpToLogOffset) {
void requestFlushRetry() {
inWriteLock(
- kvLock,
+ kvStateLock,
() -> {
if (!isClosed && flushState == FlushState.STORAGE_BLOCKED) {
transitionFlushState(FlushState.STORAGE_BLOCKED, FlushState.QUEUED);
@@ -891,27 +982,29 @@ void runScheduledFlush() {
// target instead of chasing an ever-increasing requestedFlushOffset within one run.
long flushedOffsetBefore = flushedLogOffset;
try {
- // The whole prepare -> write -> complete sequence runs under the kvLock write lock,
- // so snapshots, scans and puts can only observe states where the RocksDB content
- // matches flushedLogOffset/rowCount. The RocksDB lease is acquired strictly inside
- // kvLock, keeping the lock order kvLock -> lease on every path.
- inWriteLock(kvLock, this::doScheduledFlush);
+ // Keep the complete logical flush hidden from lookups. writePreparedFlush may issue
+ // several native RocksDB writes, so releasing kvStateLock between segments would
+ // expose a partially applied flush. mutationLock also excludes every pre-write-buffer
+ // and auto-increment mutation while the flush state is being committed.
+ inWriteLock(kvStateLock, () -> inLock(mutationLock, this::doScheduledFlush));
} catch (StorageBackpressureException e) {
delayScheduledFlush(e);
} catch (Throwable t) {
failScheduledFlush(t);
} finally {
// Publish progress (including the completed prefix of a partially rejected run) so
- // the high watermark can advance. Runs outside kvLock because the listener acquires
+ // the high watermark can advance. Runs outside kvStateLock because the listener
+ // acquires
// the replica's leaderIsrUpdateLock while the write path acquires
- // leaderIsrUpdateLock -> kvLock; invoking it under kvLock would invert that order.
+ // leaderIsrUpdateLock -> kvStateLock; invoking it under kvStateLock would invert that
+ // order.
if (flushedLogOffset > flushedOffsetBefore) {
notifyFlushComplete();
}
}
}
- @GuardedBy("kvLock")
+ @GuardedBy("kvStateLock")
private void doScheduledFlush() throws Exception {
PreparedFlush preparedFlush = prepareScheduledFlush();
if (preparedFlush == null) {
@@ -929,7 +1022,7 @@ private void doScheduledFlush() throws Exception {
private boolean tryAcquireScheduledFlush() {
return inWriteLock(
- kvLock,
+ kvStateLock,
() -> {
if (isClosed || flushState != FlushState.QUEUED) {
return false;
@@ -939,7 +1032,7 @@ private boolean tryAcquireScheduledFlush() {
});
}
- @GuardedBy("kvLock")
+ @GuardedBy("kvStateLock")
private @Nullable PreparedFlush prepareScheduledFlush() {
if (isClosed) {
// close() already forced the state machine to its terminal IDLE state.
@@ -956,7 +1049,7 @@ private boolean tryAcquireScheduledFlush() {
} catch (IllegalStateException e) {
// Orphaned PREPARED entries from a previous incomplete flush cycle. Structurally
// unreachable now that every flush run completes or aborts all its prepared entries
- // before releasing kvLock; kept as defense in depth.
+ // before releasing kvStateLock; kept as defense in depth.
LOG.warn("Found orphaned PREPARED entries in {}, aborting.", tableBucket, e);
kvPreWriteBuffer.abortAllPrepared();
transitionFlushState(FlushState.RUNNING, FlushState.STORAGE_BLOCKED);
@@ -976,7 +1069,7 @@ private boolean tryAcquireScheduledFlush() {
* immediately after it lands, so {@code flushedLogOffset}/{@code rowCount} stay consistent with
* the RocksDB content even if a later segment is rejected by the no-slowdown gate.
*/
- @GuardedBy("kvLock")
+ @GuardedBy("kvStateLock")
private void writePreparedFlush(PreparedFlush preparedFlush) throws Exception {
List The task runs under {@code kvStateLock.write -> mutationLock}. Snapshot creation uses this
+ * boundary so its RocksDB checkpoint, flushed offset, row count and auto-increment range all
+ * describe the same state.
*
* @return An executor that wraps task execution within the lock for all modification to this
* tablet.
*/
public Executor getGuardedExecutor() {
- return runnable -> inWriteLock(kvLock, runnable::run);
+ return runnable -> inWriteLock(kvStateLock, () -> inLock(mutationLock, runnable::run));
}
// get from kv pre-write buffer first, if can't find, get from rocksdb
@@ -1157,11 +1259,19 @@ private byte[] getFromBufferOrKv(KvPreWriteBuffer.Key key) throws IOException {
return value.get();
}
+ /**
+ * Reads committed values directly from RocksDB.
+ *
+ * This method intentionally does not acquire {@code mutationLock} and does not inspect the
+ * pre-write buffer. It can therefore overlap a put while preserving the existing rule that
+ * unflushed WAL records are invisible to ordinary lookup clients. The {@code kvStateLock} read
+ * lock prevents concurrent logical flush publication and native RocksDB close.
+ */
public List A higher flush target received in {@code QUEUED}, {@code RUNNING}, or {@code
* STORAGE_BLOCKED} only advances {@code requestedFlushOffset}; it does not change the state or
* enqueue another task. Closing the tablet sets {@code isClosed} and terminates the state
- * machine. All transitions must happen while holding {@code kvLock} and are applied through
- * {@code transitionFlushState}, which enforces the source state of each transition.
+ * machine. All transitions must happen while holding {@code kvStateLock} and are applied
+ * through {@code transitionFlushState}, which enforces the source state of each transition.
*/
enum FlushState {
IDLE,
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/scan/ScannerContext.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/scan/ScannerContext.java
index 6347a3d4396..42552c4bc46 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/scan/ScannerContext.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/scan/ScannerContext.java
@@ -41,7 +41,17 @@
*
* Cursor methods ({@link #advance()}, {@link #isValid()}, {@link #currentValue()}) are
* single-threaded; callers must hold the {@link #tryAcquireForUse()} flag while mutating cursor or
- * sequence-id state. {@link #close()} is thread-safe and fences on that flag.
+ * sequence-id state. {@link #close()} is thread-safe: it immediately rejects new users and defers
+ * native-resource cleanup until the current user releases the cursor.
+ *
+ * The lease is required even though point lookups are protected from RocksDB close by {@code
+ * KvTablet}'s state lock: a scanner and its native snapshot/iterator live across multiple RPCs and
+ * therefore outlive any one state-lock acquisition. Closing follows a two-phase protocol. First,
+ * {@code closed} prevents another RPC from acquiring the cursor. Second, the thread that observes
+ * both {@code closed} and {@code inUse == false} closes the native resources exactly once. This may
+ * be the closing thread, the current RPC in {@link #releaseAfterUse()}, or an acquire racing with
+ * close. The protocol never force-clears {@code inUse}, because doing so could release the lease
+ * and close an iterator while JNI is still using it.
*/
@NotThreadSafe
public class ScannerContext implements Closeable {
@@ -66,10 +76,15 @@ public class ScannerContext implements Closeable {
/** Last-access wall-clock (ms); volatile for the TTL evictor. */
private volatile long lastAccessTime;
+ /** Set when closure is requested; prevents every subsequent cursor acquisition. */
private final AtomicBoolean closed = new AtomicBoolean(false);
+ /** Exclusive ownership of the native iterator by one RPC continuation. */
private final AtomicBoolean inUse = new AtomicBoolean(false);
+ /** Makes deferred native-resource cleanup idempotent across all close/release races. */
+ private final AtomicBoolean resourcesClosed = new AtomicBoolean(false);
+
public ScannerContext(
String scannerId,
TableBucket tableBucket,
@@ -171,7 +186,8 @@ public boolean isExpired(long ttlMs, long nowMs) {
/**
* Tries to claim exclusive use of the cursor. Returns {@code false} if another thread already
* holds it, or if {@link #close()} has been initiated. Every successful acquire MUST be paired
- * with {@link #releaseAfterUse()} in a {@code finally}, otherwise {@code close()} blocks.
+ * with {@link #releaseAfterUse()} in a {@code finally}, otherwise native-resource cleanup is
+ * deferred indefinitely after {@code close()}.
*/
public boolean tryAcquireForUse() {
if (closed.get()) {
@@ -180,28 +196,41 @@ public boolean tryAcquireForUse() {
if (!inUse.compareAndSet(false, true)) {
return false;
}
- // Re-check after the CAS in case close() flipped `closed` in between; let close() proceed.
+ // Re-check after the CAS in case close() flipped `closed` in between. close() may have
+ // observed inUse=true and deferred cleanup, so this path must finish it after releasing.
if (closed.get()) {
inUse.set(false);
+ closeResources();
return false;
}
return true;
}
/**
- * Releases the flag obtained via {@link #tryAcquireForUse()}. MUST be called BEFORE any path
- * that may trigger {@link #close()} on the same thread, otherwise {@code close()} self-
- * deadlocks on its inUse fence.
+ * Releases the flag obtained via {@link #tryAcquireForUse()}. If close was requested while this
+ * RPC owned the cursor, this method also performs the deferred native-resource cleanup.
*/
public void releaseAfterUse() {
inUse.set(false);
+ if (closed.get()) {
+ closeResources();
+ }
}
+ /**
+ * Prevents new cursor users immediately and closes resources when no RPC currently owns the
+ * cursor. This method does not block waiting for an in-flight RPC; that RPC completes cleanup
+ * from {@link #releaseAfterUse()}.
+ */
@Override
public void close() {
- if (closed.compareAndSet(false, true)) {
- // force close the inUse fence so any racing tryAcquireForUse() calls
- inUse.set(false);
+ if (closed.compareAndSet(false, true) && !inUse.get()) {
+ closeResources();
+ }
+ }
+
+ private void closeResources() {
+ if (resourcesClosed.compareAndSet(false, true)) {
IOUtils.closeQuietly(iterator);
IOUtils.closeQuietly(readOptions);
IOUtils.closeQuietly(() -> rocksDBKv.getDb().releaseSnapshot(snapshot));
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java
index 7b39bd8dacf..778e9883079 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java
@@ -21,6 +21,7 @@
import org.apache.fluss.config.Configuration;
import org.apache.fluss.config.MemorySize;
import org.apache.fluss.config.TableConfig;
+import org.apache.fluss.exception.FlussRuntimeException;
import org.apache.fluss.exception.InvalidTableException;
import org.apache.fluss.exception.InvalidTargetColumnException;
import org.apache.fluss.exception.OutOfOrderSequenceException;
@@ -91,7 +92,10 @@
import javax.annotation.Nullable;
import java.io.File;
+import java.io.IOException;
import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -100,9 +104,12 @@
import java.util.Map;
import java.util.Objects;
import java.util.Random;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -124,6 +131,9 @@
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.Fail.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.spy;
/** Test for {@link KvTablet}. */
class KvTabletTest {
@@ -718,6 +728,133 @@ void testPutWithMultiThread() throws Exception {
}
}
+ @Test
+ void testLookupCanRunWhilePutIsInProgress() throws Exception {
+ PhysicalTablePath physicalTablePath = PhysicalTablePath.of(TablePath.of("testDb", "t1"));
+ CountDownLatch putEnteredSchemaLookup = new CountDownLatch(1);
+ CountDownLatch continuePut = new CountDownLatch(1);
+ AtomicBoolean blockSchemaLookup = new AtomicBoolean();
+ schemaGetter =
+ new TestingSchemaGetter(new SchemaInfo(DATA1_SCHEMA_PK, schemaId)) {
+ @Override
+ public SchemaInfo getLatestSchemaInfo() {
+ if (blockSchemaLookup.get()) {
+ putEnteredSchemaLookup.countDown();
+ try {
+ continuePut.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+ return super.getLatestSchemaInfo();
+ }
+ };
+ logTablet = createLogTablet(tempLogDir, 0L, physicalTablePath);
+ kvTablet =
+ createKvTablet(
+ physicalTablePath,
+ logTablet.getTableBucket(),
+ logTablet,
+ tmpKvDir,
+ schemaGetter,
+ new HashMap<>());
+
+ KvRecordBatch kvRecordBatch =
+ kvRecordBatchFactory.ofRecords(
+ kvRecordFactory.ofRecord("k1", new Object[] {1, "v1"}));
+ blockSchemaLookup.set(true);
+ Future