[improve][ml] Offload ledgers without check ledger length (apache#24344)

Co-authored-by: Lari Hotari <lhotari@users.noreply.github.com>
(cherry picked from commit 3ccea6a)
(cherry picked from commit 735d9d5)

Author: 2 people

tiered-storage/file-system/src/main/java/org/apache/bookkeeper/mledger/offload/filesystem/impl/FileSystemManagedLedgerOffloader.java

@@ -192,11 +192,16 @@ private LedgerReader(ReadHandle readHandle,
 
         @Override
         public void run() {
-            if (readHandle.getLength() == 0 || !readHandle.isClosed() || readHandle.getLastAddConfirmed() < 0) {
+            if (!readHandle.isClosed() || readHandle.getLastAddConfirmed() < 0) {
                 promise.completeExceptionally(
                         new IllegalArgumentException("An empty or open ledger should never be offloaded"));
                 return;
             }
+            if (readHandle.getLength() <= 0) {
+                log.warn("Ledger [{}] has zero length, but it contains {} entries. "
+                    + " Attempting to offload ledger since it contains entries.", readHandle.getId(),
+                    readHandle.getLastAddConfirmed() + 1);
+            }
             long ledgerId = readHandle.getId();
             final String managedLedgerName = extraMetadata.get(MANAGED_LEDGER_NAME);
             String storagePath = getStoragePath(storageBasePath, managedLedgerName);

tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreManagedLedgerOffloader.java

@@ -195,11 +195,16 @@ public CompletableFuture<Void> offload(ReadHandle readHandle,
             final BlobStore writeBlobStore = getBlobStore(config.getBlobStoreLocation());
             log.info("offload {} uuid {} extraMetadata {} to {} {}", readHandle.getId(), uuid, extraMetadata,
                 config.getBlobStoreLocation(), writeBlobStore);
-            if (readHandle.getLength() == 0 || !readHandle.isClosed() || readHandle.getLastAddConfirmed() < 0) {
+            if (!readHandle.isClosed() || readHandle.getLastAddConfirmed() < 0) {
                 promise.completeExceptionally(
                         new IllegalArgumentException("An empty or open ledger should never be offloaded"));
                 return;
             }
+            if (readHandle.getLength() <= 0) {
+                log.warn("[{}] Ledger [{}] has zero length, but it contains {} entries."
+                    + " Attempting to offload ledger since it contains entries.", topicName, readHandle.getId(),
+                    readHandle.getLastAddConfirmed() + 1);
+            }
             OffloadIndexBlockBuilder indexBuilder = OffloadIndexBlockBuilder.create()
                 .withLedgerMetadata(readHandle.getLedgerMetadata())
                 .withDataBlockHeaderLength(BlockAwareSegmentInputStreamImpl.getHeaderSize());