Refactor sqsReader.Receive: move queue ack waiting and message deleti…

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

@@ -21,7 +21,7 @@ func TestParquetDecoding(t *testing.T) {
 		name          string
 		file          string
 		contentType   string
-		numEvents     int
+		numEvents     uint64
 		assertAgainst string
 		config        *readerConfig
 	}{
@@ -121,7 +121,7 @@ func readJSONFromFile(t *testing.T, filepath string) []string {
 	var rawMessages []json.RawMessage
 	err = json.Unmarshal(fileBytes, &rawMessages)
 	assert.NoError(t, err)
-	var data []string
+	data := make([]string, 0, len(rawMessages))
 
 	for _, rawMsg := range rawMessages {
 		data = append(data, string(rawMsg))

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

@@ -157,7 +157,7 @@ func (in *s3Input) Run(inputContext v2.Context, pipeline beat.Pipeline) error {
 		// Create client for publishing events and receive notification of their ACKs.
 		client, err := pipeline.ConnectWith(beat.ClientConfig{
 			CloseRef:      inputContext.Cancelation,
-			EventListener: awscommon.NewEventACKHandler(),
+			EventListener: NewEventACKHandler(),
 			Processing: beat.ProcessingConfig{
 				// This input only produces events with basic types so normalization
 				// is not required.
@@ -227,8 +227,8 @@ func (in *s3Input) createSQSReceiver(ctx v2.Context, pipeline beat.Pipeline) (*s
 	}
 	in.metrics = newInputMetrics(ctx.ID, nil, in.config.MaxNumberOfMessages)
 	s3EventHandlerFactory := newS3ObjectProcessorFactory(log.Named("s3"), in.metrics, s3API, fileSelectors, in.config.BackupConfig, in.config.MaxNumberOfMessages)
-	sqsMessageHandler := newSQSS3EventProcessor(log.Named("sqs_s3_event"), in.metrics, sqsAPI, script, in.config.VisibilityTimeout, in.config.SQSMaxReceiveCount, pipeline, s3EventHandlerFactory, in.config.MaxNumberOfMessages)
-	sqsReader := newSQSReader(log.Named("sqs"), in.metrics, sqsAPI, in.config.MaxNumberOfMessages, sqsMessageHandler)
+	sqsMessageHandler := newSQSS3EventProcessor(log.Named("sqs_s3_event"), in.metrics, sqsAPI, script, in.config.VisibilityTimeout, in.config.SQSMaxReceiveCount, s3EventHandlerFactory)
+	sqsReader := newSQSReader(log.Named("sqs"), in.metrics, sqsAPI, in.config.MaxNumberOfMessages, sqsMessageHandler, pipeline)
 
 	return sqsReader, nil
 }