fixed issue in ydb table scan when the queue is full at the end of the scan

Author: zinal

NOTES.md

@@ -36,6 +36,8 @@ cd /home/zinal/Software/spark-3.3.3-bin-hadoop3
 ```sql
 create schema ydb1.spark;
 create table ydb1.spark.test1(a integer not null, b bigint, c varchar(100)) tblproperties('primary_key'='b,a');
+
+CREATE TABLE ydb1.fhrw2 TBLPROPERTIES('primary_key'='h2,unique_key') AS SELECT hash(unique_key) AS h2, x.* FROM ydb1.fhrw0 x;
 ```
 
 

src/main/java/tech/ydb/spark/connector/YdbScanViaReadTable.java

@@ -274,6 +274,22 @@ private void configureRanges(ReadTableSettings.Builder rtsb) {
         }
     }
 
+    private void putToQueue(QueueItem qi) {
+        while (true) {
+            try {
+                queue.add(qi);
+                break; // exit the "queue put" retry loop
+            } catch (IllegalStateException ise) {
+                // The unlikely case of thread interrupt should not prevent us
+                // from putting an item into the queue
+                try {
+                    Thread.sleep(35L);
+                } catch (InterruptedException ix) {
+                }
+            }
+        }
+    }
+
     static final class QueueItem {
 
         final ResultSetReader reader;
@@ -298,21 +314,9 @@ public void run() {
             try {
                 stream.start(part -> {
                     final ResultSetReader rsr = part.getResultSetReader();
-                    while (true) {
-                        try {
-                            queue.add(new QueueItem(rsr));
-                            LOG.debug("Added portion of {} rows for table {} to the queue.",
-                                    rsr.getRowCount(), tablePath);
-                            return; // exit the "queue put" retry loop - and lambda too
-                        } catch (IllegalStateException ise) {
-                            // The unlikely case of interrupt should not prevent us
-                            // from putting an item into the queue
-                            try {
-                                Thread.sleep(35L);
-                            } catch (InterruptedException ix) {
-                            }
-                        }
-                    }
+                    putToQueue(new QueueItem(rsr));
+                    LOG.debug("Added portion of {} rows for table {} to the queue.",
+                            rsr.getRowCount(), tablePath);
                 }).join().expectSuccess();
             } catch (Exception ex) {
                 boolean needReport = true;
@@ -327,7 +331,7 @@ public void run() {
                     setIssue(ex);
                 }
             }
-            queue.add(END_OF_SCAN);
+            putToQueue(END_OF_SCAN);
             LOG.debug("Completed background scan for table {}, range {}", tablePath, keyRange);
         }
     }