handle EOF when checking archive validity

lib/filewatch/read_mode/handlers/read_zip_file.rb

@@ -71,14 +71,14 @@ def close_and_ignore_ioexception(closeable)
 
     def corrupted?(watched_file)
       begin
+        start = Time.new
         file_stream = FileInputStream.new(watched_file.path)
         gzip_stream = GZIPInputStream.new(file_stream)
         buffer = Java::byte[8192].new
-        start = Time.new
         until gzip_stream.read(buffer) == -1
         end
         return false
-      rescue ZipException => e
+      rescue ZipException, Java::JavaIo::EOFException => e
         duration = Time.now - start
         logger.warn("Detected corrupted archive #{watched_file.path} file won't be processed", :message => e.message,
                     :duration => duration.round(3))

spec/helpers/spec_helper.rb

@@ -24,6 +24,12 @@ def self.corrupt_gzip(file_path)
     f.close()
   end
 
+  def self.truncate_gzip(file_path)
+    f = File.open(file_path, "ab")
+    f.truncate(100)
+    f.close()
+  end
+
   class TracerBase
     def initialize
       @tracer = Concurrent::Array.new

spec/inputs/file_read_spec.rb

@@ -245,6 +245,37 @@
           expect(IO.read(log_completed_path)).to be_empty
         end
       end
+
+      it "the truncated file is untouched" do
+        directory = Stud::Temporary.directory
+        file_path = fixture_dir.join('compressed.log.gz')
+        truncated_file_path = ::File.join(directory, 'truncated.gz')
+        FileUtils.cp(file_path, truncated_file_path)
+
+        FileInput.truncate_gzip(truncated_file_path)
+
+        log_completed_path = ::File.join(directory, "C_completed.txt")
+        f = File.new(log_completed_path, "w")
+        f.close()
+
+        conf = <<-CONFIG
+        input {
+          file {
+            type => "blah"
+            path => "#{truncated_file_path}"
+            mode => "read"
+            file_completed_action => "log_and_delete"
+            file_completed_log_path => "#{log_completed_path}"
+            check_archive_validity => true
+          }
+        }
+        CONFIG
+
+        events = input(conf) do |pipeline, queue|
+          wait(1)
+          expect(IO.read(log_completed_path)).to be_empty
+        end
+      end
     end
   end