Added Consumer.OffsetsStore() #72
Conversation
|
@confluentinc It looks like @ctrochalakis just signed our Contributor License Agreement. 👍 Always at your service, clabot |
ctrochalakis
force-pushed
the
offsets-store
branch
from
5729ae5
to
a190ca1
Compare
Expose librdkafka's `rd_kafka_offsets_store()` introduced in v0.11.0. Consumer.OffsetsStore() allows manual control over the committed offsets without sacrificing auto-commit functionality.
ctrochalakis
force-pushed
the
offsets-store
branch
from
a190ca1
to
ccda146
Compare
|
Thank you for this, will tend to it shortly. |
kafka/consumer.go
Outdated
| // Returns the stored offsets on success. If at least one offset couldn't be stored, | ||
| // an error and a list of offsets is returned. Each offset can be checked for | ||
| // specific errors via its `.Error` member. | ||
| func (c *Consumer) OffsetsStore(offsets []TopicPartition) (storedOffsets []TopicPartition, err error) { |
There was a problem hiding this comment.
Should probably be called StoreOffsets to be consistent with CommitOffsets (even though it makes it inconsistent with librdkafka, but that's a hidden gem).
kafka/consumer_test.go
Outdated
| @@ -54,7 +54,6 @@ func TestConsumerAPIs(t *testing.T) { | |||
| if err != nil { | |||
| t.Errorf("SubscribeTopics failed: %s", err) | |||
| } | |||
|
|
|||
kafka/consumer_test.go
Outdated
| t.Errorf("OffsetsStore() TopicPartition error: %s", toppar.Error) | ||
| } | ||
| } | ||
| stored, err = c.OffsetsStore(nil) |
There was a problem hiding this comment.
Is there a use for accepting a nil partition set?
There was a problem hiding this comment.
nil is a valid value for a slice, in fact it is the default value. I wanted to make sure it is explicitly handled. In the fixup I changed that to an uninitialized (default) slice.
There was a problem hiding this comment.
Right, the data type allows it, but is there a point for the StoreOffsets API to allow it? Passing a nil slice should be considered a usage error and not something we need to explicitly handle (I do agree that error handling is necessary, but parameter checking in APIs is not as fruitful as it seems IMHO unless there is a security concern to it).
There was a problem hiding this comment.
Let's call it an empty slice. I think that the API should allow passing an empty slice, whatever that means in Go. If someone chooses to pass an empty slice no error is returned. We don't explicitly handle it (as in consumer.go code) but we explicitly test that an empty []TopicPartition is properly treated.
There was a problem hiding this comment.
note that this part of the diff is now changed to:
var empty []TopicPartition
stored, err = c.StoreOffsets(empty)which seems like a valid usage scenario.
Expose librdkafka's
rd_kafka_offsets_store()introduced in v0.11.0 (not yet released).Consumer.OffsetsStore() allows manual control over the committed offsets without sacrificing auto-commit functionality.
I am not sure if returning the storedOffsets is the correct thing to do, we could stop returning them to keep the API cleaner, or return a list of the failed TopicPartitions.