test(batch): send results only after receiving all 3 batches

Author: bevzzz

src/test/java/io/weaviate/client6/v1/api/collections/batch/BatchContextTest.java

@@ -430,17 +430,18 @@ public void test_reconnect_DrainAfterStreamHangup() throws Exception {
     out.emitEventAsync(Event.STARTED);
     Future<?> backgroundAcks = backgroundThread.submit(() -> {
       try {
-        recvDataAndAck();
-        recvDataAndAck();
-        recvDataAndAck();
+        List<String> received = new ArrayList<>();
+
+        received.addAll(recvDataAndAck());
+        received.addAll(recvDataAndAck());
+        received.addAll(recvDataAndAck());
+
+        out.emitEvent(new Event.Results(received, Collections.emptyMap()));
       } catch (Exception e) {
         throw new RuntimeException(e);
       }
     });
 
-    List<String> submitted = tasks.stream().map(TaskHandle::id).toList();
-    out.emitEventAsync(new Event.Results(submitted, Collections.emptyMap()));
-
     closeContext();
     backgroundAcks.get();
 
@@ -546,14 +547,20 @@ public void test_startAfterClose() throws Exception {
   /**
    * Read the next Data message from the stream and ACK it.
    * This method does not wait for the server to process the Acks.
+   *
+   * @return IDs of tasks that were received.
    */
   private List<String> recvDataAndAck() throws InterruptedException {
     List<String> received = recvData();
     out.emitEventAsync(new Event.Acks(received));
     return received;
   }
 
-  /** Read the next Data message from the stream. */
+  /**
+   * Read the next Data message from the stream.
+   *
+   * @return IDs of tasks that were received.
+   */
   private List<String> recvData() throws InterruptedException {
     WeaviateProtoBatch.BatchStreamRequest.Data data = in.expectMessage(DATA).getData();
     return Stream.concat(