Merge branch '8.1' into mergify/bp/8.1/pr-29710

Author: v1v

.ci/apm-beats-update.groovy

@@ -119,7 +119,8 @@ def beatsUpdate() {
     }
     dir("${BASE_DIR}"){
       git(credentialsId: 'f6c7695a-671e-4f4f-a331-acdce44ff9ba',
-        url:  "[email protected]:elastic/${env.REPO}.git")
+        url:  "[email protected]:elastic/${env.REPO}.git",
+        branch: env.BRANCH_NAME)
       sh(label: 'Update Beats script', script: """
         git config --global user.email "[email protected]"
         git config --global user.name "None"

.ci/packer_cache.sh

@@ -74,9 +74,10 @@ if [ -x "$(command -v docker)" ]; then
 
   ## GitHub api returns up to 100 entries.
   ## Probably we need a different approach to search the latest minor.
-  latestMinor=$(curl -s https://api.github.com/repos/elastic/beats/branches\?per_page=100 | jq -r '.[].name' | grep "^7." | tail -1)
+  latest7Minor=$(curl -s https://api.github.com/repos/elastic/beats/branches\?per_page=100 | jq -r '.[].name' | grep "^7." | tail -1)
+  latest8Minor=$(curl -s https://api.github.com/repos/elastic/beats/branches\?per_page=100 | jq -r '.[].name' | grep "^8." | tail -1)
 
-  for branch in main $latestMinor ; do
+  for branch in main $latest7Minor $latest8Minor; do
     if [ "$branch" != "main" ] ; then
       echo "Checkout the branch $branch"
       git checkout "$branch"

CHANGELOG.next.asciidoc

@@ -47,6 +47,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...main[Check the HEAD dif
 *Filebeat*
 
 - tcp/unix input: Stop accepting connections after socket is closed. {pull}29712[29712]
+- aws-s3: fix race condition in states used by s3-poller. {issue}30123[30123] {pull}30131[30131]
 
 *Heartbeat*
 

libbeat/docs/shared-env-vars.asciidoc

@@ -45,8 +45,7 @@ To specify custom error text, use:
 Where `error_text` is custom text that will be prepended to the error
 message if the environment variable cannot be expanded.
 
-If you need to use a literal `${` in your configuration file then you can write
-`$${` to escape the expansion.
+If you need to use a special character in your configuration file, use `$` to escape the expansion. For example, you can escape `${` or `}` with `$${` or `$}`.
 
 After changing the value of an environment variable, you need to restart
 {beatname_uc} to pick up the new value.

x-pack/filebeat/input/awss3/states.go

@@ -21,7 +21,9 @@ const (
 )
 
 type listingInfo struct {
-	totObjects    int
+	totObjects int
+
+	mu            sync.Mutex
 	storedObjects int
 	errorObjects  int
 	finalCheck    bool
@@ -105,11 +107,19 @@ func (s *states) Delete(id string) {
 func (s *states) IsListingFullyStored(listingID string) bool {
 	info, _ := s.listingInfo.Load(listingID)
 	listingInfo := info.(*listingInfo)
+	listingInfo.mu.Lock()
+	defer listingInfo.mu.Unlock()
 	if listingInfo.finalCheck {
 		return false
 	}
 
 	listingInfo.finalCheck = (listingInfo.storedObjects + listingInfo.errorObjects) == listingInfo.totObjects
+
+	if (listingInfo.storedObjects + listingInfo.errorObjects) > listingInfo.totObjects {
+		s.log.Warnf("unexepected mixmatch between storedObjects (%d), errorObjects (%d) and totObjects (%d)",
+			listingInfo.storedObjects, listingInfo.errorObjects, listingInfo.totObjects)
+	}
+
 	return listingInfo.finalCheck
 }
 
@@ -154,6 +164,9 @@ func (s *states) Update(newState state, listingID string) {
 	// here we increase the number of stored object
 	info, _ := s.listingInfo.Load(listingID)
 	listingInfo := info.(*listingInfo)
+
+	listingInfo.mu.Lock()
+
 	if newState.Stored {
 		listingInfo.storedObjects++
 	}
@@ -162,6 +175,8 @@ func (s *states) Update(newState state, listingID string) {
 		listingInfo.errorObjects++
 	}
 
+	listingInfo.mu.Unlock()
+
 	if _, ok := s.statesByListingID[listingID]; !ok {
 		s.statesByListingID[listingID] = make([]state, 0)
 	}