Add sequential flag to mqtrigger #1814

daniel-shuy · 2020-10-07T17:02:47Z

Adds a sequential flag to the mqtrigger function. If set to true, consumes messages sequentially instead of concurrently.

I wasn't sure whether to add support for all 3 MQs (Kafka, Azure Queue Storage, NATS), as they currently have different behaviors (Kafka and Azure Queue Storage MQs currently consumes messages concurrently, while NATS MQ consumes messages sequentially). Therefore I separated the implementation for each MQ into its own commit so that they can be dropped if we decide not to support them.

Note that if support for the sequential flag is added to the NATS MQ, it is a breaking change, as it would mean changing the default behavior from sequential to concurrent.

This change is

codecov · 2020-10-07T17:21:44Z

Codecov Report

Merging #1814 (514479c) into master (ec22e22) will decrease coverage by 7.24%.
The diff coverage is 33.33%.

❗ Current head 514479c differs from pull request most recent head 2a3f88c. Consider uploading reports for the commit 2a3f88c to get more accurate results

@@            Coverage Diff             @@
##           master    #1814      +/-   ##
==========================================
- Coverage   28.41%   21.17%   -7.25%     
==========================================
  Files          69       52      -17     
  Lines        5040     3708    -1332     
==========================================
- Hits         1432      785     -647     
+ Misses       3390     2850     -540     
+ Partials      218       73     -145

Impacted Files	Coverage Δ
pkg/mqtrigger/scalermanager.go	`26.66% <0.00%> (-0.24%)`	⬇️
...kg/mqtrigger/messageQueue/azurequeuestorage/asq.go	`54.59% <50.00%> (-0.70%)`	⬇️
pkg/crd/crdvalidations.go	`0.00% <0.00%> (-100.00%)`	⬇️
pkg/crd/crd.go	`0.00% <0.00%> (-93.24%)`	⬇️
pkg/crd/client.go	`0.00% <0.00%> (-32.00%)`	⬇️
pkg/controller/storagesvc.go
pkg/controller/controller.go
pkg/controller/httpTriggerApi.go
pkg/controller/api.go
pkg/controller/packageApi.go
... and 12 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 154fe0d...2a3f88c. Read the comment docs.

RealHarshThakur · 2020-10-19T16:59:31Z

pkg/mqtrigger/messageQueue/nats/nats.go

+
+ var resp *http.Response
+ for attempt := 0; attempt <= trigger.Spec.MaxRetries; attempt++ {
+ // Make the request


Hey @daniel-shuy , can you please set the headers inside the for loop? This is to avoid an error I encountered at other places: https://stackoverflow.com/questions/31337891/net-http-http-contentlength-222-with-body-length-0

@harshthakur9030 should this be in a separate PR? I'm actually just moving the entire code block into an anonymous function so that I can conditionally call it as a goroutine, I didn't change any of the existing logic

Yes, that bug existed before you touched the code base. Since you have already touched this part of the code base, I was hoping you could do it. Everything remains the same, just the for loop which sets the headers needs to go into the loop which makes the requests.

Ok sure, I can do it

RealHarshThakur · 2020-10-19T17:01:24Z

pkg/mqtrigger/messageQueue/nats/nats.go

+ }
+ if err == nil && resp.StatusCode == http.StatusOK {
+ // Success, quit retrying
+ break


Can you please change this to an appropriate return statement?

? Why? If we return here the rest of the function (closing the response body, checking if the response returned an error, trigger ack message, publishing to response topic) won't execute

Oh yes! My bad. One quick question though: Shouldn't we consider for status codes between 200 and 300 to be successful rather than just 200?

I don't think the router can return anything other than HTTP 200 for a successful request? Maybe @vishal-biyani can confirm

vishal-biyani · 2020-10-20T06:09:28Z

Hey @therahulbhati This will also impact all Keda connectors I believe?

daniel-shuy · 2020-10-20T07:42:03Z

@vishal-biyani I don't think so, don't Keda connectors have their own implementation (https://github.com/fission/keda-connectors)?

Looking at the implementation of the Keda Kafka HTTP Connector (https://github.com/fission/keda-connectors/tree/master/kafka-http-connector), I don't think it will have this issue as it is consuming messages sequentially, unlike the mqtrigger Kafka connector which is consuming messages concurrently.

vishal-biyani

Hi @daniel-shuy Thanks for this PR and looks great except for 2 really minor comments. Please let me know your thoughts and I will merge this one.

vishal-biyani · 2020-10-28T16:05:19Z

pkg/mqtrigger/messageQueue/kafka/kafka.go

+ msgHandler := func() { kafkaMsgHandler(&kafka, producer, trigger, msg, consumer) }
+ if trigger.Spec.Sequential {
+ msgHandler()
+ } else {
+ go msgHandler()
+ }


Won't we need a wg.Add & wg.Done similar to AzureQueue here?

vishal-biyani · 2020-10-28T16:07:43Z

pkg/mqtrigger/messageQueue/nats/nats.go

+ if trigger.Spec.Sequential {
+ cb()
+ } else {
+ go cb()


Just checking if wg.Add and wg.Done similar to AzureQueue would be needed here?

I took a look at the AzureQueue mqtrigger implementation, and seems like it is using sync.WaitGroup to notify AzureQueueSubscription.done (a chan bool) (see asq.go#L275).

The other mqtrigger implementations do not have anything equivalent, so it shouldn't be needed.

vishal-biyani · 2020-10-28T17:30:18Z

Thanks @daniel-shuy this looks good, I will merge in a day or so, again thanks for your effort on this one 🎉

chetanpdeshmukh · 2020-11-10T15:33:14Z

Hello @daniel-shuy ,
Thanks again for the PR!

Thank you for contributing to Fission project. Could you please fill out this form, so we can send you the well deserved awesome swags :)

Team Fission

daniel-shuy · 2020-11-10T15:36:16Z

@chetanpdeshmukh yay, thanks! The form is restricted though

chetanpdeshmukh · 2020-11-10T15:38:57Z

@chetanpdeshmukh yay, thanks! The form is restricted though

Hey @daniel-shuy can you give it a try now? Should work smoothly

daniel-shuy · 2020-11-10T15:42:02Z

@chetanpdeshmukh It works now, I've filled it up, thanks!

daniel-shuy · 2021-03-01T03:46:17Z

@vishal-biyani just noticed that this PR hasn't been merged!

AnatoliyYakimov · 2021-06-30T12:54:10Z

Hello, @daniel-shuy! Nice work you did! Any updates on this pull request?
We would like to use fission in our project, and this feature would help us a lot to keep consistency of our system.

daniel-shuy · 2021-06-30T15:37:04Z

@AnatoliyYakimov Unfortunately I don't have permission to merge, I'm also wondering why the PR hasn't been merged after so long 😅

AnatoliyYakimov · 2021-06-30T15:43:36Z

Maybe they just forgot)
@vishal-biyani please merge this feature if it's ready. We would be happy to this it in next release of Fission!)

daniel-shuy · 2021-06-30T15:48:54Z

I've resolved the conflicts with the latest master branch

AnatoliyYakimov · 2021-06-30T15:52:24Z

Also looking at the code, i saw that we continue to consume messages from topic if lambda returned an error. So if lambda returns an error, we still lose order, if'd i get it right.
For cases where ordering strictly required it would be better to have an option that allows you to stop processing entierly when facing an error.
What do you think about it?

And some personal question: im not very strong in go, and don't know Fission code that deep, so i don't understend what happens if after polling some messages weren't commited. Does sacrama (go kafka lib, if i spelled it right) return this messages from cache, commit all polled messages anyway or it will try to poll again from previous commit?

daniel-shuy · 2021-06-30T16:03:56Z

@AnatoliyYakimov Oh wow, you're right, didn't think of that. We should break out of the loop on error and repoll Kafka again (since the failed message is not committed, it will be polled again), good catch!

AnatoliyYakimov · 2021-06-30T16:12:46Z

If error is not transient and we will poll again and again, we will DDOS our lambdas. So maybe we need some kind of error retry count and then stop trigger. I think in Kubeless you can set retry count, before sending to DLQ. Maybe spec could look like this:
onError: SendToErrorTopic | StopProcessing
retryCount: 5
SendToErrorTopic and StopProcessing are error handling strategies. They can implement abstract interface, so in this case we could easily create new strategies for our mqttriggers. Also, if we abstract away from MQ implementation, we can reuse these strategies for any availible MQ.

daniel-shuy · 2021-06-30T16:16:53Z

@AnatoliyYakimov actually, let me get back to you on this. sarama is returning a channel, which is looped over sequentially. I'll need to look into sarama's code to see if its caching messages

daniel-shuy mentioned this pull request Oct 7, 2020

Sequential function invokations when using Kafka triggers #1569

Open

daniel-shuy force-pushed the mqtrigger-sequential branch from f9ada3e to 514479c Compare October 11, 2020 15:17

RealHarshThakur reviewed Oct 19, 2020

View reviewed changes

vishal-biyani requested changes Oct 28, 2020

View reviewed changes

daniel-shuy requested a review from vishal-biyani October 28, 2020 16:20

vishal-biyani approved these changes Oct 28, 2020

View reviewed changes

vishal-biyani added the hacktoberfest-accepted label Nov 5, 2020

daniel-shuy added 5 commits June 30, 2021 23:47

Add sequential flag to mqtrigger command

875ea8c

Add support for sequential flag to kafka message queue

d9634e5

Add support for sequential flag to Azure Queue Storage message queue

415f4b8

Add support for sequential flag to NATS message queue

71a567f

Set request headers in retry loop for NATS mqtrigger connector

2a3f88c

daniel-shuy force-pushed the mqtrigger-sequential branch from 469d8b7 to 2a3f88c Compare June 30, 2021 15:48

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add sequential flag to mqtrigger #1814

Add sequential flag to mqtrigger #1814

daniel-shuy commented Oct 7, 2020 •

edited by soamvasani

codecov bot commented Oct 7, 2020 •

edited

RealHarshThakur Oct 19, 2020

daniel-shuy Oct 20, 2020

RealHarshThakur Oct 20, 2020

daniel-shuy Oct 20, 2020

daniel-shuy Oct 20, 2020

RealHarshThakur Oct 20, 2020

RealHarshThakur Oct 19, 2020

daniel-shuy Oct 20, 2020

RealHarshThakur Oct 20, 2020

daniel-shuy Oct 20, 2020

vishal-biyani commented Oct 20, 2020

daniel-shuy commented Oct 20, 2020

vishal-biyani left a comment

vishal-biyani Oct 28, 2020

vishal-biyani Oct 28, 2020

daniel-shuy Oct 28, 2020

vishal-biyani commented Oct 28, 2020

chetanpdeshmukh commented Nov 10, 2020

daniel-shuy commented Nov 10, 2020

chetanpdeshmukh commented Nov 10, 2020

daniel-shuy commented Nov 10, 2020

daniel-shuy commented Mar 1, 2021

AnatoliyYakimov commented Jun 30, 2021

daniel-shuy commented Jun 30, 2021

AnatoliyYakimov commented Jun 30, 2021

daniel-shuy commented Jun 30, 2021

AnatoliyYakimov commented Jun 30, 2021

daniel-shuy commented Jun 30, 2021 •

edited

AnatoliyYakimov commented Jun 30, 2021

daniel-shuy commented Jun 30, 2021

Add sequential flag to mqtrigger #1814

Are you sure you want to change the base?

Add sequential flag to mqtrigger #1814

Conversation

daniel-shuy commented Oct 7, 2020 • edited by soamvasani

codecov bot commented Oct 7, 2020 • edited

Codecov Report

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

vishal-biyani commented Oct 20, 2020

daniel-shuy commented Oct 20, 2020

vishal-biyani left a 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

vishal-biyani commented Oct 28, 2020

chetanpdeshmukh commented Nov 10, 2020

daniel-shuy commented Nov 10, 2020

chetanpdeshmukh commented Nov 10, 2020

daniel-shuy commented Nov 10, 2020

daniel-shuy commented Mar 1, 2021

AnatoliyYakimov commented Jun 30, 2021

daniel-shuy commented Jun 30, 2021

AnatoliyYakimov commented Jun 30, 2021

daniel-shuy commented Jun 30, 2021

AnatoliyYakimov commented Jun 30, 2021

daniel-shuy commented Jun 30, 2021 • edited

AnatoliyYakimov commented Jun 30, 2021

daniel-shuy commented Jun 30, 2021

daniel-shuy commented Oct 7, 2020 •

edited by soamvasani

codecov bot commented Oct 7, 2020 •

edited

daniel-shuy commented Jun 30, 2021 •

edited