snssqs: fix consumer starvation

The current implementation uses separate goroutines to process each
message retrieved by `sqsClient.ReceiveMessageWithContext`, which is
executed when `concurrencyMode` is set to `parallel`.
The `consumeSubscription` function waits for all spawned goroutines to
complete before receiving the next batch. However, this design has a
critical flaw: it starves the consumer if new messages are published
during the processing of an existing batch. These newly published
messages won't be processed until all goroutines from the original batch
have finished their work, effectively blocking the consumption of the
new messages.

This PR changes that by removing the waiting mechanism.

Signed-off-by: Gustavo Chain <[email protected]>

pubsub/aws/snssqs/snssqs.go

@@ -21,7 +21,6 @@ import (
 	"reflect"
 	"strconv"
 	"strings"
-	"sync"
 	"sync/atomic"
 	"time"
 
@@ -625,33 +624,27 @@ func (s *snsSqs) consumeSubscription(ctx context.Context, queueInfo, deadLetters
 		}
 		s.logger.Debugf("%v message(s) received on queue %s", len(messageResponse.Messages), queueInfo.arn)
 
-		var wg sync.WaitGroup
 		for _, message := range messageResponse.Messages {
 			if err := s.validateMessage(ctx, message, queueInfo, deadLettersQueueInfo); err != nil {
 				s.logger.Errorf("message is not valid for further processing by the handler. error is: %v", err)
 				continue
 			}
 
 			f := func(message *sqs.Message) {
-				defer wg.Done()
 				if err := s.callHandler(ctx, message, queueInfo); err != nil {
 					s.logger.Errorf("error while handling received message. error is: %v", err)
 				}
 			}
 
-			wg.Add(1)
 			switch s.metadata.ConcurrencyMode {
 			case pubsub.Single:
 				f(message)
 			case pubsub.Parallel:
-				wg.Add(1)
 				go func(message *sqs.Message) {
-					defer wg.Done()
 					f(message)
 				}(message)
 			}
 		}
-		wg.Wait()
 	}
 }