[AIENG-289] parallel & asynchronous file processing for better reliability & performance

CLAUDE.md

@@ -0,0 +1,2 @@
+- In Java code, use JUnit 4 & Google truth. Test needs to be added to AllTests.java or else it won't run. 
+- See README.md & CONTRIBUTING.md

README.md

@@ -24,6 +24,9 @@ configuration for [pre-commit](https://pre-commit.com). Install the hook with
 pipenv shell
 ```
 
+From this shell, run `pip install -e .` to create `launchable` wrapper script from the workspace.
+This is useful to test the CLI with real/local server.
+
 ## Run tests cli
 
 ```shell

src/main/java/com/launchableinc/ingest/commits/BackgroundWorkStatus.java

@@ -0,0 +1,8 @@
+package com.launchableinc.ingest.commits;
+
+public enum BackgroundWorkStatus {
+  IN_PROGRESS,
+  SUCCEEDED,
+  FAILED,
+  ABANDONED
+}

src/main/java/com/launchableinc/ingest/commits/BoundedExecutorService.java

@@ -0,0 +1,73 @@
+package com.launchableinc.ingest.commits;
+
+import java.util.List;
+import java.util.concurrent.AbstractExecutorService;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * {@link ExecutorService} decorator that limits the number of concurrent tasks,
+ * and make the caller block when the limit is reached.
+ */
+class BoundedExecutorService extends AbstractExecutorService {
+  private final ExecutorService delegate;
+  private final Semaphore semaphore;
+
+  BoundedExecutorService(int limit) {
+    this(Executors.newFixedThreadPool(limit), limit);
+  }
+
+  BoundedExecutorService(ExecutorService delegate, int limit) {
+    this.delegate = delegate;
+    this.semaphore = new Semaphore(limit);
+  }
+
+  @Override
+  public void execute(Runnable command) {
+    try {
+      semaphore.acquire();
+    } catch (InterruptedException e) {
+      throw new RejectedExecutionException(e);
+    }
+    try {
+      delegate.execute(() -> {
+        try {
+          command.run();
+        } finally {
+          semaphore.release();
+        }
+      });
+    } catch (RejectedExecutionException e) {
+      semaphore.release();
+      throw e;
+    }
+  }
+
+  @Override
+  public void shutdown() {
+    delegate.shutdown();
+  }
+
+  @Override
+  public List<Runnable> shutdownNow() {
+    return delegate.shutdownNow();
+  }
+
+  @Override
+  public boolean isShutdown() {
+    return delegate.isShutdown();
+  }
+
+  @Override
+  public boolean isTerminated() {
+    return delegate.isTerminated();
+  }
+
+  @Override
+  public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
+    return delegate.awaitTermination(timeout, unit);
+  }
+}

src/main/java/com/launchableinc/ingest/commits/ChunkStreamer.java

@@ -2,7 +2,6 @@
 
 import org.apache.http.entity.ContentProducer;
 
-import java.io.Closeable;
 import java.io.IOException;
 import java.io.OutputStream;
 import java.io.UncheckedIOException;
@@ -12,13 +11,13 @@
 /**
  * Accepts T, buffers them, and writes them out as a batch.
  */
-abstract class ChunkStreamer<T> implements FlushableConsumer<T>, Closeable {
+abstract class ChunkStreamer<T> implements FlushableConsumer<T> {
   /**
    * Encapsulation of how batches are sent.
    */
   private final IOConsumer<ContentProducer> sender;
   private final int chunkSize;
-  private final List<T> spool = new ArrayList<>();
+  private List<T> spool = new ArrayList<>();
 
   ChunkStreamer(IOConsumer<ContentProducer> sender, int chunkSize) {
     this.sender = sender;
@@ -50,6 +49,7 @@ public void flush() throws IOException {
 
     try {
       sender.accept(os -> writeTo(spool,os));
+      // let sender own the list -- do not reuse
     } finally {
       spool.clear();
     }