fix(java-toolkit): close reader when failed

hstreamdb · May 29, 2024 · f40764f · f40764f
1 parent e083182
commit f40764f
Showing 1 changed file with 5 additions and 2 deletions.
diff --git a/java-toolkit/lib/src/main/java/io/hstream/io/impl/SinkTaskContextImpl.java b/java-toolkit/lib/src/main/java/io/hstream/io/impl/SinkTaskContextImpl.java
@@ -124,13 +124,16 @@ public void handleInternal(Consumer<SinkRecordBatch> handler, boolean parallel)
                             }
                             retry = 0;
                         } catch (Exception e) {
-                            log.error("read records failed, retry:{}, ", retry, e);
+                            log.error("read records from shard {} failed, retry:{}, ", shard.getShardId(), retry, e);
                             retry++;
                             if (retry > maxRetry) {
                                 throw new RuntimeException("retry failed");
                             }
                             Thread.sleep(retry * 3000L);
-                            reader = null;
+                            if (reader != null) {
+                                reader.close();
+                                reader = null;
+                            }
                         }
                     }
                 } catch (Exception e) {