fix: 代码review问题处理

filebeat/input/log/harvester.go

@@ -204,6 +204,8 @@ func (h *Harvester) Run() error {
 			removedCheckTick = time.After(h.config.ScanFrequency)
 		}
 
+		removedTick := make(<-chan time.Time)
+
 	L:
 		for {
 			select {
@@ -217,13 +219,14 @@ func (h *Harvester) Run() error {
 				if h.reader.fileReader.log.fs.Removed() {
 					logp.Info("Closing harvester because file was removed: %s, wait for 60s", source)
 					// 等待 60s 才停止，避免日志没来得及采完就关掉导致漏采
-					time.Sleep(60 * time.Second)
-					break L
+					removedTick = time.After(60 * time.Second)
 				} else {
 					logp.Debug("harvester", "File was not removed: %s, check again after %v", source, h.config.ScanFrequency)
 					// update timer
 					removedCheckTick = time.After(h.config.ScanFrequency)
 				}
+			case <-removedTick:
+				break L
 			// Required when reader loop returns and reader finished
 			case <-h.done:
 				break L

libbeat/reader/readfile/line.go

@@ -80,7 +80,10 @@ func (r *LineReader) Next() ([]byte, int, error) {
 		if err != nil {
 			if r.inBuffer.Len() == 0 || !(err.Error() == "file was removed" || err.Error() == "file inactive") {
 				// buffer为空，或者不属于文件被删除以及文件不活跃的错误，直接返回空
-				return nil, 0, err
+				// return and reset consumed bytes count
+				sz := r.byteCount
+				r.byteCount = 0
+				return nil, sz, err
 			}
 
 			logp.Info("LineReader get an advance err: %s, send all %d bytes in buffer", err, r.inBuffer.Len())