[fix][tiered-storage] move the state check forward (#17020)
* [fix][tiered-storage] move the state check forward
---
*Motivation*
Move the close check forward to avoid `getLastAddConfirmed()` get
an NPE.
If the state is closed. That means the resource is closed and the
`OffloadIndexBlock` has been recycled. Which will cause an NPE when
`getLastAddCOnfirmed()`.
diff --git a/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java b/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java
index 38ba38a..ab64388 100644
--- a/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java
+++ b/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java
@@ -120,6 +120,11 @@
List<LedgerEntry> entries = new ArrayList<LedgerEntry>();
boolean seeked = false;
try {
+ if (state == State.Closed) {
+ log.warn("Reading a closed read handler. Ledger ID: {}, Read range: {}-{}",
+ ledgerId, firstEntry, lastEntry);
+ throw new BKException.BKUnexpectedConditionException();
+ }
if (firstEntry > lastEntry
|| firstEntry < 0
|| lastEntry > getLastAddConfirmed()) {
@@ -138,11 +143,6 @@
}
while (entriesToRead > 0) {
- if (state == State.Closed) {
- log.warn("Reading a closed read handler. Ledger ID: {}, Read range: {}-{}",
- ledgerId, firstEntry, lastEntry);
- throw new BKException.BKUnexpectedConditionException();
- }
long currentPosition = inputStream.getCurrentPosition();
int length = dataStream.readInt();
if (length < 0) { // hit padding or new block
@@ -189,6 +189,8 @@
promise.complete(LedgerEntriesImpl.create(entries));
} catch (Throwable t) {
+ log.error("Failed to read entries {} - {} from the offloader in ledger {}",
+ firstEntry, lastEntry, ledgerId, t);
promise.completeExceptionally(t);
entries.forEach(LedgerEntry::close);
}
diff --git a/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreManagedLedgerOffloaderTest.java b/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreManagedLedgerOffloaderTest.java
index 74d03b1..e6b0cc1 100644
--- a/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreManagedLedgerOffloaderTest.java
+++ b/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreManagedLedgerOffloaderTest.java
@@ -566,4 +566,27 @@
OffloadedLedgerMetadata offloadedLedgerMetadata2 = result.get(1);
assertEquals(toWrite.getId(), offloadedLedgerMetadata2.getLedgerId());
}
-}
\ No newline at end of file
+
+ @Test
+ public void testReadWithAClosedLedgerHandler() throws Exception {
+ ReadHandle toWrite = buildReadHandle(DEFAULT_BLOCK_SIZE, 1);
+ LedgerOffloader offloader = getOffloader();
+ UUID uuid = UUID.randomUUID();
+ offloader.offload(toWrite, uuid, new HashMap<>()).get();
+
+ ReadHandle toTest = offloader.readOffloaded(toWrite.getId(), uuid, Collections.emptyMap()).get();
+ Assert.assertEquals(toTest.getLastAddConfirmed(), toWrite.getLastAddConfirmed());
+ long lac = toTest.getLastAddConfirmed();
+ toTest.readAsync(0, lac).get();
+ toTest.closeAsync().get();
+ try {
+ toTest.readAsync(0, lac).get();
+ } catch (Exception e) {
+ if (e.getCause() instanceof BKException.BKUnexpectedConditionException) {
+ // expected exception
+ return;
+ }
+ throw e;
+ }
+ }
+}
