bugfix: deadlock; flush delete buffer after stopping the subscriber #68

muharem · 2019-06-27T15:49:59Z

few fixes related to aws subscriber:

deadlock occurs if the subscriber is stopped and received messages from aws waiting for a space in output channel, at that a consumer does not read from the output channel as it has stopped the subscriber.
removed messages are not flushed from the buffer after the subscriber is stopped and as a result handled and removed messages are not removed from sqs queue.
err var is not set to nil after each delete iteration.
stop all goroutine when the subscriber is being stopped
do not add a removed message to delete buffer if the subscriber is not running

jsafoodpanda · 2019-06-28T10:24:18Z

@muharem What happens if multiple messages are received but not consumed?
So we get 10 messages from SQS, process 5 and then die. The last 5 should not be deleted. This PR is about the first 5 that were processed and then added to the deletion queue?

muharem · 2019-06-28T10:44:55Z

@muharem What happens if multiple messages are received but not consumed?
So we get 10 messages from SQS, process 5 and then die. The last 5 should not be deleted. This PR is about the first 5 that were processed and then added to the deletion queue?

@jsafoodpanda Yes, only about those 5, which were explicitly marked as Done and putted to the delete buffer.

Now the subscriber just stops without flushing the delete buffer.

wilsont · 2019-06-28T18:37:53Z

LGTM, can you also provide unit test for this?

muharem · 2019-07-01T16:24:05Z

LGTM, can you also provide unit test for this?

@wilsont done

muharem · 2019-07-02T07:37:17Z

pubsub/awssub/sub.go

@@ -65,6 +66,8 @@ var (
 	defaultSQSConsumeBase64 = true
 )

+var sqsClientFactoryFunc = createSqsClient


helps to mock sqs client and use an actual constructor to create the subscriber for unittest

muharem · 2019-07-02T07:40:06Z

pubsub/awssub/sub.go

+		ReceiptHandle: m.message.ReceiptHandle,
+	}
+	if m.sub.isStopped() {
+		return m.sub.deleteMessageBatch(batchInput)


if the subscriber is stopped there is no goroutine to remove a message, remove it inside of the same gogoutine

muharem · 2019-07-02T07:42:28Z

pubsub/awssub/sub.go

+	atomic.SwapUint32(&s.stopped, uint32(0))
+	s.stop = make(chan chan error, 1)
+	s.flush = make(chan chan error, 1)
+	s.toDelete = make(chan *deleteRequest)


initializing all channels inside of Start func for a multiple run via one subscriber instance

muharem · 2019-07-02T07:43:29Z

pubsub/awssub/sub.go

@@ -224,49 +251,54 @@ func (s *subscriber) Start() <-chan pubsub.Message {
 			s.Logger.Printf("found %d messages", len(resp.Messages))
 			// for each message, pass to output
 			for _, msg := range resp.Messages {
-				output <- &subscriberMessage{
+				select {


to avoid deadlock when a consumer stop the subscriber and does not reading from output channel anymore

muharem · 2019-07-02T07:44:30Z

pubsub/awssub/sub.go

+		var delRequest *deleteRequest
+		var err error
+		select {
+		case flush := <-s.flush:


flush and shutdown the goroutine

dh-saurabh

Please add the explicit test cases for:

Deadlock scenario you fixed
Test assurance for unprocessed message can be recovered from the queue if the subscriber is stopped in between

dh-saurabh · 2019-07-23T09:04:26Z

pubsub/awssub/sub.go

@@ -168,12 +180,16 @@ func (m *subscriberMessage) ExtendDoneDeadline(d time.Duration) error {
 // message has been deleted.


does this description still holds good ?

The information that a message will be removed synchronically in case when the subscriber is stopped could be add but in my opinion for the client it does not make any sense.
The doc should say that message will be deleted be this function.
And SQSDeleteBufferSize configuration's doc should say to a client about delete buffer.

dh-saurabh · 2019-07-23T09:25:37Z

pubsub/awssub/sub.go

-		_, err = s.sqs.DeleteMessageBatch(batchInput)
-		delRequest.receipt <- err
+}
+


Also please add suitable comments describing what this method is doing

dh-saurabh · 2019-07-23T09:40:41Z

pubsub/awssub/sub_test.go

+	msq1.Done()
+	assert.True(t, len(sqstest.Deleted) == 0, "Message unexpectedly was removed from the delete buffer")
+
+	sub.Stop()


so this scenario mimic the case when you have three messages to process and you only processed one and stopped the subscriber and then you are flushing all the remaining messages without processing them. What happens when you restart the subscriber ? Will you get the remaining two messages ?

Can you add the tests which also confirms that you will not lose any unprocessed messages from queue

What happens when you restart the subscriber ?

Is there a use case for this? The app should not restart consumption after stoping as the only stop scenario is shutting down.

What do you mean by not lose any unprocessed messages? If they are not processed, they should remain in the SQS and not deleted.

What happens when you restart the subscriber ?

Is there a use case for this? The app should not restart consumption after stoping as the only stop scenario is shutting down.

Sure

What do you mean by not lose any unprocessed messages? If they are not processed, they should remain in the SQS and not deleted.

Please ignore this one

Will you get the remaining two messages ?

Yes

you are flushing all the remaining messages without processing them.

We are flushing only those which were marked as Done by Done method

This test case tests the deadlock
The subscriber gets 3 messages by one batch but the consumer reads from the channel only one and stop.

You can use this test case for the current version and see the deadlock

muharem · 2019-07-24T07:02:48Z

Please add the explicit test cases for:

Deadlock scenario you fixed

Already there

Test assurance for unprocessed message can be recovered from the queue if the subscriber is stopped in between

I do not get.
unprocessed message - unprocessed messages (the messages which were fetched from the queue but did not marked as Done by Done method) will be recovered to the queue by sqs functionality.

the subscriber is stopped in between - between what?

dh-saurabh · 2019-07-24T11:13:49Z

the subscriber is stopped in between - between what?

It means the subscriber starts and stopped during whatever process you were using subscriber.

dh-saurabh · 2019-07-24T11:14:29Z

unprocessed message - unprocessed messages (the messages which were fetched from the queue but did not marked as Done by Done method) will be recovered to the queue by sqs functionality.

Fair enough

dh-saurabh

As @muharem confirmed that this PR is only for removing the already processed and marked done messages from the SQS and required tests cases are there approved from my side.

bugfix flush delete buffer after stopping the subscriber

37e755c

Muharem Ismailov added 2 commits July 1, 2019 18:14

rerun subscriber; remove a msg after subscriber is stopped

417d270

unit test for aws subscriber

b4c4b17

fix aws subscriber deadlock

effc228

muharem commented Jul 2, 2019

View reviewed changes

muharem changed the title ~~bugfix flush delete buffer after stopping the subscriber~~ bugfix: deadlockj; flush delete buffer after stopping the subscriber Jul 2, 2019

muharem changed the title ~~bugfix: deadlockj; flush delete buffer after stopping the subscriber~~ bugfix: deadlock; flush delete buffer after stopping the subscriber Jul 2, 2019

jsafoodpanda approved these changes Jul 16, 2019

View reviewed changes

dh-saurabh requested changes Jul 23, 2019

View reviewed changes

godoc for awssub.deleteMessageBatch

0df11a3

dh-saurabh approved these changes Jul 24, 2019

View reviewed changes

wilsont approved these changes Jul 25, 2019

View reviewed changes

muharem merged commit 7d6318b into master Jul 31, 2019

Maxibond deleted the bugfix-remove-messages-from-delete-buffer branch December 4, 2019 04:45

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

bugfix: deadlock; flush delete buffer after stopping the subscriber #68

bugfix: deadlock; flush delete buffer after stopping the subscriber #68

muharem commented Jun 27, 2019 •

edited

Loading

jsafoodpanda commented Jun 28, 2019

muharem commented Jun 28, 2019

wilsont commented Jun 28, 2019

muharem commented Jul 1, 2019 •

edited

Loading

muharem Jul 2, 2019

muharem Jul 2, 2019

muharem Jul 2, 2019

muharem Jul 2, 2019

muharem Jul 2, 2019

dh-saurabh left a comment

dh-saurabh Jul 23, 2019 •

edited

Loading

muharem Jul 24, 2019

dh-saurabh Jul 23, 2019

muharem Jul 24, 2019

dh-saurabh Jul 23, 2019

dh-saurabh Jul 23, 2019

jsafoodpanda Jul 23, 2019

dh-saurabh Jul 23, 2019

dh-saurabh Jul 23, 2019

muharem Jul 24, 2019

muharem Jul 24, 2019

muharem commented Jul 24, 2019

dh-saurabh commented Jul 24, 2019

dh-saurabh commented Jul 24, 2019

dh-saurabh left a comment •

edited

Loading

		@@ -168,12 +180,16 @@ func (m *subscriberMessage) ExtendDoneDeadline(d time.Duration) error {
		// message has been deleted.

bugfix: deadlock; flush delete buffer after stopping the subscriber #68

bugfix: deadlock; flush delete buffer after stopping the subscriber #68

Conversation

muharem commented Jun 27, 2019 • edited Loading

jsafoodpanda commented Jun 28, 2019

muharem commented Jun 28, 2019

wilsont commented Jun 28, 2019

muharem commented Jul 1, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dh-saurabh left a comment

Choose a reason for hiding this comment

dh-saurabh Jul 23, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

muharem commented Jul 24, 2019

dh-saurabh commented Jul 24, 2019

dh-saurabh commented Jul 24, 2019

dh-saurabh left a comment • edited Loading

Choose a reason for hiding this comment

muharem commented Jun 27, 2019 •

edited

Loading

muharem commented Jul 1, 2019 •

edited

Loading

dh-saurabh Jul 23, 2019 •

edited

Loading

dh-saurabh left a comment •

edited

Loading