diff --git a/fluss-common/src/main/java/org/apache/fluss/memory/LazyMemorySegmentPool.java b/fluss-common/src/main/java/org/apache/fluss/memory/LazyMemorySegmentPool.java index cc19eaac02e..91717f51ba4 100644 --- a/fluss-common/src/main/java/org/apache/fluss/memory/LazyMemorySegmentPool.java +++ b/fluss-common/src/main/java/org/apache/fluss/memory/LazyMemorySegmentPool.java @@ -22,6 +22,8 @@ import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; import org.apache.fluss.exception.FlussRuntimeException; +import org.apache.fluss.exception.RecordTooLargeException; +import org.apache.fluss.exception.TimeoutException; import javax.annotation.concurrent.GuardedBy; import javax.annotation.concurrent.ThreadSafe; @@ -33,7 +35,9 @@ import java.util.ArrayList; import java.util.Collections; import java.util.Deque; +import java.util.LinkedHashSet; import java.util.List; +import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; @@ -66,6 +70,14 @@ public class LazyMemorySegmentPool implements MemorySegmentPool, Closeable { private volatile int pageUsage; + @GuardedBy("lock") + private final Set allocations = new LinkedHashSet<>(); + + private final Condition allocationChanged = lock.newCondition(); + + @GuardedBy("lock") + private int waitingAllocations; + @VisibleForTesting LazyMemorySegmentPool( int maxPages, int pageSize, long maxTimeToBlockMs, long perRequestMemorySize) { @@ -231,6 +243,7 @@ public void returnAll(List memory) { } pageUsage = newPageUsage; cachePages.addAll(memory); + allocationChanged.signalAll(); for (int i = 0; i < memory.size() && !waiters.isEmpty(); i++) { waiters.peekFirst().signal(); } @@ -260,6 +273,7 @@ public void close() { closed = true; cachePages.clear(); waiters.forEach(Condition::signal); + allocationChanged.signalAll(); }); } @@ -270,11 +284,144 @@ private void checkClosed() { } public int queued() { - return inLock(lock, waiters::size); + return inLock(lock, () -> waiters.size() + waitingAllocations); } @VisibleForTesting public List getAllCachePages() { return cachePages; } + + @Override + public MemoryAllocation newAllocation() { + return inLock( + lock, + () -> { + checkClosed(); + Allocation allocation = new Allocation(); + allocations.add(allocation); + return allocation; + }); + } + + /** Called only under memory pressure, with the pool lock held. */ + private void resolveAllocationDeadlock() { + int heldPages = 0; + Allocation victim = null; + for (Allocation allocation : allocations) { + int held = allocation.pages.size(); + heldPages += held; + if (held > 0) { + // An active owner can still finish, or an aborted owner is already unwinding. + if (allocation.pendingPages == 0 || allocation.aborted) { + return; + } + victim = allocation; + } + if (allocation.pendingPages > 0 && allocation.pendingPages <= maxPages - pageUsage) { + return; + } + } + // Pages outside allocation scopes may be returned independently. + if (heldPages == pageUsage && victim != null) { + // Registration order keeps older operations alive when holders block each other. + victim.aborted = true; + allocationChanged.signalAll(); + } + } + + private final class Allocation extends MemoryAllocation { + private int pendingPages; + private boolean aborted; + + private Allocation() { + super(LazyMemorySegmentPool.this); + } + + @Override + public List allocatePages(int required) throws IOException { + checkArgument(required > 0, "Requested pages must be positive."); + lock.lock(); + try { + checkAllocationOpen(); + if (required > maxPages - pages.size()) { + aborted = true; + throw new RecordTooLargeException( + "Memory allocation exceeds the memory pool capacity of " + + totalSize() + + " bytes: held pages=" + + pages.size() + + ", requested pages=" + + required + + ", page size=" + + pageSize); + } + if (required > maxPages - pageUsage) { + awaitPages(required); + } + lazilyAllocatePages(required); + List allocated = drain(required); + if (required == 1) { + pages.add(allocated.get(0)); + } else { + pages.addAll(allocated); + } + return allocated; + } finally { + lock.unlock(); + } + } + + private void awaitPages(int required) { + pendingPages = required; + waitingAllocations++; + long remaining = TimeUnit.MILLISECONDS.toNanos(maxTimeToBlockMs); + try { + while (required > maxPages - pageUsage) { + resolveAllocationDeadlock(); + checkAllocationOpen(); + if (remaining <= 0) { + throw new TimeoutException("Timed out waiting for memory allocation."); + } + remaining = allocationChanged.awaitNanos(remaining); + checkAllocationOpen(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new FlussRuntimeException(e); + } finally { + pendingPages = 0; + waitingAllocations--; + } + } + + private void checkAllocationOpen() { + checkClosed(); + if (closed) { + throw new IllegalStateException("Memory allocation is closed."); + } + if (aborted) { + // Use the existing retryable wire error so older clients can retry as well. + throw new TimeoutException( + "Memory allocation aborted because blocked allocations cannot make progress. " + + "Release the allocation and retry the operation."); + } + } + + @Override + public void returnAll(List memory) { + inLock(lock, () -> super.returnAll(memory)); + } + + @Override + public void close() { + inLock( + lock, + () -> { + super.close(); + allocations.remove(this); + allocationChanged.signalAll(); + }); + } + } } diff --git a/fluss-common/src/main/java/org/apache/fluss/memory/MemoryAllocation.java b/fluss-common/src/main/java/org/apache/fluss/memory/MemoryAllocation.java new file mode 100644 index 00000000000..3d32dafeaba --- /dev/null +++ b/fluss-common/src/main/java/org/apache/fluss/memory/MemoryAllocation.java @@ -0,0 +1,135 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.memory; + +import org.apache.fluss.annotation.Internal; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.IdentityHashMap; +import java.util.List; +import java.util.Set; + +import static org.apache.fluss.utils.Preconditions.checkState; + +/** + * Pages owned by one operation and returned together on close. Use with try-with-resources so + * failed or cancelled operations release their pages. An allocation has a single allocating thread; + * closing it must not race with that thread or with users of its pages. + */ +@Internal +public class MemoryAllocation implements MemorySegmentPool, AutoCloseable { + + private final MemorySegmentPool pool; + protected final List pages = new ArrayList<>(); + protected boolean closed; + + MemoryAllocation(MemorySegmentPool pool) { + this.pool = pool; + } + + @Override + public MemorySegment nextSegment() throws IOException { + return allocatePages(1).get(0); + } + + @Override + public List allocatePages(int required) throws IOException { + checkState(!closed, "Memory allocation is closed."); + List allocated = pool.allocatePages(required); + pages.addAll(allocated); + return allocated; + } + + @Override + public void returnPage(MemorySegment segment) { + returnAll(Collections.singletonList(segment)); + } + + @Override + public void returnAll(List memory) { + checkState(!closed, "Memory allocation is closed."); + if (memory.isEmpty()) { + return; + } + checkState(memory.size() <= pages.size(), "Returned more pages than this allocation owns."); + if (ownsAllInOrder(memory)) { + pool.returnAll(memory); + pages.clear(); + return; + } + + // Validate the entire return before publishing any page to the pool. + Set returned = Collections.newSetFromMap(new IdentityHashMap<>()); + for (MemorySegment page : memory) { + checkState(returned.add(page), "Page is returned more than once."); + } + int owned = 0; + for (MemorySegment page : pages) { + if (returned.contains(page)) { + owned++; + } + } + checkState(owned == returned.size(), "Page does not belong to this allocation."); + pool.returnAll(memory); + pages.removeIf(returned::contains); + } + + private boolean ownsAllInOrder(List memory) { + if (memory.size() != pages.size()) { + return false; + } + int index = 0; + for (MemorySegment page : memory) { + if (page != pages.get(index++)) { + return false; + } + } + return true; + } + + @Override + public int pageSize() { + return pool.pageSize(); + } + + @Override + public long totalSize() { + return pool.totalSize(); + } + + @Override + public int freePages() { + return pool.freePages(); + } + + @Override + public long availableMemory() { + return pool.availableMemory(); + } + + @Override + public void close() { + if (!closed) { + pool.returnAll(pages); + pages.clear(); + closed = true; + } + } +} diff --git a/fluss-common/src/main/java/org/apache/fluss/memory/MemorySegmentPool.java b/fluss-common/src/main/java/org/apache/fluss/memory/MemorySegmentPool.java index c36e44c5337..5d81dfb8dab 100644 --- a/fluss-common/src/main/java/org/apache/fluss/memory/MemorySegmentPool.java +++ b/fluss-common/src/main/java/org/apache/fluss/memory/MemorySegmentPool.java @@ -28,6 +28,15 @@ @Internal public interface MemorySegmentPool { + /** + * Opens an allocation whose pages are returned together on close. The caller must release + * references to those pages before closing it. Bounded pools may abort a blocked allocation to + * allow another allocation to finish; callers must unwind and close the aborted allocation. + */ + default MemoryAllocation newAllocation() { + return new MemoryAllocation(this); + } + /** * Get the page size of each page this pool holds. * diff --git a/fluss-common/src/test/java/org/apache/fluss/memory/MemoryAllocationTest.java b/fluss-common/src/test/java/org/apache/fluss/memory/MemoryAllocationTest.java new file mode 100644 index 00000000000..4d4bac1680b --- /dev/null +++ b/fluss-common/src/test/java/org/apache/fluss/memory/MemoryAllocationTest.java @@ -0,0 +1,391 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.memory; + +import org.apache.fluss.exception.FlussRuntimeException; +import org.apache.fluss.exception.RecordTooLargeException; +import org.apache.fluss.exception.TimeoutException; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +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 static org.apache.fluss.testutils.common.CommonTestUtils.retry; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Regression tests for batch allocation progress and cleanup. */ +class MemoryAllocationTest { + + @ParameterizedTest + @ValueSource(strings = {"ordered", "reversed", "partial", "single"}) + void testReturnOwnedPages(String order) throws Exception { + try (LazyMemorySegmentPool pool = new LazyMemorySegmentPool(4, 128, 1000, 128); + MemoryAllocation allocation = pool.newAllocation(); + MemoryAllocation other = pool.newAllocation()) { + List owned = allocation.allocatePages(3); + MemorySegment foreign = other.nextSegment(); + List returned; + switch (order) { + case "ordered": + returned = owned; + break; + case "reversed": + returned = Arrays.asList(owned.get(2), owned.get(1), owned.get(0)); + break; + case "partial": + returned = Arrays.asList(owned.get(2), owned.get(0)); + break; + default: + returned = Collections.singletonList(owned.get(1)); + } + allocation.returnAll(returned); + assertThat(pool.freePages()).isEqualTo(returned.size()); + List reused = allocation.allocatePages(returned.size()); + assertThat(reused) + .containsExactlyInAnyOrderElementsOf(returned) + .doesNotContain(foreign); + allocation.close(); + assertThat(pool.freePages()).isEqualTo(3); + other.close(); + assertThat(pool.freePages()).isEqualTo(4); + } + } + + @ParameterizedTest + @ValueSource(strings = {"duplicate", "foreign", "partial-duplicate", "partial-foreign"}) + void testInvalidReturnLeavesOwnershipUnchanged(String invalid) throws Exception { + try (LazyMemorySegmentPool pool = new LazyMemorySegmentPool(4, 128, 1000, 128); + MemoryAllocation allocation = pool.newAllocation(); + MemoryAllocation other = pool.newAllocation()) { + List owned = allocation.allocatePages(3); + MemorySegment foreign = other.nextSegment(); + List returned; + switch (invalid) { + case "duplicate": + returned = Arrays.asList(owned.get(0), owned.get(0), owned.get(2)); + break; + case "foreign": + returned = Arrays.asList(owned.get(0), owned.get(1), foreign); + break; + case "partial-duplicate": + returned = Arrays.asList(owned.get(0), owned.get(0)); + break; + default: + returned = Arrays.asList(owned.get(0), foreign); + } + assertThatThrownBy(() -> allocation.returnAll(returned)) + .isInstanceOf(IllegalStateException.class); + assertThat(pool.freePages()).isZero(); + allocation.returnAll(owned); + assertThat(pool.freePages()).isEqualTo(3); + allocation.close(); + assertThat(pool.freePages()).isEqualTo(3); + other.returnPage(foreign); + assertThat(pool.freePages()).isEqualTo(4); + } + } + + @Test + void testCumulativeAllocationExceedsCapacity() throws Exception { + try (LazyMemorySegmentPool pool = pool()) { + assertThatThrownBy( + () -> { + try (MemoryAllocation allocation = pool.newAllocation()) { + allocation.nextSegment(); + allocation.nextSegment(); + allocation.nextSegment(); + } + }) + .isInstanceOf(RecordTooLargeException.class); + assertThat(pool.freePages()).isEqualTo(2); + try (MemoryAllocation retry = pool.newAllocation()) { + assertThat(retry.allocatePages(2)).hasSize(2); + } + assertThat(pool.freePages()).isEqualTo(2); + } + } + + @Test + void testWaitForActiveOwner() throws Exception { + ExecutorService executor = Executors.newSingleThreadExecutor(); + try (LazyMemorySegmentPool pool = pool(); + MemoryAllocation owner = pool.newAllocation()) { + owner.allocatePages(2); + Future waiter = + executor.submit( + () -> { + try (MemoryAllocation allocation = pool.newAllocation()) { + return allocation.nextSegment(); + } + }); + awaitWaiters(pool, 1); + assertThat(waiter.isDone()).isFalse(); + owner.close(); + assertThat(waiter.get(10, TimeUnit.SECONDS)).isNotNull(); + assertThat(pool.freePages()).isEqualTo(2); + } finally { + executor.shutdownNow(); + } + } + + @Test + void testDeadlockAbortsYoungerHolderAndRetrySucceeds() throws Exception { + ExecutorService executor = Executors.newSingleThreadExecutor(); + try (LazyMemorySegmentPool pool = pool(); + MemoryAllocation older = pool.newAllocation(); + MemoryAllocation younger = pool.newAllocation()) { + older.nextSegment(); + younger.nextSegment(); + Future completion = + executor.submit( + () -> { + try (MemoryAllocation allocation = older) { + return allocation.nextSegment(); + } + }); + awaitWaiters(pool, 1); + assertThatThrownBy(younger::nextSegment) + .isInstanceOf(TimeoutException.class) + .hasMessageContaining("cannot make progress"); + // Arbitration must not recycle pages that the aborted caller may still access. + assertThat(pool.freePages()).isZero(); + younger.close(); + assertThat(completion.get(10, TimeUnit.SECONDS)).isNotNull(); + try (MemoryAllocation retry = pool.newAllocation()) { + assertThat(retry.allocatePages(2)).hasSize(2); + } + assertThat(pool.freePages()).isEqualTo(2); + assertThat(pool.queued()).isZero(); + } finally { + executor.shutdownNow(); + } + } + + @Test + void testDeadlockWakesAlreadyWaitingVictim() throws Exception { + ExecutorService executor = Executors.newSingleThreadExecutor(); + try (LazyMemorySegmentPool pool = pool(); + MemoryAllocation older = pool.newAllocation(); + MemoryAllocation younger = pool.newAllocation()) { + older.nextSegment(); + younger.nextSegment(); + Future victim = + executor.submit( + () -> { + try (MemoryAllocation allocation = younger) { + assertThatThrownBy(allocation::nextSegment) + .isInstanceOf(TimeoutException.class); + } + }); + awaitWaiters(pool, 1); + assertThat(older.nextSegment()).isNotNull(); + victim.get(10, TimeUnit.SECONDS); + older.close(); + assertThat(pool.freePages()).isEqualTo(2); + assertThat(pool.queued()).isZero(); + } finally { + executor.shutdownNow(); + } + } + + @Test + void testWaitForUnscopedPagesAndPartialReturn() throws Exception { + ExecutorService executor = Executors.newSingleThreadExecutor(); + try (LazyMemorySegmentPool pool = pool(); + MemoryAllocation allocation = pool.newAllocation()) { + List external = pool.allocatePages(1); + MemorySegment first = allocation.nextSegment(); + Future waiter = executor.submit(allocation::nextSegment); + awaitWaiters(pool, 1); + assertThat(waiter.isDone()).isFalse(); + pool.returnAll(external); + assertThat(waiter.get(10, TimeUnit.SECONDS)).isNotNull(); + allocation.returnPage(first); + assertThat(pool.freePages()).isEqualTo(1); + allocation.close(); + allocation.close(); + assertThat(pool.freePages()).isEqualTo(2); + } finally { + executor.shutdownNow(); + } + } + + @Test + void testTimeoutReturnsHeldPages() throws Exception { + try (LazyMemorySegmentPool pool = new LazyMemorySegmentPool(2, 128, 10, 128); + MemoryAllocation active = pool.newAllocation()) { + active.nextSegment(); + assertThatThrownBy( + () -> { + try (MemoryAllocation allocation = pool.newAllocation()) { + allocation.nextSegment(); + allocation.nextSegment(); + } + }) + .isInstanceOf(TimeoutException.class) + .hasMessageContaining("Timed out"); + assertThat(pool.freePages()).isEqualTo(1); + assertThat(pool.queued()).isZero(); + } + } + + @Test + void testEmptyWaiterDoesNotPreventDeadlockRecovery() throws Exception { + ExecutorService executor = Executors.newFixedThreadPool(2); + try (LazyMemorySegmentPool pool = pool(); + MemoryAllocation older = pool.newAllocation(); + MemoryAllocation younger = pool.newAllocation()) { + older.nextSegment(); + younger.nextSegment(); + Future empty = + executor.submit( + () -> { + try (MemoryAllocation allocation = pool.newAllocation()) { + return allocation.nextSegment(); + } + }); + awaitWaiters(pool, 1); + Future survivor = + executor.submit( + () -> { + try (MemoryAllocation allocation = older) { + return allocation.nextSegment(); + } + }); + awaitWaiters(pool, 2); + assertThatThrownBy(younger::nextSegment).isInstanceOf(TimeoutException.class); + younger.close(); + assertThat(empty.get(10, TimeUnit.SECONDS)).isNotNull(); + assertThat(survivor.get(10, TimeUnit.SECONDS)).isNotNull(); + assertThat(pool.freePages()).isEqualTo(2); + } finally { + executor.shutdownNow(); + } + } + + @Test + void testPoolCloseWakesAllocation() throws Exception { + ExecutorService executor = Executors.newSingleThreadExecutor(); + try (LazyMemorySegmentPool pool = pool(); + MemoryAllocation active = pool.newAllocation()) { + active.nextSegment(); + Future waiter = + executor.submit( + () -> { + try (MemoryAllocation allocation = pool.newAllocation()) { + allocation.nextSegment(); + assertThatThrownBy(allocation::nextSegment) + .isInstanceOf(FlussRuntimeException.class) + .hasMessageContaining("pool closed"); + } + return null; + }); + awaitWaiters(pool, 1); + pool.close(); + waiter.get(10, TimeUnit.SECONDS); + active.close(); + assertThat(pool.queued()).isZero(); + assertThat(pool.freePages()).isEqualTo(2); + } finally { + executor.shutdownNow(); + } + } + + @Test + void testInterruptedAllocationReturnsPages() throws Exception { + ExecutorService executor = Executors.newSingleThreadExecutor(); + CountDownLatch released = new CountDownLatch(1); + try (LazyMemorySegmentPool pool = pool(); + MemoryAllocation active = pool.newAllocation()) { + active.nextSegment(); + Future waiter = + executor.submit( + () -> { + try (MemoryAllocation allocation = pool.newAllocation()) { + allocation.nextSegment(); + allocation.nextSegment(); + } finally { + released.countDown(); + } + return null; + }); + awaitWaiters(pool, 1); + waiter.cancel(true); + assertThat(released.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(pool.queued()).isZero(); + assertThat(pool.freePages()).isEqualTo(1); + } finally { + executor.shutdownNow(); + } + } + + @Test + void testConcurrentAllocationsMakeProgress() throws Exception { + ExecutorService executor = Executors.newFixedThreadPool(4); + try (LazyMemorySegmentPool pool = new LazyMemorySegmentPool(4, 128, Long.MAX_VALUE, 128)) { + List> tasks = new ArrayList<>(); + for (int thread = 0; thread < 4; thread++) { + tasks.add( + executor.submit( + () -> { + int completed = 0; + while (completed < 25) { + try (MemoryAllocation allocation = pool.newAllocation()) { + for (int page = 0; page < 4; page++) { + allocation.nextSegment(); + Thread.yield(); + } + completed++; + } catch (TimeoutException retryable) { + // Each operation fits alone; retry after releasing all + // held pages. + } + } + return null; + })); + } + for (Future task : tasks) { + task.get(10, TimeUnit.SECONDS); + } + assertThat(pool.freePages()).isEqualTo(4); + assertThat(pool.queued()).isZero(); + } finally { + executor.shutdownNow(); + } + } + + private static LazyMemorySegmentPool pool() { + return new LazyMemorySegmentPool(2, 128, Long.MAX_VALUE, 128); + } + + private static void awaitWaiters(LazyMemorySegmentPool pool, int expected) throws Exception { + retry(Duration.ofSeconds(10), () -> assertThat(pool.queued()).isEqualTo(expected)); + } +} diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvWriteProcessor.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvWriteProcessor.java index f67bfaa627b..2215d2a096d 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvWriteProcessor.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvWriteProcessor.java @@ -21,6 +21,7 @@ import org.apache.fluss.compression.ArrowCompressionInfo; import org.apache.fluss.exception.DeletionDisabledException; import org.apache.fluss.exception.SchemaNotExistException; +import org.apache.fluss.memory.MemoryAllocation; import org.apache.fluss.memory.MemorySegmentPool; import org.apache.fluss.metadata.ChangelogImage; import org.apache.fluss.metadata.DeleteBehavior; @@ -160,61 +161,64 @@ public LogAppendInfo putAsLeader( throws Exception { WriteContext writeContext = createWriteContext(kvRecords, targetColumns, mergeMode); RowType latestRowType = writeContext.latestSchema.getRowType(); - WalBuilder walBuilder = createWalBuilder(writeContext.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 { - if (rowTtlTimestampProvider != null) { - rowTtlTimestampProvider.prepareForBatch(clock.milliseconds()); - } - processKvRecords( - kvRecords, - kvRecords.schemaId(), - writeContext.rowMerger, - writeContext.autoIncrementUpdater, - walBuilder, - latestSchemaRow, - logEndOffsetOfPrevBatch, - stateAccessor, - originalPartitionName, - historicalValueLookup); - - // 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 state pre-write - // buffer already written. - if (logAppendInfo.duplicated()) { - stateAccessor.truncateTo(logEndOffsetOfPrevBatch, TruncateReason.DUPLICATED); + try (MemoryAllocation allocation = memorySegmentPool.newAllocation()) { + WalBuilder walBuilder = + createWalBuilder(writeContext.latestSchemaId, latestRowType, allocation); + 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 { + if (rowTtlTimestampProvider != null) { + rowTtlTimestampProvider.prepareForBatch(clock.milliseconds()); + } + processKvRecords( + kvRecords, + kvRecords.schemaId(), + writeContext.rowMerger, + writeContext.autoIncrementUpdater, + walBuilder, + latestSchemaRow, + logEndOffsetOfPrevBatch, + stateAccessor, + originalPartitionName, + historicalValueLookup); + + // 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 state pre-write + // buffer already written. + if (logAppendInfo.duplicated()) { + stateAccessor.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 state pre-write buffer, + // 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 state pre-write buffer. + stateAccessor.truncateTo(logEndOffsetOfPrevBatch, TruncateReason.ERROR); + throw t; + } finally { + // deallocate the memory and arrow writer used by the wal builder + walBuilder.deallocate(); } - 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 state pre-write buffer, - // 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 state pre-write buffer. - stateAccessor.truncateTo(logEndOffsetOfPrevBatch, TruncateReason.ERROR); - throw t; - } finally { - // deallocate the memory and arrow writer used by the wal builder - walBuilder.deallocate(); } } @@ -527,7 +531,8 @@ private static boolean shouldIgnoreDeletion(RowMerger currentMerger) { return false; } - private WalBuilder createWalBuilder(int schemaId, RowType rowType) throws Exception { + private WalBuilder createWalBuilder( + int schemaId, RowType rowType, MemorySegmentPool memorySegmentPool) throws Exception { switch (logFormat) { case INDEXED: if (kvFormat == KvFormat.COMPACTED) { diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/wal/ArrowWalBuilder.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/wal/ArrowWalBuilder.java index b46a5fd1027..c69258479e5 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/wal/ArrowWalBuilder.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/wal/ArrowWalBuilder.java @@ -38,9 +38,14 @@ public class ArrowWalBuilder implements WalBuilder { public ArrowWalBuilder(int schemaId, ArrowWriter writer, MemorySegmentPool memorySegmentPool) throws IOException { this.memorySegmentPool = memorySegmentPool; - this.outputView = new ManagedPagedOutputView(memorySegmentPool); - this.recordsBuilder = - MemoryLogRecordsArrowBuilder.builder(schemaId, writer, outputView, false, null); + try { + this.outputView = new ManagedPagedOutputView(memorySegmentPool); + this.recordsBuilder = + MemoryLogRecordsArrowBuilder.builder(schemaId, writer, outputView, false, null); + } catch (Throwable t) { + writer.close(); + throw t; + } } @Override diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletWalMemoryTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletWalMemoryTest.java new file mode 100644 index 00000000000..6b60b54fc1b --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletWalMemoryTest.java @@ -0,0 +1,470 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.server.kv; + +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.config.MemorySize; +import org.apache.fluss.config.TableConfig; +import org.apache.fluss.exception.RecordTooLargeException; +import org.apache.fluss.exception.TimeoutException; +import org.apache.fluss.memory.LazyMemorySegmentPool; +import org.apache.fluss.memory.MemorySegmentPool; +import org.apache.fluss.memory.TestingMemorySegmentPool; +import org.apache.fluss.metadata.KvFormat; +import org.apache.fluss.metadata.LogFormat; +import org.apache.fluss.metadata.PhysicalTablePath; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.SchemaGetter; +import org.apache.fluss.metadata.SchemaInfo; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.record.BinaryValue; +import org.apache.fluss.record.ChangeType; +import org.apache.fluss.record.FileLogProjection; +import org.apache.fluss.record.KvRecord; +import org.apache.fluss.record.KvRecordBatch; +import org.apache.fluss.record.KvRecordTestUtils; +import org.apache.fluss.record.LogRecord; +import org.apache.fluss.record.LogRecordBatch; +import org.apache.fluss.record.LogRecordReadContext; +import org.apache.fluss.record.LogRecords; +import org.apache.fluss.record.TestData; +import org.apache.fluss.record.TestingSchemaGetter; +import org.apache.fluss.row.encode.ValueEncoder; +import org.apache.fluss.server.kv.autoinc.AutoIncrementManager; +import org.apache.fluss.server.kv.autoinc.TestingSequenceGeneratorFactory; +import org.apache.fluss.server.kv.prewrite.KvPreWriteBuffer.Key; +import org.apache.fluss.server.kv.prewrite.KvPreWriteBuffer.KvEntry; +import org.apache.fluss.server.kv.prewrite.KvPreWriteBuffer.Value; +import org.apache.fluss.server.kv.rowmerger.DefaultRowMerger; +import org.apache.fluss.server.kv.rowmerger.RowMerger; +import org.apache.fluss.server.log.FetchIsolation; +import org.apache.fluss.server.log.LogAppendInfo; +import org.apache.fluss.server.log.LogTablet; +import org.apache.fluss.server.log.LogTestUtils; +import org.apache.fluss.server.metrics.group.TestingMetricGroups; +import org.apache.fluss.shaded.arrow.org.apache.arrow.memory.RootAllocator; +import org.apache.fluss.types.RowType; +import org.apache.fluss.utils.CloseableIterator; +import org.apache.fluss.utils.clock.SystemClock; +import org.apache.fluss.utils.concurrent.FlussScheduler; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import javax.annotation.Nullable; + +import java.io.File; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Random; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.apache.fluss.compression.ArrowCompressionInfo.DEFAULT_COMPRESSION; +import static org.apache.fluss.record.TestData.DATA1_SCHEMA_PK; +import static org.apache.fluss.testutils.DataTestUtils.compactedRow; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for WAL memory allocation, rollback and retry. */ +class KvTabletWalMemoryTest { + private static final short schemaId = 1; + private final Configuration conf = new Configuration(); + private final RowType baseRowType = TestData.DATA1_ROW_TYPE; + private final KvRecordTestUtils.KvRecordBatchFactory kvRecordBatchFactory = + KvRecordTestUtils.KvRecordBatchFactory.of(schemaId); + private final KvRecordTestUtils.KvRecordFactory kvRecordFactory = + KvRecordTestUtils.KvRecordFactory.of(baseRowType); + + private @TempDir File tempLogDir; + private @TempDir File tmpKvDir; + + private TestingSchemaGetter schemaGetter = + new TestingSchemaGetter(new SchemaInfo(DATA1_SCHEMA_PK, schemaId)); + private LogTablet logTablet; + private KvTablet kvTablet; + private ExecutorService executor; + private MemorySegmentPool walMemoryPool = new TestingMemorySegmentPool(10 * 1024); + private LogFormat walLogFormat = LogFormat.ARROW; + + @BeforeEach + void beforeEach() { + executor = Executors.newFixedThreadPool(2); + } + + @AfterEach + void afterEach() { + executor.shutdownNow(); + } + + private void initLogTabletAndKvTablet(Schema schema, Map tableConfig) + throws Exception { + PhysicalTablePath path = PhysicalTablePath.of(TablePath.of("testDb", "t1")); + schemaGetter = new TestingSchemaGetter(new SchemaInfo(schema, schemaId)); + logTablet = createLogTablet(tempLogDir, 0L, path); + kvTablet = + createKvTablet( + path, + logTablet.getTableBucket(), + logTablet, + tmpKvDir, + schemaGetter, + tableConfig, + RowMerger.create( + new TableConfig(Configuration.fromMap(tableConfig)), + KvFormat.COMPACTED, + schemaGetter)); + } + + private LogTablet createLogTablet(File tempLogDir, long tableId, PhysicalTablePath tablePath) + throws Exception { + File logTabletDir = + LogTestUtils.makeRandomLogTabletDir( + tempLogDir, tablePath.getDatabaseName(), tableId, tablePath.getTableName()); + return LogTablet.create( + tempLogDir, + tablePath, + logTabletDir, + conf, + new AtomicBoolean( + conf.get(ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED)), + TestingMetricGroups.TABLET_SERVER_METRICS, + 0, + new FlussScheduler(1), + walLogFormat, + 1, + true, + SystemClock.getInstance(), + true); + } + + private KvTablet createKvTablet( + PhysicalTablePath tablePath, + TableBucket tableBucket, + LogTablet logTablet, + File tmpKvDir, + SchemaGetter schemaGetter, + Map tableConfig, + RowMerger rowMerger) + throws Exception { + TableConfig tableConf = new TableConfig(Configuration.fromMap(tableConfig)); + AutoIncrementManager autoIncrementManager = + new AutoIncrementManager( + schemaGetter, + tablePath.getTablePath(), + new TableConfig(new Configuration()), + new TestingSequenceGeneratorFactory()); + return KvTablet.create( + tablePath, + tableBucket, + logTablet, + tmpKvDir, + conf, + TestingMetricGroups.TABLET_SERVER_METRICS, + new RootAllocator(Long.MAX_VALUE), + walMemoryPool, + KvFormat.COMPACTED, + rowMerger, + DEFAULT_COMPRESSION, + schemaGetter, + tableConf.getChangelogImage(), + KvManager.getDefaultRateLimiter(), + autoIncrementManager, + SystemClock.getInstance(), + tableConf); + } + + @ParameterizedTest + @ValueSource(strings = {"COMPACTED", "ARROW"}) + void testOversizedWalRollsBackAndSubsequentWriteSucceeds(String format) throws Exception { + Configuration poolConfig = new Configuration(); + poolConfig.set(ConfigOptions.SERVER_BUFFER_MEMORY_SIZE, MemorySize.parse("8kb")); + poolConfig.set(ConfigOptions.SERVER_BUFFER_PAGE_SIZE, MemorySize.parse("4kb")); + poolConfig.set( + ConfigOptions.SERVER_BUFFER_PER_REQUEST_MEMORY_SIZE, MemorySize.parse("4kb")); + walLogFormat = LogFormat.valueOf(format); + try (LazyMemorySegmentPool pool = + LazyMemorySegmentPool.createServerBufferPool(poolConfig)) { + walMemoryPool = pool; + initLogTabletAndKvTablet( + DATA1_SCHEMA_PK, Collections.singletonMap("table.changelog.image", "FULL")); + KvTablet tablet = kvTablet; + LogTablet log = logTablet; + try { + byte[] key = "k1".getBytes(); + KvRecordBatch initial = + kvRecordBatchFactory.ofRecords( + Collections.singletonList( + kvRecordFactory.ofRecord(key, new Object[] {1, "initial"})), + 100L, + 0); + tablet.putAsLeader(initial, null); + Value original = tablet.getKvPreWriteBuffer().get(Key.of(key)); + assertThat(original).isNotNull(); + long offset = log.localLogEndOffset(); + List oversized = new ArrayList<>(); + Random random = new Random(42); + for (int i = 0; i < 1000; i++) { + oversized.add( + kvRecordFactory.ofRecord( + key, + new Object[] { + 1, + "updated-" + i + "-" + Long.toHexString(random.nextLong()) + })); + } + assertThatThrownBy( + () -> + tablet.putAsLeader( + kvRecordBatchFactory.ofRecords(oversized, 100L, 1), + null)) + .isInstanceOf(RecordTooLargeException.class); + assertThat(log.localLogEndOffset()).isEqualTo(offset); + assertThat(tablet.getKvPreWriteBuffer().get(Key.of(key))).isEqualTo(original); + assertThat(pool.availableMemory()).isEqualTo(pool.totalSize()); + + KvRecordBatch next = + kvRecordBatchFactory.ofRecords( + Collections.singletonList( + kvRecordFactory.ofRecord(key, new Object[] {1, "next"})), + 100L, + 1); + assertThat(tablet.putAsLeader(next, null).duplicated()).isFalse(); + assertThat(log.localLogEndOffset()).isEqualTo(offset + 2); + LogRecordBatch written = + readLogRecords(log, offset, null).batches().iterator().next(); + assertThat(written.writerId()).isEqualTo(100L); + assertThat(written.batchSequence()).isEqualTo(1); + assertThat(written.isValid()).isTrue(); + try (LogRecordReadContext context = + walLogFormat == LogFormat.COMPACTED + ? LogRecordReadContext.createCompactedRowReadContext( + baseRowType, schemaId, schemaGetter) + : LogRecordReadContext.createArrowReadContext( + baseRowType, schemaId, schemaGetter); + CloseableIterator records = written.records(context)) { + LogRecord before = records.next(); + assertThat(before.getChangeType()).isEqualTo(ChangeType.UPDATE_BEFORE); + assertThat(before.getRow().getInt(0)).isEqualTo(1); + assertThat(before.getRow().getString(1).toString()).isEqualTo("initial"); + LogRecord after = records.next(); + assertThat(after.getChangeType()).isEqualTo(ChangeType.UPDATE_AFTER); + assertThat(after.getRow().getInt(0)).isEqualTo(1); + assertThat(after.getRow().getString(1).toString()).isEqualTo("next"); + assertThat(records.hasNext()).isFalse(); + } + assertThat(pool.availableMemory()).isEqualTo(pool.totalSize()); + } finally { + try { + tablet.close(); + } finally { + log.close(); + } + } + } + } + + @Test + void testConcurrentWalAllocationRollsBackAndRetries() throws Exception { + Configuration poolConfig = new Configuration(); + poolConfig.set(ConfigOptions.SERVER_BUFFER_MEMORY_SIZE, MemorySize.parse("2kb")); + poolConfig.set(ConfigOptions.SERVER_BUFFER_PAGE_SIZE, MemorySize.parse("1kb")); + poolConfig.set( + ConfigOptions.SERVER_BUFFER_PER_REQUEST_MEMORY_SIZE, MemorySize.parse("1kb")); + walLogFormat = LogFormat.COMPACTED; + List tablets = new ArrayList<>(); + List logs = new ArrayList<>(); + try (LazyMemorySegmentPool pool = + LazyMemorySegmentPool.createServerBufferPool(poolConfig)) { + AtomicBoolean coordinate = new AtomicBoolean(); + CountDownLatch firstNeedsMoreMemory = new CountDownLatch(1); + CountDownLatch bothNeedMoreMemory = new CountDownLatch(2); + walMemoryPool = pool; + + byte[] key = "k1".getBytes(); + KvRecordBatch initial = + kvRecordBatchFactory.ofRecords( + Collections.singletonList( + kvRecordFactory.ofRecord(key, new Object[] {1, "initial"})), + 100L, + 0); + List values = + Arrays.asList( + "first-" + String.join("", Collections.nCopies(200, "a")), + "second-" + String.join("", Collections.nCopies(200, "b")), + "third-" + String.join("", Collections.nCopies(200, "c"))); + List updates = new ArrayList<>(); + for (String value : values) { + updates.add(kvRecordFactory.ofRecord(key, new Object[] {1, value})); + } + KvRecordBatch batch = kvRecordBatchFactory.ofRecords(updates, 100L, 1); + try { + for (int i = 0; i < 2; i++) { + PhysicalTablePath path = + PhysicalTablePath.of(TablePath.of("testDb", "competing" + i)); + LogTablet log = createLogTablet(tempLogDir, i, path); + logs.add(log); + final int tabletIndex = i; + RowMerger merger = + new DefaultRowMerger(KvFormat.COMPACTED, null) { + @Override + public BinaryValue merge( + @Nullable BinaryValue oldValue, BinaryValue newValue) { + if (coordinate.get() + && newValue.row + .getString(1) + .toString() + .equals(values.get(2))) { + // Pause after two updates, while the WAL still fits in one + // page. + assertThat( + tablets.get(tabletIndex) + .getKvPreWriteBuffer() + .getMaxLSN()) + .isGreaterThan( + logs.get(tabletIndex).localLogEndOffset()); + bothNeedMoreMemory.countDown(); + firstNeedsMoreMemory.countDown(); + try { + assertThat( + bothNeedMoreMemory.await( + 10, TimeUnit.SECONDS)) + .isTrue(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError( + "Interrupted while coordinating WAL writes", e); + } + } + return super.merge(oldValue, newValue); + } + }; + KvTablet tablet = + createKvTablet( + path, + log.getTableBucket(), + log, + new File(tmpKvDir, "tablet" + i), + new TestingSchemaGetter( + new SchemaInfo(DATA1_SCHEMA_PK, schemaId)), + Collections.singletonMap("table.changelog.image", "FULL"), + merger); + tablets.add(tablet); + tablet.putAsLeader(initial, null); + } + Value original = tablets.get(1).getKvPreWriteBuffer().get(Key.of(key)); + assertThat(original).isNotNull(); + List originalEntries = + new ArrayList<>(tablets.get(1).getKvPreWriteBuffer().getAllKvEntries()); + long offset = logs.get(1).localLogEndOffset(); + coordinate.set(true); + Future older = + executor.submit(() -> tablets.get(0).putAsLeader(batch, null)); + assertThat(firstNeedsMoreMemory.await(10, TimeUnit.SECONDS)).isTrue(); + Future younger = + executor.submit(() -> tablets.get(1).putAsLeader(batch, null)); + assertThatThrownBy(() -> younger.get(10, TimeUnit.SECONDS)) + .isInstanceOf(ExecutionException.class) + .hasCauseInstanceOf(TimeoutException.class) + .hasStackTraceContaining("cannot make progress"); + assertThat(older.get(10, TimeUnit.SECONDS).duplicated()).isFalse(); + assertThat(logs.get(1).localLogEndOffset()).isEqualTo(offset); + assertThat(tablets.get(1).getKvPreWriteBuffer().get(Key.of(key))) + .isEqualTo(original); + assertThat(pool.availableMemory()).isEqualTo(pool.totalSize()); + + // Retry the exact failed batch, retaining its writer id, sequence and payload. + assertThat(tablets.get(1).getKvPreWriteBuffer().getAllKvEntries()) + .containsExactlyElementsOf(originalEntries); + coordinate.set(false); + assertThat(tablets.get(1).putAsLeader(batch, null).duplicated()).isFalse(); + for (int i = 0; i < 2; i++) { + assertThat(logs.get(i).localLogEndOffset()).isEqualTo(offset + 6); + assertThat(tablets.get(i).getKvPreWriteBuffer().get(Key.of(key)).get()) + .isEqualTo( + ValueEncoder.encodeValue( + schemaId, + compactedRow( + baseRowType, new Object[] {1, values.get(2)}))); + LogRecordBatch written = + readLogRecords(logs.get(i), offset, null).batches().iterator().next(); + assertThat(written.writerId()).isEqualTo(100L); + assertThat(written.batchSequence()).isEqualTo(1); + assertThat(written.isValid()).isTrue(); + try (LogRecordReadContext context = + LogRecordReadContext.createCompactedRowReadContext( + baseRowType, schemaId, schemaGetter); + CloseableIterator records = written.records(context)) { + String previous = "initial"; + for (String value : values) { + LogRecord before = records.next(); + assertThat(before.getChangeType()).isEqualTo(ChangeType.UPDATE_BEFORE); + assertThat(before.getRow().getString(1).toString()).isEqualTo(previous); + LogRecord after = records.next(); + assertThat(after.getChangeType()).isEqualTo(ChangeType.UPDATE_AFTER); + assertThat(after.getRow().getString(1).toString()).isEqualTo(value); + previous = value; + } + assertThat(records.hasNext()).isFalse(); + } + } + assertThat(pool.availableMemory()).isEqualTo(pool.totalSize()); + assertThat(pool.queued()).isZero(); + } finally { + // Unblock allocations even when an assertion detects a deadlock regression. + pool.close(); + executor.shutdownNow(); + assertThat(executor.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + for (KvTablet tablet : tablets) { + tablet.close(); + } + for (LogTablet log : logs) { + log.close(); + } + } + } + } + + private LogRecords readLogRecords( + LogTablet logTablet, long startOffset, @Nullable FileLogProjection projection) + throws Exception { + return logTablet + .read( + startOffset, + Integer.MAX_VALUE, + FetchIsolation.LOG_END, + false, + projection, + null) + .getRecords(); + } +}