Cleanup states from registrar when the files are removed

CHANGELOG.next.asciidoc

@@ -182,6 +182,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Fix the "No such input type exist: 'salesforce'" error on the Windows/AIX platform. {pull}41664[41664]
 - Fix missing key in streaming input logging. {pull}41600[41600]
 - Improve S3 object size metric calculation to support situations where Content-Length is not available. {pull}41755[41755]
+- Effectively clean states from removed files when clean_removed is set to True {pull}41747[41747]
 
 *Heartbeat*
 

filebeat/input/file/state.go

@@ -39,6 +39,7 @@ type State struct {
 	Meta           map[string]string `json:"meta" struct:"meta,omitempty"`
 	FileStateOS    file.StateOS      `json:"FileStateOS" struct:"FileStateOS"`
 	IdentifierName string            `json:"identifier_name" struct:"identifier_name"`
+	CleanRemoved   bool              `json:"clean_removed" struct:"clean_removed"`
 }
 
 // NewState creates a new file state

filebeat/input/file/states.go

@@ -18,6 +18,7 @@
 package file
 
 import (
+	"os"
 	"sync"
 	"time"
 
@@ -115,12 +116,18 @@ func (s *States) CleanupWith(fn func(string)) (int, int) {
 	numCanExpire := 0
 
 	L := len(s.states)
+	checkFileRemove := func(name string) bool {
+		if _, err := os.Stat(name); os.IsNotExist(err) {
+			return true
+		}
+		return false
+	}
 	for i := 0; i < L; {
 		state := &s.states[i]
 		canExpire := state.TTL > 0
 		expired := (canExpire && currentTime.Sub(state.Timestamp) > state.TTL)
 
-		if state.TTL == 0 || expired {
+		if state.TTL == 0 || expired || (state.CleanRemoved && checkFileRemove(state.Source)) {
 			if !state.Finished {
 				logp.Err("State for %s should have been dropped, but couldn't as state is not finished.", state.Source)
 				i++

filebeat/input/log/harvester.go

@@ -166,6 +166,11 @@ func NewHarvester(
 		h.state.TTL = h.config.CleanInactive
 	}
 
+	// Enable/disable cleaning removed files from the registrar
+	if h.config.CleanRemoved {
+		h.state.CleanRemoved = h.config.CleanRemoved
+	}
+
 	// Add outlet signal so harvester can also stop itself
 	return h, nil
 }
@@ -444,7 +449,10 @@ func (h *Harvester) onMessage(
 	// Check if json fields exist
 	var jsonFields mapstr.M
 	if f, ok := fields["json"]; ok {
-		jsonFields = f.(mapstr.M)
+		jsonFields, ok = f.(mapstr.M)
+		if !ok {
+			h.logger.Debugf("Error converting jsonFields from state %v", state.IdentifierName)
+		}
 	}
 
 	var meta mapstr.M