From 75ecb69df1c956edd6dbc42f26221f21e7ab85ca Mon Sep 17 00:00:00 2001 From: void-ptr974 Date: Fri, 17 Jul 2026 07:53:28 +0800 Subject: [PATCH] Fix explicitLac race restoring deleted ledgers --- .../storage/ldb/LedgerMetadataIndex.java | 25 +++-- .../storage/ldb/LedgerMetadataIndexTest.java | 103 ++++++++++++++++++ 2 files changed, 116 insertions(+), 12 deletions(-) create mode 100644 bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/LedgerMetadataIndexTest.java diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LedgerMetadataIndex.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LedgerMetadataIndex.java index 9037d4d0753..7ded806d520 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LedgerMetadataIndex.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LedgerMetadataIndex.java @@ -385,21 +385,22 @@ boolean setStorageStateFlags(int expected, int newFlags) throws IOException { } void setExplicitLac(long ledgerId, ByteBuf lac) throws IOException { - LedgerData ledgerData = ledgers.get(ledgerId); - if (ledgerData != null) { - LedgerData newLedgerData = new LedgerData().copyFrom(ledgerData) - .setExplicitLac(ByteBufUtil.getBytes(lac)); + ReentrantLock lock = lockForLedger(ledgerId); + lock.lock(); + try { + LedgerData ledgerData = ledgers.get(ledgerId); + if (ledgerData != null) { + LedgerData newLedgerData = new LedgerData().copyFrom(ledgerData) + .setExplicitLac(ByteBufUtil.getBytes(lac)); - if (ledgers.put(ledgerId, newLedgerData) == null) { - // Ledger had been deleted - ledgersCount.incrementAndGet(); - return; - } else { + ledgers.put(ledgerId, newLedgerData); log.debug().attr("ledgerId", ledgerId).log("Set explicitLac on ledger"); + pendingLedgersUpdates.add(new SimpleEntry(ledgerId, newLedgerData)); + } else { + // unknown ledger here } - pendingLedgersUpdates.add(new SimpleEntry(ledgerId, newLedgerData)); - } else { - // unknown ledger here + } finally { + lock.unlock(); } } diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/LedgerMetadataIndexTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/LedgerMetadataIndexTest.java new file mode 100644 index 00000000000..8312b389ca4 --- /dev/null +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/LedgerMetadataIndexTest.java @@ -0,0 +1,103 @@ +/* + * + * 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.bookkeeper.bookie.storage.ldb; + +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.spy; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import java.io.File; +import java.nio.file.Files; +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 org.apache.bookkeeper.bookie.Bookie; +import org.apache.bookkeeper.conf.ServerConfiguration; +import org.apache.bookkeeper.stats.NullStatsLogger; +import org.junit.Test; + +/** + * Unit test for {@link LedgerMetadataIndex}. + */ +public class LedgerMetadataIndexTest { + + @Test + public void testSetExplicitLacDoesNotResurrectDeletedLedger() throws Exception { + File tmpDir = Files.createTempDirectory("bk-ledger-metadata-index").toFile(); + ExecutorService executor = Executors.newFixedThreadPool(2); + ByteBuf explicitLac = spy(Unpooled.directBuffer(Long.BYTES * 2)); + + try (LedgerMetadataIndex index = new LedgerMetadataIndex(new ServerConfiguration(), + KeyValueStorageRocksDB.factory, tmpDir.getAbsolutePath(), NullStatsLogger.INSTANCE)) { + long ledgerId = 1234L; + index.setMasterKey(ledgerId, new byte[] {'m', 'k'}); + explicitLac.writeLong(ledgerId); + explicitLac.writeLong(10L); + + CountDownLatch lacCopyStarted = new CountDownLatch(1); + CountDownLatch allowLacCopy = new CountDownLatch(1); + CountDownLatch deleteStarted = new CountDownLatch(1); + CountDownLatch deleteFinished = new CountDownLatch(1); + + doAnswer(invocation -> { + lacCopyStarted.countDown(); + assertTrue(allowLacCopy.await(10, TimeUnit.SECONDS)); + return invocation.callRealMethod(); + }).when(explicitLac).getBytes(anyInt(), any(byte[].class)); + + Future setExplicitLac = executor.submit(() -> { + index.setExplicitLac(ledgerId, explicitLac); + return null; + }); + + assertTrue(lacCopyStarted.await(10, TimeUnit.SECONDS)); + + Future delete = executor.submit(() -> { + deleteStarted.countDown(); + try { + index.delete(ledgerId); + } finally { + deleteFinished.countDown(); + } + return null; + }); + + assertTrue(deleteStarted.await(10, TimeUnit.SECONDS)); + deleteFinished.await(1, TimeUnit.SECONDS); + allowLacCopy.countDown(); + + setExplicitLac.get(10, TimeUnit.SECONDS); + delete.get(10, TimeUnit.SECONDS); + + assertThrows(Bookie.NoLedgerException.class, () -> index.get(ledgerId)); + } finally { + explicitLac.release(); + executor.shutdownNow(); + } + } +}