Fix empty lines sent

Empty lines should not be sent as an event. To make sure this does not happen a test was added.

This was probably broken in elastic#2090

filebeat/harvester/log.go

@@ -91,8 +91,9 @@ func (h *Harvester) Harvest() {
 
 		text := string(message.Content)
 
-		// Check if data should be added to event
-		if h.shouldExportLine(text) {
+		// Check if data should be added to event. Only export events where Bytes > 0
+		// Content length cannot be checked because of JSON
+		if message.Bytes > 0 && h.shouldExportLine(text) {
 			event.ReadTime = message.Ts
 			event.Bytes = message.Bytes
 			event.Text = &text

filebeat/tests/system/test_harvester.py

@@ -146,3 +146,53 @@ def test_close_eof(self):
 
         # Make sure the state for the file was persisted
         assert len(data) == 1
+
+
+    def test_empty_line(self):
+        """
+        Checks that no empty events are sent for an empty line but state is still updated
+        """
+        self.render_config_template(
+            path=os.path.abspath(self.working_dir) + "/log/test.log",
+        )
+        os.mkdir(self.working_dir + "/log/")
+
+        logfile = self.working_dir + "/log/test.log"
+
+        filebeat = self.start_beat()
+
+        with open(logfile, 'w') as f:
+            f.write("Hello world\n")
+
+        # Let it read the file
+        self.wait_until(
+            lambda: self.output_has(lines=1), max_timeout=10)
+
+        with open(logfile, 'a') as f:
+            f.write("\n")
+
+        expectedOffset = 13
+
+        if os.name == "nt":
+            # Two additional newline chars
+            expectedOffset += 2
+
+        # Wait until offset for new line is updated
+        self.wait_until(
+            lambda: self.log_contains(
+                "offset: " + str(expectedOffset)),
+            max_timeout=15)
+
+        with open(logfile, 'a') as f:
+            f.write("Third line\n")
+
+        # Make sure only 2 events are written
+        self.wait_until(
+            lambda: self.output_has(lines=2), max_timeout=10)
+
+        filebeat.check_kill_and_wait()
+
+        data = self.get_registry()
+
+        # Make sure the state for the file was persisted
+        assert len(data) == 1