Write any records in the queue before ending session.

parent cc943be5
Pipeline #4390 passed with stage
in 9 minutes and 31 seconds
......@@ -166,6 +166,9 @@ class DuplexOutgoingSession implements SyncSession, EventListener {
dataToFlush = true;
}
}
// Write any records that were already in the queue
ThrowingRunnable<IOException> task;
while ((task = writerTasks.poll()) != null) task.run();
streamWriter.sendEndOfStream();
} catch (InterruptedException e) {
LOG.info("Interrupted while waiting for a record to write");
......
......@@ -95,6 +95,9 @@ class SimplexOutgoingSession implements SyncSession, EventListener {
if (task == CLOSE) break;
task.run();
}
// Write any records that were already in the queue
ThrowingRunnable<IOException> task;
while ((task = writerTasks.poll()) != null) task.run();
streamWriter.sendEndOfStream();
} catch (InterruptedException e) {
LOG.info("Interrupted while waiting for a record to write");
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment