fix iteration (#413)

* fix iteration

* remove unneccesary if

* add a test case

operator/input/file/file.go

@@ -207,6 +207,8 @@ OUTER:
 			// Empty file, don't read it until we can compare its fingerprint
 			fps = append(fps[:i], fps[i+1:]...)
 			files = append(files[:i], files[i+1:]...)
+			i--
+			continue
 		}
 		for j := i + 1; j < len(fps); j++ {
 			fp2 := fps[j]
@@ -217,6 +219,7 @@ OUTER:
 				}
 				fps = append(fps[:i], fps[i+1:]...)
 				files = append(files[:i], files[i+1:]...)
+				i--
 				continue OUTER
 			}
 		}

operator/input/file/file_test.go

@@ -487,6 +487,30 @@ func TestSplitWrite(t *testing.T) {
 	waitForMessage(t, logReceived, "testlog1testlog2")
 }
 
+func TestIgnoreEmptyFiles(t *testing.T) {
+	t.Parallel()
+	operator, logReceived, tempDir := newTestFileOperator(t, nil, nil)
+	operator.persister = testutil.NewMockPersister("test")
+	defer operator.Stop()
+
+	temp := openTemp(t, tempDir)
+	temp2 := openTemp(t, tempDir)
+	temp3 := openTemp(t, tempDir)
+	temp4 := openTemp(t, tempDir)
+
+	writeString(t, temp, "testlog1\n")
+	writeString(t, temp3, "testlog2\n")
+	operator.poll(context.Background())
+
+	waitForMessages(t, logReceived, []string{"testlog1", "testlog2"})
+
+	writeString(t, temp2, "testlog3\n")
+	writeString(t, temp4, "testlog4\n")
+	operator.poll(context.Background())
+
+	waitForMessages(t, logReceived, []string{"testlog3", "testlog4"})
+}
+
 func TestDecodeBufferIsResized(t *testing.T) {
 	t.Parallel()
 	operator, logReceived, tempDir := newTestFileOperator(t, nil, nil)