[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

launchable/commands/record/commit.py

@@ -108,6 +108,9 @@ def exec_jar(name: str, source: str, max_days: int, app: Application, is_collect
     # using subprocess.check_out with shell=False and a list of command to prevent vulnerability
     # https://knowledge-base.secureflag.com/vulnerabilities/code_injection/os_command_injection_python.html
     command = [java]
+    debug_opts = os.getenv("LAUNCHABLE_JAVA_DEBUG")
+    if debug_opts:
+        command.extend(debug_opts.split())
     command.extend(_build_proxy_option(os.getenv("HTTPS_PROXY")))
     command.extend([
         "-jar",

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,80 @@
+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;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * {@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;
+  /** # of threads that are blocked trying to {@link #execute(Runnable)}. Just for diagnostics. */
+  private final AtomicInteger blockCount = new AtomicInteger(0);
+
+  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 {
+      blockCount.incrementAndGet();
+      semaphore.acquire();
+    } catch (InterruptedException e) {
+      throw new RejectedExecutionException(e);
+    } finally {
+      blockCount.decrementAndGet();
+    }
+
+    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;
@@ -48,11 +47,12 @@ public void flush() throws IOException {
       return;
     }
 
-    try {
-      sender.accept(os -> writeTo(spool,os));
-    } finally {
-      spool.clear();
-    }
+    // let sender own the list -- do not reuse
+    // for this to work, we need to resolve `this.spool` here, not in the lambda
+    List<T> ref = this.spool;
+    this.spool = new ArrayList<>();
+
+    sender.accept(os -> writeTo(ref,os));
   }
 
   protected abstract void writeTo(List<T> spool, OutputStream os) throws IOException;