Offset manager: make initial offset configurable #520
Conversation
|
Given that this allows you to send the return value directly to |
| pom.offset = block.Offset | ||
| pom.metadata = block.Metadata | ||
| } else { | ||
| pom.offset = pom.parent.conf.Consumer.Offsets.Initial |
There was a problem hiding this comment.
you don't actually want to return OffsetNewest here, you want to ask the broker what the earliest available offset is and return that.
There was a problem hiding this comment.
Oh, or is this intentional... that's really not clear.
There was a problem hiding this comment.
I prefer returning the constant. This way to user of the API can know there's no offset stored, and do something specific in that case (e.g. log). Moreover, in the end you feed to value to ConsumePartition, which will always do a FetchOffset API call to validate/lookup the actual offset to use. So this saves you one roundtrip to the broker.
There was a problem hiding this comment.
Right. I would prefer to have this if in the actual Offset() method I think though; then we can put the +1 in as well and all the processing logic ends up in one place. Thoughts?
There was a problem hiding this comment.
Also then it should probably be called NextOffset?
|
I am not convinced this makes sense when the offset manager is just a subcomponent of the higher-level consumer.
Is that true? What about the migration case where you want to pass a distinct constant (manual) offset for each partition? |
|
For the migration case, I was thinking either of the following models:
|
wvanbergen
force-pushed
the
offset-manager-initial
branch
from
c9ab30a
to
6498cff
Compare
|
Renamed Also, the godoc comments probably deserve some 👀 as well to make sure it's clear how this functions should be used. |
|
I added a functional test for the offset manager. I discovered a bug with it: once you have managed a partition, and closed the partition offset manager, you cannot open a new partition offset manager for that partition. As a consumer this has to work: it's possible to start consuming a partition, then have another instance manage that partition, but later get the partition assigned back to you if the other consumer instances shuts down. My attempt at fixing: 3d1316a |
| // will eventually be flushed to the cluster based on configuration. You should only set the offset of | ||
| // messages that have been completely processed. | ||
| SetOffset(offset int64, metadata string) | ||
| // NextOffset returns the next offset should be consumed for the managed partition, accompanied by the |
There was a problem hiding this comment.
"next offset that should be consumed..."
wvanbergen
force-pushed
the
offset-manager-initial
branch
from
3d1316a
to
cedd9c3
Compare
… done, so we can register a new one later.
wvanbergen
force-pushed
the
offset-manager-initial
branch
from
cedd9c3
to
0ae0505
Compare
Offset manager: make initial offset configurable
|
👍 |
This allows you to set the initial offset
Offset()will return if no offset was committed yet for the partition.Right now, we depend on the implicit behaviour of the Kafka broker, which is returning -1. This patch makes the behaviour explicit, and allows you to change that to either
OffsetOldestor OffsetNewest`.I prefer using a config setting instead of an extra argument to ManagePartition, because this value will always be the same for every partition in a consumer.
@Shopify/kafka