consumer: rework Consumer API #52

rogpeppe · 2019-10-23T16:37:49Z

We'd like to make it easier to have a more flexible strategy for
retrying message sending (e.g. retry only on some errors
or for a message-dependant time), so remove
the retry logic from felice, leaving it up to the caller
to determine.

This means that we need to pass the Context to the handler
(so that a retrying handler can know when to abort when the
context is done) and also enough information that appropriate
metrics can be sent.

Doing this means we can also simplify the configuration, because
we don't need to log or send metrics here any more (that can
be done in handler wrappers themselves).

We also a address couple of issues in the existing implementation:

there is currently a race between Serve and Stop - Serve needs to
be called in its own goroutine, so there's a race between setting
Consumer.consumer and Stop calling Consumer.consumer.Close.
Return.Errors is set to true, but there's nothing that reads from the errors channel, which makes that pointless.
All the other types use Close instead of Stop, so we rename Stop to Close accordingly.
Don't require the full config for creating a MessageConverterV1.

Fixes issue #50.

skateinmars · 2019-10-23T17:04:17Z

consumer/consumer.go

+		if err := h.handleMessage(ctx, msg, claim); err != nil {
+			if c.consumer.config.Discarded != nil {
+				c.consumer.config.Discarded(msg, err)
+			}


Doesn't that naming imply that the message will always get "rejected"? How would retries get handled for instance?

With the new handler API, the handler itself can handle retries - in my view retrying is more domain-specific and hence closer to kafka-go's domain than felice's. The plan is to add a RetryHandler type (essentially middleware for handlers) that will handle retries like the current logic does, but because it's just an arbitrary wrapper, it becomes easy to tweak the behaviour (for example to retry for a period depending on the message contents).

The default behaviour of kafka-go would be to use a RetryHandler (the API is planned to remain backwardly compatible). This is my current idea as to what the RetryHandler API might look like:

// RetryHandler implements Handler by retrying type RetryHandler struct { // Handler holds the handler that will be retried. Handler Handler // Strategy is used to determine how the handler // retries HandleMessage calls. If there are no more available // attempts, the last error returned from HandleMessage // will be returnefd Strategy func(m *Message) retry.Strategy // Logf is used to log a message when an attempt // to send the message failed. Logf func(format string, args ...interface{}) // IsFatal reports whether the given // error should be considered fatal. Fatal // errors will cause the retry loop to abort. IsFatal func(err error) bool } func (h RetryHandler) HandleMessage(ctx context.Context, m *Message) error

On reflection, I designed that a while ago, and if this logic lives in kafka-go, the Logf field is probably unnecessary.

Makes sense! 👍

(re: Strategy I would maybe pass both message and error to the strategy func and let it decide which error to return)

Yeah, there's bound to be room for changing things there. We'll probably look at some real scenarios and see how it fits. Also, the nice thing about this is it doesn't have to be one-size-fits-all - if you have a special requirement, you can always just roll your own handler or middleware.

yaziine

Thank you for that @rogpeppe, I left you few questions.

yaziine · 2019-10-24T08:02:07Z

consumer/consumer.go

-	c.handlers.Set(topic, HandlerConfig{
-		Handler:   h,
-		Converter: converter,
+	c.mu.Lock()


I don't see why do we need a mutex here. Handle will never be called concurrently IIUC.

Don't we call it within the go routines (one per parition)?

No. Handle is called to setup the handlers between creating the Consumer and calling Serve.

It's just for consistency - we document that the mutex guards the consumerGroup field, so it's good to use it for access to that, even if we don't think that Handle will ever be called concurrently.

yaziine · 2019-10-24T08:07:45Z

consumer/consumer.go

+// can achieve by calling Close. It returns an error only if Close
+// was called and the consumer shut down cleanly.
+//
+// Serve will return an error if it is called more than once,
+// the Consumer has been closed, or no handlers have
+// been registered.


I am not sure to understand well when Serve should return an error or not. I think that this comment can be improved.

improved comment

yaziine · 2019-10-24T12:24:29Z

consumer/handler.go

+func (h *topicHandler) handleMessage(
+	ctx context.Context,
+	sm *sarama.ConsumerMessage,
+	claim sarama.ConsumerGroupClaim,


In which case the sarama.ConsumerGroupClaim parameter is needed?

We use the claim so that the Message type can implement HighWaterMarkOffset.

tealeg

Nice work. A few small comments inline.

tealeg · 2019-10-24T14:33:13Z

consumer/config.go

-	// KafkaAddrs holds kafka brokers addresses. There must be at least
-	// one entry in the slice.
-	// The default value is "localhost:9092".
+	// KafkaAddrs holds the kafka broker addresses in host:port


🌷 Incidentally, I'd love to know if there's a proper name for "host:port" format. I know it's defined as part of URI, but it is used outside that context so often.

Me too. In previous places, I've tried be consistent and use "HostPort" and "Host" where appropriate, but URLs mess everything up by allowing both :)

tealeg · 2019-10-24T14:35:17Z

consumer/consumer.go

-	quit     chan struct{}
+	// consumerGroup holds the consumerGroup started by Serve.
+	consumerGroup sarama.ConsumerGroup
+	// handlers holds the set of handlers.


This comment isn't really adding much, IMHO.

tealeg · 2019-10-24T14:36:26Z

consumer/consumer.go

-	c.handlers.Set(topic, HandlerConfig{
-		Handler:   h,
-		Converter: converter,
+	c.mu.Lock()


Don't we call it within the go routines (one per parition)?

tealeg · 2019-10-24T14:39:53Z

consumer/consumer.go

+// Serve runs the consumer and listens for new messages on the given
+// topics. Serve will block until it is instructed to stop, which you
+// can achieve by calling Close. It returns an error only if Close
+// was called and the consumer shut down cleanly.


This second sentence here is somewhat surprising.

You mean you'd expect "Close" to be named "Stop" ?
You may be right in that case, but I generally think that Close is the name we use in Go for closing resources, and this is really just another resource to be cleaned up.

tealeg · 2019-10-24T14:45:17Z

consumer/doc.go

-//    h := handler.HandlerFunc(func(msg *consumer.Message) error {
-//        fmt.Printf("%+v", *msg)
+//    h := handler.HandlerFunc(func(ctx context.Context, m *Message) error {
+//        fmt.Printf("%+v", msg)
 //    })
 package consumer


As we're shifting responsibility for error handling and retry to then user now, perhaps we should add some indications of how that is achieved in the docs here.

We'd like to make it easier to have a more flexible strategy for retrying message sending (e.g. retry only on some errors or for a message-dependant time), so remove the retry logic from felice, leaving it up to the caller to determine. This means that we need to pass the Context to the handler (so that a retrying handler can know when to abort when the context is done) and also enough information that appropriate metrics can be sent. Doing this means we can also simplify the configuration, because we don't need to log or send metrics here any more (that can be done in handler wrappers themselves). We also a address couple of issues in the existing implementation: - there is currently a race between Serve and Stop - Serve needs to be called in its own goroutine, so there's a race between setting Consumer.consumer and Stop calling Consumer.consumer.Close. - Return.Errors is set to true, but there's nothing that reads from the errors channel, which makes that pointless. - All the other types use Close instead of Stop, so we rename Stop to Close accordingly. - Don't require the full config for creating a MessageConverterV1. Fixes issue #50.

rogpeppe requested review from yaziine, asdine and tealeg October 23, 2019 16:38

skateinmars reviewed Oct 23, 2019

View reviewed changes

yaziine approved these changes Oct 24, 2019

View reviewed changes

tealeg approved these changes Oct 24, 2019

View reviewed changes

rogpeppe force-pushed the rog-001-consumer-revamp branch from 2f73a59 to 1d0a529 Compare October 25, 2019 09:09

rogpeppe merged commit 86d4bbe into master Oct 25, 2019

rogpeppe mentioned this pull request Oct 25, 2019

API improvements #50

Closed

tealeg mentioned this pull request Oct 29, 2019

Lack of integration tests #48

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

consumer: rework Consumer API #52

consumer: rework Consumer API #52

rogpeppe commented Oct 23, 2019

skateinmars Oct 23, 2019

rogpeppe Oct 23, 2019 •

edited

Loading

skateinmars Oct 23, 2019

skateinmars Oct 23, 2019

rogpeppe Oct 23, 2019

yaziine left a comment

yaziine Oct 24, 2019

tealeg Oct 24, 2019

yaziine Oct 25, 2019

rogpeppe Oct 25, 2019

yaziine Oct 24, 2019

rogpeppe Oct 25, 2019

yaziine Oct 24, 2019

rogpeppe Oct 25, 2019

tealeg left a comment

tealeg Oct 24, 2019

rogpeppe Oct 24, 2019

tealeg Oct 24, 2019

rogpeppe Oct 25, 2019

tealeg Oct 24, 2019

tealeg Oct 24, 2019

rogpeppe Oct 24, 2019

tealeg Oct 24, 2019

consumer: rework Consumer API #52

consumer: rework Consumer API #52

Conversation

rogpeppe commented Oct 23, 2019

Choose a reason for hiding this comment

rogpeppe Oct 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

yaziine 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

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

tealeg 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

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

rogpeppe Oct 23, 2019 •

edited

Loading