WIP

Author: j-coll

.github/workflows/test-xetabase-workflow.yml

@@ -98,7 +98,8 @@ jobs:
         run: |
           cd opencga-enterprise
           ln -s ../opencga opencga-home
-          ./build.sh --test --test-level runShortTests --prepare-branches --test-save-reports --task ${{ inputs.task }} --storage-hadoop ${{ vars.HADOOP_FLAVOUR }} --prepare-hadoop
+          ls -la build.sh
+          ./build.sh --test --test-level runShortTests --prepare-branches --test-save-reports --task ${{ inputs.task }} --storage-hadoop ${{ vars.HADOOP_FLAVOUR }} --prepare-hadoop --verbose
         env:
           THIRDPARTY_READ_TOKEN: ${{ secrets.THIRDPARTY_READ_TOKEN }}
       - name: Upload reports results to Github

commons-lib/src/main/java/org/opencb/commons/run/ParallelTaskRunner.java

@@ -182,6 +182,10 @@ public ParallelTaskRunner.Config build() {
                 if (capacity < 0) {
                     capacity = numTasks * 2;
                 }
+                if (sorted && numTasks == 1) {
+                    logger.debug("Sorted ParallelTaskRunner with a single task. Setting sorted to false.");
+                    sorted = false;
+                }
                 return new ParallelTaskRunner.Config(numTasks, batchSize, capacity, abortOnFail, sorted, readQueuePutTimeoutSeconds);
             }
         }
@@ -802,16 +806,17 @@ public Void call() throws InterruptedException {
                             }
                         }
                     } else if (writeBlockingQueueFuture != null) {
-                        status = TaskRunnableStatus.WRITING_BATCH_TO_QUEUE;
                         // Sorted PTR should not have to drain!
-                        CompletableFuture<Batch<O>> future = new CompletableFuture<>();
-                        future.complete(new Batch<O>(batchResult, batch.position + 1));
-                        while (!writeBlockingQueueFuture.offer(future, TIMEOUT_CHECK, TimeUnit.SECONDS)) {
-                            if (isAbortPending()) {
-                                logger.warn("Abort task thread on fail");
-                                break;
-                            }
-                        }
+                        throw new IllegalStateException("Sorted ParallelTaskRunner should not have to drain!");
+//                        status = TaskRunnableStatus.WRITING_BATCH_TO_QUEUE;
+//                        CompletableFuture<Batch<O>> future = new CompletableFuture<>();
+//                        future.complete(new Batch<O>(batchResult, batch.position + 1));
+//                        while (!writeBlockingQueueFuture.offer(future, TIMEOUT_CHECK, TimeUnit.SECONDS)) {
+//                            if (isAbortPending()) {
+//                                logger.warn("Abort task thread on fail");
+//                                break;
+//                            }
+//                        }
                     }
                 }
             } catch (Error e) {