finish emitting any records still in the queue before is_done() or raising exceptions

Author: brianjlai

airbyte_cdk/sources/concurrent_source/concurrent_source.py

@@ -147,7 +147,10 @@ def _consume_from_queue(
                 airbyte_message_or_record_or_exception,
                 concurrent_stream_processor,
             )
-            if concurrent_stream_processor.is_done() and queue.empty():
+            # In the event that a partition raises an exception, anything remaining in
+            # the queue will be missed because is_done() can raise an exception and exit
+            # out of this loop before remaining items are consumed
+            if queue.empty() and concurrent_stream_processor.is_done():
                 # all partitions were generated and processed. we're done here
                 break