fix to handle an exception for each record

scalardb/src/scalardb/transfer.clj

@@ -127,12 +127,12 @@
 (defn- read-record
   "Read and update the specified record with a transaction.
   Return nil if the read fails or a conflict happens."
-  [test storage i]
+  [test i]
   (try
     (let [tx (scalar/start-transaction test)
           tx-result (.get tx (prepare-get i))
           ;; Need Storage API to read the transaction metadata
-          result (.get storage (prepare-get i))]
+          result (.get @(:storage test) (prepare-get i))]
       ;; Put the same balance to check conflicts with in-flight transactions
       (->> (calc-new-balance tx-result 0)
            (prepare-put i)
@@ -144,18 +144,22 @@
       (warn (.getMessage e))
       nil)))
 
-(defn read-all-with-retry
-  [test n]
-  (scalar/check-transaction-connection! test)
-  (scalar/check-storage-connection! test)
+(defn- read-record-with-retry
+  [test i]
   (scalar/with-retry
     (fn [test]
       (scalar/prepare-transaction-service! test)
       (scalar/prepare-storage-service! test))
     test
-    (let [results (pmap #(read-record test @(:storage test) %) (range n))]
-      (when (every? #(some? %) results)
-        results))))
+    (read-record test i)))
+
+(defn read-all-with-retry
+  [test n]
+  (scalar/check-transaction-connection! test)
+  (scalar/check-storage-connection! test)
+  (try
+    (pmap #(read-record-with-retry test %) (range n))
+    (catch Exception _ nil)))
 
 (defrecord TransferClient [initialized? n initial-balance max-txs]
   client/Client