Repository: redpanda-data/connect (language: go)

Issue:
Files are not all removed for azure_blob_storage with max_in_flight value greater than 1
Hi, we're observing that, if we set `max_in_flight` greater than 1, all files are not removed. With log level at `DEBUG` we can see that all files are handled, so it's seems to works.

This is our input resource configuration

```
---
input_resources:
  - label: main
    azure_blob_storage:
      storage_connection_string: ${AZURE_STORAGE_CONNECTION_STRING}
      storage_sas_token: ${AZURE_STORAGE_SAS_TOKEN:""}
      container: ${AZURE_CONTAINER}
      delete_objects: ${AZURE_STORAGE_INPUT_DELETE_OBJECT:true}
      prefix: "replay/"
      scanner:
        to_the_end: {}
      targets_input:
        azure_queue_storage:
          storage_connection_string: ${AZURE_STORAGE_CONNECTION_STRING}
          queue_name: ${AZURE_QUEUE_IN}
          dequeue_visibility_timeout: ${AZURE_DEQUEUE_TIMEOUT:30s}
          max_in_flight: ${AZURE_QUEUE_MAX_IN_FLIGHT:1}
          track_properties: false
        processors:
          - mapping: |-
              let json = content().decode("base64").parse_json()
              if $json.eventType == "Microsoft.Storage.BlobCreated" {
                root.name = $json.data.url.parse_url().path.trim_prefix("/" + env("AZURE_CONTAINER") + "/")
              } else {
                root = deleted()
              }
```


Discussion / hints:

Fix patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 13d5e09f8a..ad9ae44758 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,6 +11,10 @@ All notable changes to this project will be documented in this file.
 - Field `handle_logical_types` added to `parquet_decode` input to provide better handling of Parquet logical types (@ooesili)
 - New `gateway` input. (@Jeffail)
 
+### Fixed
+
+- The `azure_blob_storage` input now drops `targets_input` notifications and emits a warning log message for blobs which have been deleted before Connect was able to read them. (@mihaitodor)
+
 ## 4.50.0 - 2025-03-18
 
 ### Added
diff --git a/go.mod b/go.mod
index 8f36b886fe..7ed9dcf42b 100644
--- a/go.mod
+++ b/go.mod
@@ -10,12 +10,12 @@ require (
 	cloud.google.com/go/pubsub v1.45.1
 	cloud.google.com/go/storage v1.43.0
 	cloud.google.com/go/vertexai v0.12.0
-	github.com/Azure/azure-sdk-for-go/sdk/azcore v1.14.0
-	github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.8.0
-	github.com/Azure/azure-sdk-for-go/sdk/data/azcosmos v1.0.3
-	github.com/Azure/azure-sdk-for-go/sdk/data/aztables v1.2.0
-	github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.4.1
-	github.com/Azure/azure-sdk-for-go/sdk/storage/azdatalake v1.2.1
+	github.com/Azure/azure-sdk-for-go/sdk/azcore v1.17.1
+	github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.8.2
+	github.com/Azure/azure-sdk-for-go/sdk/data/azcosmos v1.3.0
+	github.com/Azure/azure-sdk-for-go/sdk/data/aztables v1.3.0
+	github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.6.0
+	github.com/Azure/azure-sdk-for-go/sdk/storage/azdatalake v1.4.0
 	github.com/Azure/azure-sdk-for-go/sdk/storage/azqueue v1.0.0
 	github.com/Azure/go-amqp v1.0.5
 	github.com/ClickHouse/clickhouse-go/v2 v2.27.1
@@ -216,7 +216,7 @@ require (
 	github.com/Azure/azure-sdk-for-go v68.0.0+incompatible // indirect
 	github.com/Azure/azure-sdk-for-go/sdk/internal v1.10.0 // indirect
 	github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 // indirect
-	github.com/AzureAD/microsoft-authentication-library-for-go v1.2.2 // indirect
+	github.com/AzureAD/microsoft-authentication-library-for-go v1.3.3 // indirect
 	github.com/ClickHouse/ch-go v0.61.5 // indirect
 	github.com/DataDog/zstd v1.5.2 // indirect
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.48.1 // indirect
diff --git a/go.sum b/go.sum
index 656f983580..7adea3c054 100644
--- a/go.sum
+++ b/go.sum
@@ -653,17 +653,17 @@ github.com/Azure/azure-pipeline-go v0.2.3/go.mod h1:x841ezTBIMG6O3lAcl8ATHnsOPVl
 github.com/Azure/azure-sdk-for-go v68.0.0+incompatible h1:fcYLmCpyNYRnvJbPerq7U0hS+6+I79yEDJBqVNcqUzU=
 github.com/Azure/azure-sdk-for-go v68.0.0+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc=
 github.com/Azure/azure-sdk-for-go/sdk/azcore v0.19.0/go.mod h1:h6H6c8enJmmocHUbLiiGY6sx7f9i+X3m1CHdd5c6Rdw=
-github.com/Azure/azure-sdk-for-go/sdk/azcore v1.14.0 h1:nyQWyZvwGTvunIMxi1Y9uXkcyr+I7TeNrr/foo4Kpk8=
-github.com/Azure/azure-sdk-for-go/sdk/azcore v1.14.0/go.mod h1:l38EPgmsp71HHLq9j7De57JcKOWPyhrsW1Awm1JS6K0=
+github.com/Azure/azure-sdk-for-go/sdk/azcore v1.17.1 h1:DSDNVxqkoXJiko6x8a90zidoYqnYYa6c1MTzDKzKkTo=
+github.com/Azure/azure-sdk-for-go/sdk/azcore v1.17.1/go.mod h1:zGqV2R4Cr/k8Uye5w+dgQ06WJtEcbQG/8J7BB6hnCr4=
 github.com/Azure/azure-sdk-for-go/sdk/azidentity v0.11.0/go.mod h1:HcM1YX14R7CJcghJGOYCgdezslRSVzqwLf/q+4Y2r/0=
-github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.8.0 h1:B/dfvscEQtew9dVuoxqxrUKKv8Ih2f55PydknDamU+g=
-github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.8.0/go.mod h1:fiPSssYvltE08HJchL04dOy+RD4hgrjph0cwGGMntdI=
-github.com/Azure/azure-sdk-for-go/sdk/azidentity/cache v0.3.0 h1:+m0M/LFxN43KvULkDNfdXOgrjtg6UYJPFBJyuEcRCAw=
-github.com/Azure/azure-sdk-for-go/sdk/azidentity/cache v0.3.0/go.mod h1:PwOyop78lveYMRs6oCxjiVyBdyCgIYH6XHIVZO9/SFQ=
-github.com/Azure/azure-sdk-for-go/sdk/data/azcosmos v1.0.3 h1:gBWC0dYF3aO+7xGxL0Ccjv9BmnV30C8VZIrUPlMct6g=
-github.com/Azure/azure-sdk-for-go/sdk/data/azcosmos v1.0.3/go.mod h1:7LBWaO4KRASAo9VpfhpxQKkdY6PBwkv9UDKzL9Sajuw=
-github.com/Azure/azure-sdk-for-go/sdk/data/aztables v1.2.0 h1:aJG+Jxd9/rrLwf8R1Ko0RlOBTJASs/lGQJ8b9AdlKTc=
-github.com/Azure/azure-sdk-for-go/sdk/data/aztables v1.2.0/go.mod h1:41ONblJrPxDcnVr+voS+3xXWy/KnZLh+7zY5s6woAlQ=
+github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.8.2 h1:F0gBpfdPLGsw+nsgk6aqqkZS1jiixa5WwFe3fk/T3Ys=
+github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.8.2/go.mod h1:SqINnQ9lVVdRlyC8cd1lCI0SdX4n2paeABd2K8ggfnE=
+github.com/Azure/azure-sdk-for-go/sdk/azidentity/cache v0.3.2 h1:yz1bePFlP5Vws5+8ez6T3HWXPmwOK7Yvq8QxDBD3SKY=
+github.com/Azure/azure-sdk-for-go/sdk/azidentity/cache v0.3.2/go.mod h1:Pa9ZNPuoNu/GztvBSKk9J1cDJW6vk/n0zLtV4mgd8N8=
+github.com/Azure/azure-sdk-for-go/sdk/data/azcosmos v1.3.0 h1:RGcdpSElvcXCwxydI0xzOBu1Gvp88OoiTGfbtO/z1m0=
+github.com/Azure/azure-sdk-for-go/sdk/data/azcosmos v1.3.0/go.mod h1:YwUyrNUtcZcibA99JcfCP6UUp95VVQKO2MJfBzgJDwA=
+github.com/Azure/azure-sdk-for-go/sdk/data/aztables v1.3.0 h1:NnE8y/opvxowwNcSNHubQUiSSEhfk3dmooLGAOmPuKs=
+github.com/Azure/azure-sdk-for-go/sdk/data/aztables v1.3.0/go.mod h1:GhHzPHiiHxZloo6WvKu9X7krmSAKTyGoIwoKMbrKTTA=
 github.com/Azure/azure-sdk-for-go/sdk/internal v0.7.0/go.mod h1:yqy467j36fJxcRV2TzfVZ1pCb5vxm4BtZPUdYWe/Xo8=
 github.com/Azure/azure-sdk-for-go/sdk/internal v1.10.0 h1:ywEEhmNahHBihViHepv3xPBn1663uRv2t2q/ESv9seY=
 github.com/Azure/azure-sdk-for-go/sdk/internal v1.10.0/go.mod h1:iZDifYGJTIgIIkYRNWPENUnqx6bJ2xnSDFI2tjwZNuY=
@@ -673,10 +673,10 @@ github.com/Azure/azure-sdk-for-go/sdk/keyvault/internal v0.7.1 h1:FbH3BbSb4bvGlu
 github.com/Azure/azure-sdk-for-go/sdk/keyvault/internal v0.7.1/go.mod h1:9V2j0jn9jDEkCkv8w/bKTNppX/d0FVA1ud77xCIP4KA=
 github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/storage/armstorage v1.6.0 h1:PiSrjRPpkQNjrM8H0WwKMnZUdu1RGMtd/LdGKUrOo+c=
 github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/storage/armstorage v1.6.0/go.mod h1:oDrbWx4ewMylP7xHivfgixbfGBT6APAwsSoHRKotnIc=
-github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.4.1 h1:cf+OIKbkmMHBaC3u78AXomweqM0oxQSgBXRZf3WH4yM=
-github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.4.1/go.mod h1:ap1dmS6vQKJxSMNiGJcq4QuUQkOynyD93gLw6MDF7ek=
-github.com/Azure/azure-sdk-for-go/sdk/storage/azdatalake v1.2.1 h1:qhYAuEKEz8jDGV1Tyf3Gz44ppGZP53xJGi4qdH2PLsc=
-github.com/Azure/azure-sdk-for-go/sdk/storage/azdatalake v1.2.1/go.mod h1:8QbU31D+n+SDJDstJXJWnOBKAgv8ZTz/x7su5KQ31Kk=
+github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.6.0 h1:UXT0o77lXQrikd1kgwIPQOUect7EoR/+sbP4wQKdzxM=
+github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.6.0/go.mod h1:cTvi54pg19DoT07ekoeMgE/taAwNtCShVeZqA+Iv2xI=
+github.com/Azure/azure-sdk-for-go/sdk/storage/azdatalake v1.4.0 h1:FVP7qKI1g9rcEgnxiDRmOzvI2l4ydNIYSRR/qMMFQdQ=
+github.com/Azure/azure-sdk-for-go/sdk/storage/azdatalake v1.4.0/go.mod h1:KkcFZGL0F/6ooKPl8Ub1EPtGOCVXBayeWuJ1IQomreA=
 github.com/Azure/azure-sdk-for-go/sdk/storage/azqueue v1.0.0 h1:lJwNFV+xYjHREUTHJKx/ZF6CJSt9znxmLw9DqSTvyRU=
 github.com/Azure/azure-sdk-for-go/sdk/storage/azqueue v1.0.0/go.mod h1:GfT0aGew8Qj5yiQVqOO5v7N8fanbJGyUoHqXg56qcVY=
 github.com/Azure/azure-storage-blob-go v0.14.0/go.mod h1:SMqIBi+SuiQH32bvyjngEewEeXoPfKMgWlBDaYf6fck=
@@ -692,8 +692,8 @@ github.com/Azure/go-autorest/logger v0.2.1/go.mod h1:T9E3cAhj2VqvPOtCYAvby9aBXkZ
 github.com/Azure/go-autorest/tracing v0.6.0/go.mod h1:+vhtPC754Xsa23ID7GlGsrdKBpUA79WCAKPPZVC2DeU=
 github.com/AzureAD/microsoft-authentication-extensions-for-go/cache v0.1.1 h1:WJTmL004Abzc5wDB5VtZG2PJk5ndYDgVacGqfirKxjM=
 github.com/AzureAD/microsoft-authentication-extensions-for-go/cache v0.1.1/go.mod h1:tCcJZ0uHAmvjsVYzEFivsRTN00oz5BEsRgQHu5JZ9WE=
-github.com/AzureAD/microsoft-authentication-library-for-go v1.2.2 h1:XHOnouVk1mxXfQidrMEnLlPk9UMeRtyBTnEFtxkV0kU=
-github.com/AzureAD/microsoft-authentication-library-for-go v1.2.2/go.mod h1:wP83P5OoQ5p6ip3ScPr0BAq0BvuPAvacpEuSzyouqAI=
+github.com/AzureAD/microsoft-authentication-library-for-go v1.3.3 h1:H5xDQaE3XowWfhZRUpnfC+rGZMEVoSiji+b+/HFAPU4=
+github.com/AzureAD/microsoft-authentication-library-for-go v1.3.3/go.mod h1:wP83P5OoQ5p6ip3ScPr0BAq0BvuPAvacpEuSzyouqAI=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/BurntSushi/toml v1.4.1-0.20240526193622-a339e1f7089c h1:pxW6RcqyfI9/kWtOwnv/G+AzdKuy2ZrqINhenH4HyNs=
 github.com/BurntSushi/toml v1.4.1-0.20240526193622-a339e1f7089c/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho=
diff --git a/internal/impl/aws/input_s3.go b/internal/impl/aws/input_s3.go
index d418a717e9..676ad5daf2 100644
--- a/internal/impl/aws/input_s3.go
+++ b/internal/impl/aws/input_s3.go
@@ -583,6 +583,7 @@ func (s *sqsTargetReader) readSQSEvents(ctx context.Context) ([]*s3ObjectTarget,
 						keyNotFound := false
 						if apiErr := smithy.APIError(nil); errors.As(err, &apiErr) {
 							if _, ok := apiErr.(*s3types.NoSuchKey); ok {
+								s.log.Warnf("Dropping SQS notification for missing key %q: %s", object.key, err)
 								keyNotFound = true
 							}
 						}
@@ -603,9 +604,6 @@ func (s *sqsTargetReader) readSQSEvents(ctx context.Context) ([]*s3ObjectTarget,
 						} else {
 							ackOnce.Do(func() {
 								if atomic.AddInt32(&pendingAcks, -1) == 0 {
-									if keyNotFound {
-										s.log.Warnf("Dropping SQS notification for missing key %q: %s", object.key, err)
-									}
 									aerr = s.ackSQSMessage(ctx, sqsMsg)
 								}
 							})
diff --git a/internal/impl/azure/input_blob_storage.go b/internal/impl/azure/input_blob_storage.go
index 8910b503a7..e68e16f075 100644
--- a/internal/impl/azure/input_blob_storage.go
+++ b/internal/impl/azure/input_blob_storage.go
@@ -27,6 +27,7 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/runtime"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/bloberror"
 	"github.com/Jeffail/gabs/v2"
 
 	"github.com/redpanda-data/benthos/v4/public/service"
@@ -299,7 +300,15 @@ func (a *azureTargetStreamReader) Pop(ctx context.Context) (*azureObjectTarget,
 			a.pending = append(a.pending, &azureObjectTarget{
 				key: name,
 				ackFn: func(ctx context.Context, err error) (aerr error) {
-					if err != nil {
+					keyNotFound := false
+					var rErr *azcore.ResponseError
+					if errors.As(err, &rErr) {
+						if rErr.ErrorCode == string(bloberror.BlobNotFound) {
+							a.log.Warnf("Skipping missing blob: %s", name)
+							keyNotFound = true
+						}
+					}
+					if err != nil && !keyNotFound {
 						nackOnce.Do(func() {
 							// Prevent future acks from triggering a delete.
 							atomic.StoreInt32(&pendingAcks, -1)


Test patch:
diff --git a/internal/impl/azure/integration_test.go b/internal/impl/azure/integration_test.go
index ba43ef53ab..f76006fa6d 100644
--- a/internal/impl/azure/integration_test.go
+++ b/internal/impl/azure/integration_test.go
@@ -271,7 +271,7 @@ input:
 		).Run(
 			t, template,
 			integration.StreamTestOptVarSet("VAR1", dummyQueue),
-			integration.StreamTestOptVarSet("VAR2", "UseDevelopmentStorage=true;"),
+			integration.StreamTestOptVarSet("VAR2", connString),
 		)
 	})
 }