alexeybogdan95 / sqspoller Goto Github PK

Please take a look at next test:
https://github.com/IhnatKlimchuk/SqsPoller/blob/playground/test/SqsPoller.Tests.Unit/ConsumerResolverTests.cs#L184

Task cancelled this way causes ContinueWith task see task.IsCancelled == true but still deleting message from sqs.
example of consumer https://github.com/IhnatKlimchuk/SqsPoller/blob/playground/test/SqsPoller.Tests.Unit/ThrowingMessageConsumer.cs#L14

same issue is reproduced while stopping host

This issue is pretty important as caused some data loss on prod envs.

Let us take following Json object:

 {
    "name":"test",
    "date":"2021-03-23T10:14:40.115+00:00"
 }

As you see, time is UTC, but if such object will be message body, it will be deserialized incorrectly. Its Kind property will have DateTimeKind.Local value. It seems to be easily fixed by adjusting Newtonsoft.Json settings. I reproduced same issue on test project with only Newtonsoft.Json serializing/deserializing

private async Task ProcessEvent(Message message, CancellationToken cancellationToken, SemaphoreSlim semaphore, string queueUrl)
        {
            try
            {
                await _consumerResolver.Resolve(message, cancellationToken);
                DeleteMessage(queueUrl, message, cancellationToken);
            }
            catch (OperationCanceledException e)
            {
                _logger.LogWarning(e,
                    "Failed to handle message with id {message_id} and ReceiptHandle {receipt_handle}. Task has been cancelled");
            }
            catch (Exception e)
            {
                _logger.LogError(e,
                    "Failed to handle message with id {message_id} and ReceiptHandle {receipt_handle}");
            }
            finally
            {
                semaphore.Release();
            }
        }

Error during handling can be resulted by race condition and/or similar issue. Issue will cause retry by SQS, but you can't avoid Error logs, otherwise message will be assumed "consumed" correctly.

Add ability for one of

• set log level on message handling error
• pass consumption result to avoid deleting message
• custom SqlPoller exception for "Warning" level similar to OperationCanceledException

Please see https://github.com/IhnatKlimchuk/SqsPoller/tree/playground/test/SqsPoller.LoadTest for more details.

SqsPoller tries to read all available messages. In case of huge amount of messages that leads to:

• OOM
• Timeouts on http calls if so exist in consumer (due to local queue LIFO in task scheduler, limited connection pool, bottleneck in consumer)
• unplanned degradation on other parts of service (api calls in case service has such)