Repository navigation
pcgroups: Add consume error handler and recover from terminal consumer errors - #106
Open
arminazimi wants to merge 1 commit into
Open
arminazimi wants to merge 1 commit into
arminazimi wants to merge 1 commit into
Conversation
…r errors ElasticConsume and StaticConsume started the underlying Consume() without an error handler, so when the member's consumer was terminated (e.g. deleted) the member silently stopped processing messages while still holding its place in the group, and the caller had no way to find out. Add a WithConsumeErrHandler option (variadic, backwards compatible) that receives the errors reported by the consumer. On a terminal error the member's control routine now reacts: an elastic member drops its consumer so the existing catch-all re-creates it and resumes consuming, a static member stops and reports the error on Done(). Fixes synadia-io#41 Signed-off-by: a.azimi <arminazimi4030@gmail.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Fixes #41
ElasticConsumeandStaticConsumecalledConsume()without an error handler. When the member's consumer was terminated (e.g. deleted), nats.go stopped the consume context, but the member kept its place in the group without processing anything, and the caller had no way to find out.Changes
WithConsumeErrHandler(func(err error))option, passed as a trailing variadic...ConsumeOptso existing callers keep compiling. It receives every error reported by the member's underlying consumer. A nil handler is rejected.ErrConsumerDeleted,ErrBadRequest, the ones after which nats.go stops the consume context) are now forwarded to the instance's control routine:Done().Behavior note
For the static group I chose to report through
Done()rather than re-creating the durable consumer, since a deleted static consumer means it was removed from outside. Happy to change this if you prefer re-creating it.Tests
New tests in
pcgroups/testdelete the member's consumer and check that the handler getsErrConsumerDeleted, that the static member reports it onDone(), and that the elastic member resumes consuming. The elastic test times out without the fix.go test -race ./...passes inpcgroups/test.