[New Physical IO] Migrating new design with Blob

buildSrc/src/main/kotlin/buildlogic.java-common-conventions.gradle.kts

@@ -109,6 +109,7 @@ tasks.named<Test>("test") {
 }
 
 tasks.test {
+    maxHeapSize = "2G"
     // Report is generated and verification is run after tests
     finalizedBy(tasks.jacocoTestReport, tasks.jacocoTestCoverageVerification)
 }

input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/S3SeekableInputStreamFactory.java

@@ -82,12 +82,17 @@ public S3SeekableInputStreamFactory(
     this.objectMetadataStore =
         new MetadataStore(objectClient, telemetry, configuration.getPhysicalIOConfiguration());
     this.objectFormatSelector = new ObjectFormatSelector(configuration.getLogicalIOConfiguration());
-    this.objectBlobStore =
-        new BlobStore(objectClient, telemetry, configuration.getPhysicalIOConfiguration(), metrics);
     // TODO: calling applications should be able to pass in a thread pool if they so wish
     this.threadPool =
         Executors.newFixedThreadPool(
             configuration.getPhysicalIOConfiguration().getThreadPoolSize());
+    this.objectBlobStore =
+        new BlobStore(
+            objectClient,
+            telemetry,
+            configuration.getPhysicalIOConfiguration(),
+            metrics,
+            threadPool);
     objectBlobStore.schedulePeriodicCleanup();
   }
 
@@ -194,5 +199,6 @@ public void close() throws IOException {
     this.objectMetadataStore.close();
     this.objectBlobStore.close();
     this.telemetry.close();
+    this.threadPool.shutdown();
   }
 }

input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/data/Blob.java

@@ -94,7 +94,7 @@ public int read(long pos) throws IOException {
     try {
       lock.readLock().lock();
       blockManager.makePositionAvailable(pos, ReadMode.SYNC);
-      return blockManager.getBlock(pos).get().read(pos);
+      return blockManager.getBlock(pos).get().read(pos); // TODO add if block exist check
     } finally {
       lock.readLock().unlock();
     }
@@ -126,6 +126,8 @@ public int read(byte[] buf, int off, int len, long pos) throws IOException {
 
       while (numBytesRead < len && nextPosition < contentLength()) {
         final long nextPositionFinal = nextPosition;
+
+        // TODO throw IOException here
         Block nextBlock =
             blockManager
                 .getBlock(nextPosition)
@@ -134,7 +136,7 @@ public int read(byte[] buf, int off, int len, long pos) throws IOException {
                         new IllegalStateException(
                             String.format(
                                 "This block object key %s (for position %s) should have been available.",
-                                objectKey.getS3URI().toString(), nextPositionFinal)));
+                                objectKey.getS3URI(), nextPositionFinal)));
 
         int bytesRead = nextBlock.read(buf, off + numBytesRead, len - numBytesRead, nextPosition);
 

input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/data/BlobStore.java

@@ -20,6 +20,7 @@
 import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -48,6 +49,7 @@ public class BlobStore implements Closeable {
   private final ObjectClient objectClient;
   private final Telemetry telemetry;
   private final PhysicalIOConfiguration configuration;
+  private final ExecutorService threadPool;
 
   @Getter private final Metrics metrics;
   final BlobStoreIndexCache indexCache;
@@ -62,12 +64,14 @@ public class BlobStore implements Closeable {
    * @param telemetry an instance of {@link Telemetry} to use
    * @param configuration the PhysicalIO configuration
    * @param metrics an instance of {@link Metrics} to track metrics across the factory
+   * @param threadPool a thread pool for async operations
    */
   public BlobStore(
       @NonNull ObjectClient objectClient,
       @NonNull Telemetry telemetry,
       @NonNull PhysicalIOConfiguration configuration,
-      @NonNull Metrics metrics) {
+      @NonNull Metrics metrics,
+      @NonNull ExecutorService threadPool) {
     this.objectClient = objectClient;
     this.telemetry = telemetry;
     this.metrics = metrics;
@@ -82,6 +86,7 @@ public BlobStore(
               return cleanupThread;
             });
     this.configuration = configuration;
+    this.threadPool = threadPool;
   }
 
   /** Schedules a periodic cleanup task to sync the blop map with the index cache */
@@ -132,14 +137,15 @@ public Blob get(
                 uri,
                 metadata,
                 new BlockManager(
-                    uri,
+                    objectKey,
                     objectClient,
                     metadata,
                     telemetry,
                     configuration,
                     metrics,
                     indexCache,
-                    openStreamInformation),
+                    openStreamInformation,
+                    threadPool),
                 telemetry));
   }