From f846e291aee075824e3002eca40aa4f7b57ffa88 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Wed, 21 May 2025 19:43:08 +0300 Subject: [PATCH 1/3] Fix ByteBuf leak in ManagedLedgerTest.avoidUseSameOpAddEntryBetweenDifferentLedger --- .../org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java index 9074e8f985c78..ede9439540cd5 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java +++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java @@ -3324,7 +3324,7 @@ public void avoidUseSameOpAddEntryBetweenDifferentLedger() throws Exception { List oldOps = new ArrayList<>(); for (int i = 0; i < 10; i++) { OpAddEntry op = OpAddEntry.createNoRetainBuffer(ledger, - ByteBufAllocator.DEFAULT.buffer(128).retain(), null, null, new AtomicBoolean()); + ByteBufAllocator.DEFAULT.buffer(128), null, null, new AtomicBoolean()); if (i > 4) { op.setLedger(mock(LedgerHandle.class)); } @@ -3341,6 +3341,7 @@ public void avoidUseSameOpAddEntryBetweenDifferentLedger() throws Exception { } else { Assert.assertSame(oldOp, newOp); } + oldOp.getData().release(); } } From 1133d3d5557d432703339ef65a500214700ff89c Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Wed, 21 May 2025 19:46:08 +0300 Subject: [PATCH 2/3] Fix ByteBuf leak in TripleLongPriorityQueueTest.testShrink --- .../common/util/collections/TripleLongPriorityQueueTest.java | 1 + 1 file changed, 1 insertion(+) diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueTest.java index 01748e2b4ed97..b64c071fe039a 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueTest.java @@ -190,6 +190,7 @@ public void testShrink() throws Exception { // Trigger shrinking pq.clear(); capacity = scaleCapacity - (int)(scaleCapacity * 0.5f * 0.9f); + pq.close(); } private void triggerScaleOut(int initialCapacity, TripleLongPriorityQueue pq) { From 2efe9869c5a6b35b9349334426a2950a5c9e4a95 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Wed, 21 May 2025 21:04:51 +0300 Subject: [PATCH 3/3] Fix leaks in BlobStoreManagedLedgerOffloaderTest / BlobStoreManagedLedgerOffloaderStreamingTest --- ...reManagedLedgerOffloaderStreamingTest.java | 30 ++++++++++ .../BlobStoreManagedLedgerOffloaderTest.java | 57 +++++++++++++++++-- 2 files changed, 83 insertions(+), 4 deletions(-) diff --git a/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreManagedLedgerOffloaderStreamingTest.java b/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreManagedLedgerOffloaderStreamingTest.java index 67dcb73b9e58a..e03eaa1a78719 100644 --- a/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreManagedLedgerOffloaderStreamingTest.java +++ b/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreManagedLedgerOffloaderStreamingTest.java @@ -30,6 +30,7 @@ import java.util.Map; import java.util.Random; import java.util.UUID; +import lombok.Cleanup; import org.apache.bookkeeper.client.BKException; import org.apache.bookkeeper.client.api.LedgerEntries; import org.apache.bookkeeper.client.api.LedgerEntry; @@ -97,6 +98,7 @@ private BlobStoreManagedLedgerOffloader getOffloader(String bucket, BlobStore mo @Test public void testHappyCase() throws Exception { + @Cleanup LedgerOffloader offloader = getOffloader(new HashMap() {{ put(TieredStorageConfiguration.MAX_OFFLOAD_SEGMENT_SIZE_IN_BYTES, "1000"); put(config.getKeys(TieredStorageConfiguration.METADATA_FIELD_MAX_BLOCK_SIZE).get(0), "5242880"); @@ -107,6 +109,7 @@ public void testHappyCase() throws Exception { long beginLedger = 0; long beginEntry = 0; log.error("try begin offload"); + @Cleanup OffloadHandle offloadHandle = offloader .streamingOffload(ml, uuid, beginLedger, beginEntry, new HashMap<>()).get(); //Segment should closed because size in bytes full @@ -123,6 +126,7 @@ public void testHappyCase() throws Exception { @Test public void testReadAndWrite() throws Exception { + @Cleanup LedgerOffloader offloader = getOffloader(new HashMap() {{ put(TieredStorageConfiguration.MAX_OFFLOAD_SEGMENT_SIZE_IN_BYTES, "1000"); put(config.getKeys(TieredStorageConfiguration.METADATA_FIELD_MAX_BLOCK_SIZE).get(0), "5242880"); @@ -136,6 +140,7 @@ public void testReadAndWrite() throws Exception { Map driverMeta = new HashMap() {{ put(TieredStorageConfiguration.METADATA_FIELD_BUCKET, BUCKET); }}; + @Cleanup OffloadHandle offloadHandle = offloader .streamingOffload(ml, uuid, beginLedger, beginEntry, driverMeta).get(); @@ -159,7 +164,9 @@ public void testReadAndWrite() throws Exception { .setUidMsb(uuid.getMostSignificantBits()) .setComplete(true).setEndEntryId(9).build()); + @Cleanup final ReadHandle readHandle = offloader.readOffloaded(0, contextBuilder.build(), driverMeta).get(); + @Cleanup final LedgerEntries ledgerEntries = readHandle.readAsync(0, 9).get(); for (LedgerEntry ledgerEntry : ledgerEntries) { @@ -172,6 +179,7 @@ public void testReadAndWrite() throws Exception { @Test public void testReadAndWriteAcrossLedger() throws Exception { + @Cleanup LedgerOffloader offloader = getOffloader(new HashMap() {{ put(TieredStorageConfiguration.MAX_OFFLOAD_SEGMENT_SIZE_IN_BYTES, "2000"); put(config.getKeys(TieredStorageConfiguration.METADATA_FIELD_MAX_BLOCK_SIZE).get(0), "5242880"); @@ -185,6 +193,7 @@ public void testReadAndWriteAcrossLedger() throws Exception { Map driverMeta = new HashMap() {{ put(TieredStorageConfiguration.METADATA_FIELD_BUCKET, BUCKET); }}; + @Cleanup OffloadHandle offloadHandle = offloader .streamingOffload(ml, uuid, beginLedger, beginEntry, driverMeta).get(); @@ -216,7 +225,9 @@ public void testReadAndWriteAcrossLedger() throws Exception { .setUidMsb(uuid.getMostSignificantBits()) .setComplete(true).setEndEntryId(9).build()); + @Cleanup final ReadHandle readHandle = offloader.readOffloaded(0, contextBuilder.build(), driverMeta).get(); + @Cleanup final LedgerEntries ledgerEntries = readHandle.readAsync(0, 9).get(); for (LedgerEntry ledgerEntry : ledgerEntries) { @@ -226,7 +237,9 @@ public void testReadAndWriteAcrossLedger() throws Exception { assertEquals(storedData, entryBytes); } + @Cleanup final ReadHandle readHandle2 = offloader.readOffloaded(1, contextBuilder.build(), driverMeta).get(); + @Cleanup final LedgerEntries ledgerEntries2 = readHandle2.readAsync(0, 9).get(); for (LedgerEntry ledgerEntry : ledgerEntries2) { @@ -239,11 +252,13 @@ public void testReadAndWriteAcrossLedger() throws Exception { @Test public void testReadAndWriteAcrossSegment() throws Exception { + @Cleanup LedgerOffloader offloader = getOffloader(new HashMap() {{ put(TieredStorageConfiguration.MAX_OFFLOAD_SEGMENT_SIZE_IN_BYTES, "1000"); put(config.getKeys(TieredStorageConfiguration.METADATA_FIELD_MAX_BLOCK_SIZE).get(0), "5242880"); put(TieredStorageConfiguration.MAX_OFFLOAD_SEGMENT_ROLLOVER_TIME_SEC, "600"); }}); + @Cleanup LedgerOffloader offloader2 = getOffloader(new HashMap() {{ put(TieredStorageConfiguration.MAX_OFFLOAD_SEGMENT_SIZE_IN_BYTES, "1000"); put(config.getKeys(TieredStorageConfiguration.METADATA_FIELD_MAX_BLOCK_SIZE).get(0), "5242880"); @@ -258,6 +273,7 @@ public void testReadAndWriteAcrossSegment() throws Exception { Map driverMeta = new HashMap() {{ put(TieredStorageConfiguration.METADATA_FIELD_BUCKET, BUCKET); }}; + @Cleanup OffloadHandle offloadHandle = offloader .streamingOffload(ml, uuid, beginLedger, beginEntry, driverMeta).get(); @@ -276,6 +292,7 @@ public void testReadAndWriteAcrossSegment() throws Exception { assertEquals(offloadResult.endEntry, 9); //Segment should closed because size in bytes full + @Cleanup OffloadHandle offloadHandle2 = offloader2 .streamingOffload(ml, uuid2, beginLedger, 10, driverMeta).get(); for (int i = 0; i < 10; i++) { @@ -301,7 +318,9 @@ public void testReadAndWriteAcrossSegment() throws Exception { .setComplete(true).setEndEntryId(19).build() ); + @Cleanup final ReadHandle readHandle = offloader.readOffloaded(0, contextBuilder.build(), driverMeta).get(); + @Cleanup final LedgerEntries ledgerEntries = readHandle.readAsync(0, 19).get(); for (LedgerEntry ledgerEntry : ledgerEntries) { @@ -314,6 +333,7 @@ public void testReadAndWriteAcrossSegment() throws Exception { @Test public void testRandomRead() throws Exception { + @Cleanup LedgerOffloader offloader = getOffloader(new HashMap() {{ put(TieredStorageConfiguration.MAX_OFFLOAD_SEGMENT_SIZE_IN_BYTES, "1000"); put(config.getKeys(TieredStorageConfiguration.METADATA_FIELD_MAX_BLOCK_SIZE).get(0), "5242880"); @@ -333,6 +353,7 @@ public void testRandomRead() throws Exception { Map driverMeta = new HashMap() {{ put(TieredStorageConfiguration.METADATA_FIELD_BUCKET, BUCKET); }}; + @Cleanup OffloadHandle offloadHandle = offloader .streamingOffload(ml, uuid, beginLedger, beginEntry, driverMeta).get(); @@ -351,6 +372,7 @@ public void testRandomRead() throws Exception { assertEquals(offloadResult.endEntry, 9); //Segment should closed because size in bytes full + @Cleanup OffloadHandle offloadHandle2 = offloader2 .streamingOffload(ml, uuid2, beginLedger, 10, driverMeta).get(); for (int i = 0; i < 10; i++) { @@ -376,6 +398,7 @@ public void testRandomRead() throws Exception { .setComplete(true).setEndEntryId(19).build() ); + @Cleanup final ReadHandle readHandle = offloader.readOffloaded(0, contextBuilder.build(), driverMeta).get(); for (int i = 0; i <= 19; i++) { @@ -387,6 +410,7 @@ public void testRandomRead() throws Exception { begin = end; end = temp; } + @Cleanup final LedgerEntries ledgerEntries = readHandle.readAsync(begin, end).get(); for (LedgerEntry ledgerEntry : ledgerEntries) { final EntryImpl storedEntry = (EntryImpl) entries.get((int) ledgerEntry.getEntryId()); @@ -399,6 +423,7 @@ public void testRandomRead() throws Exception { @Test public void testInvalidEntryIds() throws Exception { + @Cleanup LedgerOffloader offloader = getOffloader(new HashMap() {{ put(TieredStorageConfiguration.MAX_OFFLOAD_SEGMENT_SIZE_IN_BYTES, "1000"); put(config.getKeys(TieredStorageConfiguration.METADATA_FIELD_MAX_BLOCK_SIZE).get(0), "5242880"); @@ -412,6 +437,7 @@ public void testInvalidEntryIds() throws Exception { Map driverMeta = new HashMap() {{ put(TieredStorageConfiguration.METADATA_FIELD_BUCKET, BUCKET); }}; + @Cleanup OffloadHandle offloadHandle = offloader .streamingOffload(ml, uuid, beginLedger, beginEntry, driverMeta).get(); @@ -435,6 +461,7 @@ public void testInvalidEntryIds() throws Exception { .setUidMsb(uuid.getMostSignificantBits()) .setComplete(true).setEndEntryId(9).build()); + @Cleanup final ReadHandle readHandle = offloader.readOffloaded(0, contextBuilder.build(), driverMeta).get(); try { readHandle.read(-1, -1); @@ -451,6 +478,7 @@ public void testInvalidEntryIds() throws Exception { @Test public void testReadNotExistLedger() throws Exception { + @Cleanup LedgerOffloader offloader = getOffloader(new HashMap() {{ put(TieredStorageConfiguration.MAX_OFFLOAD_SEGMENT_SIZE_IN_BYTES, "1000"); put(config.getKeys(TieredStorageConfiguration.METADATA_FIELD_MAX_BLOCK_SIZE).get(0), "5242880"); @@ -464,6 +492,7 @@ public void testReadNotExistLedger() throws Exception { Map driverMeta = new HashMap() {{ put(TieredStorageConfiguration.METADATA_FIELD_BUCKET, BUCKET); }}; + @Cleanup OffloadHandle offloadHandle = offloader .streamingOffload(ml, uuid, beginLedger, beginEntry, driverMeta).get(); @@ -487,6 +516,7 @@ public void testReadNotExistLedger() throws Exception { .setUidMsb(uuid.getMostSignificantBits()) .setComplete(true).setEndEntryId(9).build()); + @Cleanup final ReadHandle readHandle = offloader.readOffloaded(0, contextBuilder.build(), driverMeta).get(); // delete blob(ledger) 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 c8bcb67aa2cdd..6c56341e9c304 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 @@ -41,6 +41,7 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import lombok.Cleanup; import org.apache.bookkeeper.client.BKException; import org.apache.bookkeeper.client.api.LedgerEntries; import org.apache.bookkeeper.client.api.LedgerEntry; @@ -113,6 +114,7 @@ private BlobStoreManagedLedgerOffloader getOffloader(String bucket, BlobStore mo @Test(timeOut = 600000) // 10 minutes. public void testHappyCase() throws Exception { + @Cleanup LedgerOffloader offloader = getOffloader(); offloader.offload(buildReadHandle(), UUID.randomUUID(), new HashMap<>()).get(); } @@ -125,6 +127,7 @@ public void testBucketDoesNotExist() throws Exception { return; } + @Cleanup LedgerOffloader offloader = getOffloader("some-non-existant-bucket-name"); try { offloader.offload(buildReadHandle(), UUID.randomUUID(), new HashMap<>()).get(); @@ -137,12 +140,15 @@ public void testBucketDoesNotExist() throws Exception { @Test(timeOut = 600000) // 10 minutes. public void testOffloadAndRead() throws Exception { + @Cleanup ReadHandle toWrite = buildReadHandle(DEFAULT_BLOCK_SIZE, 3); + @Cleanup LedgerOffloader offloader = getOffloader(); UUID uuid = UUID.randomUUID(); offloader.offload(toWrite, uuid, new HashMap<>()).get(); + @Cleanup ReadHandle toTest = offloader.readOffloaded(toWrite.getId(), uuid, Collections.emptyMap()).get(); assertEquals(toTest.getLastAddConfirmed(), toWrite.getLastAddConfirmed()); @@ -167,7 +173,9 @@ public void testOffloadAndRead() throws Exception { @Test(timeOut = 60000) public void testReadHandlerState() throws Exception { + @Cleanup ReadHandle toWrite = buildReadHandle(DEFAULT_BLOCK_SIZE, 3); + @Cleanup LedgerOffloader offloader = getOffloader(); UUID uuid = UUID.randomUUID(); @@ -176,14 +184,17 @@ public void testReadHandlerState() throws Exception { BlobStoreBackedReadHandleImpl toTest = (BlobStoreBackedReadHandleImpl) offloader.readOffloaded(toWrite.getId(), uuid, Collections.emptyMap()).get(); Assert.assertEquals(toTest.getLastAddConfirmed(), toWrite.getLastAddConfirmed()); Assert.assertEquals(toTest.getState(), BlobStoreBackedReadHandleImpl.State.Opened); - toTest.read(0, 1); + @Cleanup + LedgerEntries ledgerEntries = toTest.read(0, 1); toTest.close(); Assert.assertEquals(toTest.getState(), BlobStoreBackedReadHandleImpl.State.Closed); } @Test(timeOut = 600000) // 10 minutes. public void testOffloadAndReadMetrics() throws Exception { + @Cleanup ReadHandle toWrite = buildReadHandle(DEFAULT_BLOCK_SIZE, 3); + @Cleanup LedgerOffloader offloader = getOffloader(); UUID uuid = UUID.randomUUID(); @@ -193,6 +204,7 @@ public void testOffloadAndReadMetrics() throws Exception { extraMap.put("ManagedLedgerName", managedLegerName); offloader.offload(toWrite, uuid, extraMap).get(); + @Cleanup LedgerOffloaderStatsImpl offloaderStats = (LedgerOffloaderStatsImpl) this.offloaderStats; assertEquals(offloaderStats.getOffloadError(topic), 0); @@ -203,7 +215,9 @@ public void testOffloadAndReadMetrics() throws Exception { Map map = new HashMap<>(); map.putAll(offloader.getOffloadDriverMetadata()); map.put("ManagedLedgerName", managedLegerName); + @Cleanup ReadHandle toTest = offloader.readOffloaded(toWrite.getId(), uuid, map).get(); + @Cleanup LedgerEntries toTestEntries = toTest.read(0, toTest.getLastAddConfirmed()); Iterator toTestIter = toTestEntries.iterator(); while (toTestIter.hasNext()) { @@ -218,6 +232,7 @@ public void testOffloadAndReadMetrics() throws Exception { @Test public void testOffloadFailInitDataBlockUpload() throws Exception { + @Cleanup ReadHandle readHandle = buildReadHandle(); UUID uuid = UUID.randomUUID(); String failureString = "fail InitDataBlockUpload"; @@ -230,6 +245,7 @@ public void testOffloadFailInitDataBlockUpload() throws Exception { .doThrow(new RuntimeException(failureString)) .when(spiedBlobStore).initiateMultipartUpload(any(), any(), any()); + @Cleanup BlobStoreManagedLedgerOffloader offloader = getOffloader(spiedBlobStore); offloader.offload(readHandle, uuid, new HashMap<>()).get(); Assert.fail("Should throw exception when initiateMultipartUpload"); @@ -243,6 +259,7 @@ public void testOffloadFailInitDataBlockUpload() throws Exception { @Test public void testOffloadFailDataBlockPartUpload() throws Exception { + @Cleanup ReadHandle readHandle = buildReadHandle(); UUID uuid = UUID.randomUUID(); String failureString = "fail DataBlockPartUpload"; @@ -255,6 +272,7 @@ public void testOffloadFailDataBlockPartUpload() throws Exception { .doThrow(new RuntimeException(failureString)) .when(spiedBlobStore).uploadMultipartPart(any(), anyInt(), any()); + @Cleanup BlobStoreManagedLedgerOffloader offloader = getOffloader(spiedBlobStore); offloader.offload(readHandle, uuid, new HashMap<>()).get(); Assert.fail("Should throw exception for when uploadPart"); @@ -268,6 +286,7 @@ public void testOffloadFailDataBlockPartUpload() throws Exception { @Test public void testOffloadFailDataBlockUploadComplete() throws Exception { + @Cleanup ReadHandle readHandle = buildReadHandle(); UUID uuid = UUID.randomUUID(); String failureString = "fail DataBlockUploadComplete"; @@ -282,6 +301,7 @@ public void testOffloadFailDataBlockUploadComplete() throws Exception { .doNothing() .when(spiedBlobStore).abortMultipartUpload(any()); + @Cleanup BlobStoreManagedLedgerOffloader offloader = getOffloader(spiedBlobStore); offloader.offload(readHandle, uuid, new HashMap<>()).get(); @@ -297,6 +317,7 @@ public void testOffloadFailDataBlockUploadComplete() throws Exception { @Test public void testOffloadFailPutIndexBlock() throws Exception { + @Cleanup ReadHandle readHandle = buildReadHandle(); UUID uuid = UUID.randomUUID(); String failureString = "fail putObject"; @@ -308,6 +329,7 @@ public void testOffloadFailPutIndexBlock() throws Exception { .doThrow(new RuntimeException(failureString)) .when(spiedBlobStore).putBlob(any(), any()); + @Cleanup BlobStoreManagedLedgerOffloader offloader = getOffloader(spiedBlobStore); offloader.offload(readHandle, uuid, new HashMap<>()).get(); @@ -323,6 +345,7 @@ public void testOffloadFailPutIndexBlock() throws Exception { @Test(timeOut = 600000) public void testOffloadReadRandomAccess() throws Exception { + @Cleanup ReadHandle toWrite = buildReadHandle(DEFAULT_BLOCK_SIZE, 3); long[][] randomAccesses = new long[10][2]; Random r = new Random(0); @@ -338,11 +361,13 @@ public void testOffloadReadRandomAccess() throws Exception { randomAccesses[i][1] = second; } + @Cleanup LedgerOffloader offloader = getOffloader(); UUID uuid = UUID.randomUUID(); offloader.offload(toWrite, uuid, new HashMap<>()).get(); + @Cleanup ReadHandle toTest = offloader.readOffloaded(toWrite.getId(), uuid, Collections.emptyMap()).get(); assertEquals(toTest.getLastAddConfirmed(), toWrite.getLastAddConfirmed()); @@ -369,11 +394,14 @@ public void testOffloadReadRandomAccess() throws Exception { @Test public void testOffloadReadInvalidEntryIds() throws Exception { + @Cleanup ReadHandle toWrite = buildReadHandle(DEFAULT_BLOCK_SIZE, 1); + @Cleanup LedgerOffloader offloader = getOffloader(); UUID uuid = UUID.randomUUID(); offloader.offload(toWrite, uuid, new HashMap<>()).get(); + @Cleanup ReadHandle toTest = offloader.readOffloaded(toWrite.getId(), uuid, Collections.emptyMap()).get(); assertEquals(toTest.getLastAddConfirmed(), toWrite.getLastAddConfirmed()); @@ -392,9 +420,11 @@ public void testOffloadReadInvalidEntryIds() throws Exception { @Test public void testDeleteOffloaded() throws Exception { + @Cleanup ReadHandle readHandle = buildReadHandle(DEFAULT_BLOCK_SIZE, 1); UUID uuid = UUID.randomUUID(); + @Cleanup BlobStoreManagedLedgerOffloader offloader = getOffloader(); // verify object exist after offload @@ -411,6 +441,7 @@ public void testDeleteOffloaded() throws Exception { @Test public void testDeleteOffloadedFail() throws Exception { String failureString = "fail deleteOffloaded"; + @Cleanup ReadHandle readHandle = buildReadHandle(DEFAULT_BLOCK_SIZE, 1); UUID uuid = UUID.randomUUID(); @@ -420,6 +451,7 @@ public void testDeleteOffloadedFail() throws Exception { .doThrow(new RuntimeException(failureString)) .when(spiedBlobStore).removeBlobs(any(), any()); + @Cleanup BlobStoreManagedLedgerOffloader offloader = getOffloader(spiedBlobStore); try { @@ -451,6 +483,7 @@ public void testOffloadEmpty() throws Exception { Mockito.doReturn(1234L).when(readHandle).getId(); UUID uuid = UUID.randomUUID(); + @Cleanup LedgerOffloader offloader = getOffloader(); try { @@ -463,6 +496,7 @@ public void testOffloadEmpty() throws Exception { @Test public void testReadUnknownDataVersion() throws Exception { + @Cleanup ReadHandle toWrite = buildReadHandle(DEFAULT_BLOCK_SIZE, 1); BlobStoreManagedLedgerOffloader offloader = getOffloader(); @@ -502,6 +536,7 @@ public void testReadUnknownDataVersion() throws Exception { @Test public void testReadUnknownIndexVersion() throws Exception { + @Cleanup ReadHandle toWrite = buildReadHandle(DEFAULT_BLOCK_SIZE, 1); BlobStoreManagedLedgerOffloader offloader = getOffloader(); @@ -539,17 +574,22 @@ public void testReadUnknownIndexVersion() throws Exception { @Test public void testReadEOFException() throws Throwable { + @Cleanup ReadHandle toWrite = buildReadHandle(DEFAULT_BLOCK_SIZE, 1); + @Cleanup LedgerOffloader offloader = getOffloader(); UUID uuid = UUID.randomUUID(); offloader.offload(toWrite, uuid, new HashMap<>()).get(); + @Cleanup ReadHandle toTest = offloader.readOffloaded(toWrite.getId(), uuid, Collections.emptyMap()).get(); Assert.assertEquals(toTest.getLastAddConfirmed(), toWrite.getLastAddConfirmed()); - toTest.readAsync(0, toTest.getLastAddConfirmed()).get(); + @Cleanup + LedgerEntries ledgerEntries = toTest.readAsync(0, toTest.getLastAddConfirmed()).get(); try { - toTest.readAsync(0, 0).get(); + @Cleanup + LedgerEntries ledgerEntries2 = toTest.readAsync(0, 0).get(); } catch (Exception e) { Assert.fail("Get unexpected exception when reading entries", e); } @@ -557,7 +597,9 @@ public void testReadEOFException() throws Throwable { @Test(timeOut = 600000) // 10 minutes. public void testScanLedgers() throws Exception { + @Cleanup ReadHandle toWrite = buildReadHandle(DEFAULT_BLOCK_SIZE, 3); + @Cleanup LedgerOffloader offloader = getOffloader(); UUID uuid = UUID.randomUUID(); @@ -585,15 +627,19 @@ public void testScanLedgers() throws Exception { @Test public void testReadWithAClosedLedgerHandler() throws Exception { + @Cleanup ReadHandle toWrite = buildReadHandle(DEFAULT_BLOCK_SIZE, 1); + @Cleanup LedgerOffloader offloader = getOffloader(); UUID uuid = UUID.randomUUID(); offloader.offload(toWrite, uuid, new HashMap<>()).get(); + @Cleanup 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(); + @Cleanup + LedgerEntries ledgerEntries = toTest.readAsync(0, lac).get(); toTest.closeAsync().get(); try { toTest.readAsync(0, lac).get(); @@ -608,11 +654,14 @@ public void testReadWithAClosedLedgerHandler() throws Exception { @Test public void testReadNotExistLedger() throws Exception { + @Cleanup ReadHandle toWrite = buildReadHandle(DEFAULT_BLOCK_SIZE, 3); + @Cleanup LedgerOffloader offloader = getOffloader(); UUID uuid = UUID.randomUUID(); offloader.offload(toWrite, uuid, new HashMap<>()).get(); + @Cleanup ReadHandle offloadRead = offloader.readOffloaded(toWrite.getId(), uuid, Collections.emptyMap()).get(); assertEquals(offloadRead.getLastAddConfirmed(), toWrite.getLastAddConfirmed());