Merge pull request #22 from mmolimar/fix/close_reader

Close reader in source task
mmolimar · Dec 25, 2017 · 032dbd9 · 032dbd9
2 parents 015296d + 5f1d0b5
commit 032dbd9
Showing 1 changed file with 1 addition and 2 deletions.
diff --git a/src/main/java/com/github/mmolimar/kafka/connect/fs/FsSourceTask.java b/src/main/java/com/github/mmolimar/kafka/connect/fs/FsSourceTask.java
@@ -69,9 +69,8 @@ public List<SourceRecord> poll() throws InterruptedException {
             final List<SourceRecord> results = new ArrayList<>();
             List<FileMetadata> files = filesToProcess();
             files.forEach(metadata -> {
-                try {
+                try (FileReader reader = policy.offer(metadata, context.offsetStorageReader())) {
                     log.info("Processing records for file {}", metadata);
-                    FileReader reader = policy.offer(metadata, context.offsetStorageReader());
                     while (reader.hasNext()) {
                         results.add(convert(metadata, reader.currentOffset(), reader.next()));
                     }