Fix finding job IDs in ingest batcher store

gchq · Dec 18, 2024 · 794451a · 794451a
1 parent 9c2c75b
commit 794451a
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 5 deletions.
diff --git a/.../system-test-dsl/src/main/java/sleeper/systemtest/dsl/ingest/SystemTestIngestBatcher.java b/.../system-test-dsl/src/main/java/sleeper/systemtest/dsl/ingest/SystemTestIngestBatcher.java
@@ -64,7 +64,7 @@ public SystemTestIngestBatcher sendSourceFilesExpectingJobs(int expectedJobs, St
                             () -> jobIdsInStore().filter(not(jobIdsBefore::contains)).toList(),
                             ids -> ids.size() >= expectedJobs);
             if (newJobIds.size() > expectedJobs) {
-                throw new IllegalStateException("More jobs were created than expected, found " + newJobIds.size() + ", expected" + expectedJobs);
+                throw new IllegalStateException("More jobs were created than expected, found " + newJobIds.size() + ", expected " + expectedJobs);
             }
             createdJobIds.addAll(newJobIds);
         } catch (InterruptedException e) {
@@ -94,6 +94,7 @@ private Stream<String> jobIdsInStore() {
         return driver.batcherStore().getAllFilesNewestFirst().stream()
                 .filter(request -> tableIds.contains(request.getTableId()))
                 .filter(request -> request.getJobId() != null)
-                .map(FileIngestRequest::getJobId);
+                .map(FileIngestRequest::getJobId)
+                .distinct();
     }
 }
diff --git a/...sl/src/test/java/sleeper/systemtest/dsl/testutil/drivers/InMemoryIngestBatcherDriver.java b/...sl/src/test/java/sleeper/systemtest/dsl/testutil/drivers/InMemoryIngestBatcherDriver.java
@@ -19,11 +19,13 @@
 import sleeper.ingest.batcher.core.FileIngestRequest;
 import sleeper.ingest.batcher.core.IngestBatcher;
 import sleeper.ingest.batcher.core.IngestBatcherStore;
+import sleeper.ingest.core.job.IngestJob;
 import sleeper.systemtest.dsl.SystemTestContext;
 import sleeper.systemtest.dsl.ingest.IngestBatcherDriver;
 import sleeper.systemtest.dsl.instance.SystemTestInstanceContext;
 
 import java.time.Instant;
+import java.util.ArrayList;
 import java.util.List;
 
 import static sleeper.core.properties.table.TableProperty.TABLE_ID;
@@ -54,14 +56,16 @@ public void sendFiles(List<String> files) {
                     .receivedTime(Instant.now())
                     .build());
         }
+        List<IngestJob> jobs = new ArrayList<>();
         IngestBatcher.builder()
                 .instanceProperties(instance.getInstanceProperties())
                 .tablePropertiesProvider(instance.getTablePropertiesProvider())
                 .store(store)
-                .queueClient((queueUrl, job) -> {
-                    ingest.send(job, context);
-                })
+                .queueClient((queueUrl, job) -> jobs.add(job))
                 .build().batchFiles();
+        for (IngestJob job : jobs) {
+            ingest.send(job, context);
+        }
     }
 
     @Override