Task Supervisor for message consumers #179

akoutmos · 2020-04-26T18:56:15Z

This PR is an initial attempt at addressing issue #46. Still in the process of adding tests and solidifying the implementation, so leaving the PR in draft until complete (also looking for implementation feedback).

Description

Originally, the GenRMQ.Consumer module would call spawn/1 to asynchronously execute a message handler. This change instead leverages Task.Supervisor to execute message processing handlers. This should be a transparent change to the end user as this does not impact how GenRMQ.Consumer concurrently executes message handlers.

This introduces the usage of OTP 21's handle_continue to ensure that the Task.Supervisor is started before any mailbox messages are processed. As a result, gen_rmq would no longer be compatible with OTP releases prior to 21.

Checklist

I have added unit tests to cover my changes.
I have improved the code quality of this repo. (refactoring, or reduced number of static analyser issues)
I have updated the documentation accordingly

coveralls · 2020-04-26T19:00:38Z

Coverage increased (+0.7%) to 90.333% when pulling 72d45ee on task_supervisor_consumer into 73954a1 on master.

coveralls · 2020-04-26T19:00:39Z

Coverage increased (+0.2%) to 89.773% when pulling ccc0367 on task_supervisor_consumer into e120328 on master.

.travis.yml

lib/consumer.ex

mkorszun · 2020-04-27T11:13:30Z

@akoutmos looking good - left some questions.

…act on running tasks

akoutmos · 2020-05-04T19:32:24Z

I added some preliminary tests for this functionality, updated the telemetry documentation (as there is now a telemetry event for failed tasks), and also refactored the consumer a bit to be able to act upon Tasks that it has spawned under the supervisor. There are still a few more edge cases that I want to cover in the tests, so I will keep this in draft until I feel as though I have covered those code paths.

akoutmos · 2020-05-13T20:12:47Z

I think this one is at a good point to review now. I added the ability for consumer processes to wait on in flight tasks under the Task.Supervisor before terminating and also added tests to validate the flow. I also added an additional telemetry event to report on when Tasks fail.

One question I have is if we should expose an optional configuration for how long to wait on running Tasks. Currently I am just using the default value that is set for yield_many https://hexdocs.pm/elixir/Task.html#yield_many/2

vorce · 2020-05-14T16:07:23Z

@akoutmos I think yes we have to let users chose the task timeout. I suspect users will have very different expectations on their message processor, depending on their business logic.

akoutmos · 2020-05-14T16:16:29Z

I'll add that tonight and leave the default as the yield_many value.

spier · 2020-05-14T16:19:52Z

@akoutmos can I just say: You are awesome :) Great contributions in this PR (and elsewhere). Thanks so much!

akoutmos · 2020-05-14T16:35:22Z

Thanks for the kind words @spier! I appreciate it :)

akoutmos · 2020-05-15T01:14:07Z

Went ahead and added the terminate_timeout option for users to configure how long to wait on in-flight Tasks. I also updated the docs to let them know about the option.

mkorszun · 2020-05-15T07:56:45Z

@akoutmos amazing job 🥇- I will reserve some time during the weekend to review it.

vorce · 2020-05-15T07:54:57Z

documentation/guides/consumer/telemetry_events.md

@@ -42,4 +42,9 @@ GenRMQ emits [Telemetry][telemetry] events for consumers. It currently exposes t
  - Measurement: `%{time: System.monotonic_time}`
  - Metadata: `%{module: atom, reason: atom}`

+- `[:gen_rmq, :consumer, :task, :error]` - Dispatched by a GenRMQ consumer when a supervised Task fails to process a message


vorce · 2020-05-15T07:55:10Z

lib/consumer.ex

-    {:ok, state}
+  @doc false
+  @impl GenServer
+  def handle_continue(:init, state) do


vorce · 2020-05-15T08:00:18Z

lib/consumer.ex

+        {:DOWN, ref, :process, _pid, reason},
+        %{module: module, config: config, running_tasks: running_tasks} = state
+      ) do
+    if Map.has_key?(running_tasks, ref) do


What happens in this case to the message? Will it be rejected/nacked?

Good call....we would probably want some sort of configuration so the user can decide what happens to the message if the task fails. Perhaps we also provide the option to send the message to the deadletter exchange. Also on that same note, we may also want some configuration around how long the Tasks take to complete. Currently it is set to the async_nolink value of 5 seconds. Thoughts?

Maybe you should be able to set a callback to deal with the message? But I think the sensible default here would be to nack the message.

I like the callback, smth like: handle_error(reason, state). Then in this callback user has all the power to ack, reject or requeue.

btw, are we capturing here also task timeouts?

My latest commit adds the ability to configure a timeout for tasks. It was a little tricky to add that in given that Task.Supervisor does not provide that functionality for async_nolink, but I think I have a good implementation in place to handle this functionality. I misspoke earlier in regards to the 5 second timeout...that is for async_stream and async_stream_nolink.

As for a custom callback to handle the failure, are you thinking about rewriting the consumer module to a macro so that we can leverage defoverridable if the user does not want to use the default error handler? Or would handle_error be a required behaviour callback that needs to be implemented by the user?

@akoutmos I would prefer to avoid rewriting consumer module to a macro at this stage (just to limit the scope of changes).

Since these changes will be released as a major version, requiring error callback to be implemented should be fine, right? @vorce what do you think?

We could also consider skipping error callback and just reject the messages. The problem here is that users might skip dead-letter configuration for their consumers.

@mkorszun @akoutmos Hm hm. Yeah I think making each consumer implement a callback is OK. At least then we encourage users to think about this case. We can have a good implementation in the example consumer that logs and nacks the message.

Sounds good. I'll add an additional callback handle_error that will be triggered on task exception or task timeout along with an example or 2. I agree that this is probably the way to go so that gen_rmq does not impose any assumptions upon the user. They can deal with the error the way that best fits their needs.

Awesome, and thanks again for all the work @akoutmos

test/gen_rmq_consumer_test.exs

lib/consumer.ex

README.md

mkorszun

@akoutmos once again looks great. Thanks for the amazing work 👏 I agree with @vorce that it would be good to notify users whenever task fails (with default behaviour to dead-letter the message).

akoutmos · 2020-05-17T15:58:01Z

Another thing that I thought about today was that there is no way to throttle the consumer. With how it currently is, it can keep on consuming messages and spawning new tasks under the supervisor. Thoughts on adding an additional configuration to have the user set the number of concurrent tasks running? max_concurrent_tasks ?

vorce · 2020-05-17T18:43:40Z

@akoutmos good point! I think we can introduce that later if it turns out to be a problem for some users. The current/master implementation has the same problem.

akoutmos · 2020-05-23T02:49:43Z

I added the handle_error callback for consumers that are configured to execute via Tasks. I'll need to also hook it up to trigger the handle_error callback when running in synchronous mode. But wanted to get some feedback on the approach before adding that in as well.

mkorszun · 2020-05-23T18:34:56Z

lib/consumer.ex

+  `reason` - atom denoting the type of error
+
+  ## Examples:
+  To reject the message message that caused the Task to fail you can do something like so:


double message

mkorszun · 2020-05-23T18:37:17Z

lib/message_task.ex

@@ -0,0 +1,25 @@
+defmodule GenRMQ.MessageTask do


Good idea to wrap it like this 👍

mkorszun · 2020-05-23T18:40:07Z

@akoutmos error handling looks good 👍

…sage handling

akoutmos · 2020-05-25T03:57:16Z

I wrapped up that last of the tests and added handle_error callback support for non concurrent consumers. I'll give this a final look over tomorrow...but I think from a functionality perspective there is nothing else on my todo list. Let me know if you want to see any more tests/features :)

mkorszun · 2020-05-25T07:32:50Z

@akoutmos great, thank you 🙏 I plan to merge it today/tomorrow.

akoutmos · 2020-05-25T17:58:55Z

Updated some docs and some minor error handling tweaks. Nothing else pops out at me for clean up. All good on this side

spier · 2020-05-25T19:36:41Z

🎉

spier · 2020-05-25T19:48:01Z

@mkorszun just curious, what is the release process for the library? I remember you mentioning that we would have to release a new major(?) version for this?

mkorszun · 2020-05-25T20:15:25Z

@spier we need a major release because we have dropped support for some older versions of the OTP. Release process will involve:

bumping lib version to 3.0.0
documenting differences / migration steps between v2 and v3
merging all above changes to master and doing a github release
publishing to hex.pm

spier · 2020-05-26T07:14:36Z

@mkorszun Thanks for the writeup.

I am going through the steps, to understand them better. I might write a new issue to document the process for ourselves, and additional admins in the future. I will try to review what you do for this 3.0.0 release and then try myself on writing this up if that is ok?

Notes

bumping lib version to 3.0.0 => at least in mix.exs and README.md.
documenting differences / migration steps between v2 and v3 => probably in the wiki? I looked for migrations steps from v1 to v2 as a sample. maybe we did not write those then?
merging all above changes to master and doing a github release => releases
publishing to hex.pm => https://hex.pm/packages/gen_rmq

(unrelated) 30k party

Btw I realized that we are at 28173 downloads all time and currently at 803 downloads last 7 days. So we should hit the 30k downloads milestone in about 2.5 weeks. How does one throw a virtual party? :)

mkorszun · 2020-05-26T08:00:47Z

@spier sounds good. Your notes are already quite precise :)

regarding 30k party - I believe we need it 🎉

Initial attempt at using a Task Supervisor for message consumers

ccc0367

Adjusting testing travis matrix

994625f

akoutmos changed the title ~~Initial attempt at using a Task Supervisor for message consumers~~ Task Supervisor for message consumers Apr 26, 2020

mkorszun reviewed Apr 27, 2020

View reviewed changes

.travis.yml Show resolved Hide resolved

mkorszun reviewed Apr 27, 2020

View reviewed changes

lib/consumer.ex Outdated Show resolved Hide resolved

Remove unnecessary task supervisor start

c360540

akoutmos mentioned this pull request May 4, 2020

Bump amqp from 1.2.1 to 1.4.2 #182

Merged

Added telemetry event for failed task and adjusted :DOWN function to …

db0b2aa

…act on running tasks

akoutmos added 4 commits May 12, 2020 23:39

Added ability to yield for running tasks during GenServer termination

d5698e5

Added tests for Task Supervisor await

fb685b8

Fixing credo error

2f3313e

Fixing lingering process mailbox test error

0996fb5

akoutmos marked this pull request as ready for review May 13, 2020 19:42

akoutmos requested review from Shemeikka and vorce as code owners May 13, 2020 19:42

akoutmos added 2 commits May 13, 2020 15:48

Merge branch 'master' into task_supervisor_consumer

ccb0896

Fixing documentation

4eb7ef1

akoutmos added 2 commits May 14, 2020 20:59

Added ability to configure termination timeout

9ce9184

Minor documentation tweaks

7d016ca

vorce reviewed May 15, 2020

View reviewed changes

mkorszun reviewed May 15, 2020

View reviewed changes

lib/consumer.ex Show resolved Hide resolved

Make error tests clearer

6a28687

mkorszun reviewed May 16, 2020

View reviewed changes

README.md Outdated Show resolved Hide resolved

mkorszun approved these changes May 16, 2020

View reviewed changes

Fixed README docs

59b3e91

Shemeikka approved these changes May 17, 2020

View reviewed changes

akoutmos added 2 commits May 19, 2020 23:19

Added ability to timeout slow message handler tasks

0b8298c

Added handle_error callback along with tests

073078c

mkorszun reviewed May 23, 2020

View reviewed changes

lib/message_task.ex

@@ -0,0 +1,25 @@

defmodule GenRMQ.MessageTask do

Copy link

Collaborator

mkorszun May 23, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good idea to wrap it like this 👍

akoutmos added 2 commits May 24, 2020 21:51

Fixed flaky tests and added exception handling for non-concurrent mes…

bc2397a

…sage handling

Added test for handle_error during syncronoush handle_message

e14bfde

Enhanced docs and tweaked synchronous handle_message error case

72d45ee

mkorszun merged commit bbe426b into master May 25, 2020

mkorszun deleted the task_supervisor_consumer branch May 25, 2020 19:19

Task Supervisor for message consumers #179

Task Supervisor for message consumers #179

Conversation

akoutmos commented Apr 26, 2020 • edited Loading

Description

Checklist

coveralls commented Apr 26, 2020 • edited Loading

coveralls commented Apr 26, 2020

mkorszun commented Apr 27, 2020

akoutmos commented May 4, 2020 • edited Loading

akoutmos commented May 13, 2020 • edited Loading

vorce commented May 14, 2020 • edited Loading

akoutmos commented May 14, 2020

spier commented May 14, 2020

akoutmos commented May 14, 2020

akoutmos commented May 15, 2020

mkorszun commented May 15, 2020

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

akoutmos May 20, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vorce May 20, 2020 • edited Loading

Choose a reason for hiding this comment

akoutmos May 20, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mkorszun left a comment

Choose a reason for hiding this comment

akoutmos commented May 17, 2020

vorce commented May 17, 2020

akoutmos commented May 23, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mkorszun commented May 23, 2020

akoutmos commented May 25, 2020

mkorszun commented May 25, 2020

akoutmos commented May 25, 2020

spier commented May 25, 2020

spier commented May 25, 2020

mkorszun commented May 25, 2020 • edited Loading

spier commented May 26, 2020

Notes

(unrelated) 30k party

mkorszun commented May 26, 2020

akoutmos commented Apr 26, 2020 •

edited

Loading

coveralls commented Apr 26, 2020 •

edited

Loading

akoutmos commented May 4, 2020 •

edited

Loading

akoutmos commented May 13, 2020 •

edited

Loading

vorce commented May 14, 2020 •

edited

Loading

akoutmos May 20, 2020 •

edited

Loading

vorce May 20, 2020 •

edited

Loading

akoutmos May 20, 2020 •

edited

Loading

mkorszun commented May 25, 2020 •

edited

Loading