fix entry duplication at beginning of file(s) (#579)

operator/builtin/input/file/file.go

@@ -235,7 +235,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++ {
@@ -244,6 +245,7 @@ OUTER:
 				// Exclude
 				fps = append(fps[:i], fps[i+1:]...)
 				files = append(files[:i], files[i+1:]...)
+				i--
 				continue OUTER
 			}
 		}

operator/builtin/input/file/file_test.go

@@ -324,6 +324,29 @@ func TestSplitWrite(t *testing.T) {
 	waitForMessage(t, logReceived, "testlog1testlog2")
 }
 
+func TestIgnoreEmptyFiles(t *testing.T) {
+	t.Parallel()
+	operator, logReceived, tempDir := newTestFileOperator(t, nil, nil)
+	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)