Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func SecurityOptions ¶
func SecurityOptions(protocol string, sslConf *config.KafkaSSL, saslConf *config.KafkaSASL) ([]kgo.Opt, error)
SecurityOptions translates the security_protocol / ssl / sasl config blocks into franz-go client options. The config keys mirror librdkafka's, which is what the Python engine configures.
Types ¶
type OffsetSeeker ¶ added in v1.0.5
type OffsetSeeker struct {
// contains filtered or unexported fields
}
OffsetSeeker carries the pipeline's durable positions into the consumer group's join, so a restart resumes where the state database says it stopped rather than where the group happens to sit.
It exists because the obvious approach does not work. Committing those positions before consumption starts issues an OffsetCommit with an empty member ID, which Kafka accepts only while the group is Empty. The group is not empty on exactly the restart this feature exists for: a killed process stays a member until its session times out, so the commit fails with UNKNOWN_MEMBER_ID and the pipeline crash-loops until the old session expires. Measured 3/3 against a group with one other member.
franz-go's AdjustFetchOffsetsFn runs inside the join instead, after the group's committed offsets are fetched and before consumption begins. It needs no commit and no member ID, so group membership stops mattering.
func NewOffsetSeeker ¶ added in v1.0.5
func NewOffsetSeeker() *OffsetSeeker
NewOffsetSeeker returns a seeker holding no positions, which leaves the group's own offsets untouched.
func (*OffsetSeeker) Adjust ¶ added in v1.0.5
func (s *OffsetSeeker) Adjust(_ context.Context, fetched map[string]map[int32]kgo.Offset) (map[string]map[int32]kgo.Offset, error)
Adjust replaces the group's committed offset with the durable one, for every partition the pipeline holds a mark for. This is where a disagreement between Kafka and the state database is settled, and the state database wins.
A stored mark names the last offset processed, so consumption resumes at the next one -- the same +1 convention CommitMarks uses.
Only partitions already present in the fetched map are touched. franz-go assigns exactly the map this returns (consumer_group.go:1737), so adding a partition here would make this member consume a partition the group never assigned it. A partition with no mark keeps the group's offset, so a pipeline that gains partitions still follows auto_offset_reset for them.
func (*OffsetSeeker) SetMarks ¶ added in v1.0.5
func (s *OffsetSeeker) SetMarks(marks *core.Marks)
SetMarks records the positions to resume from. Call it before the first poll; the join has not happened yet at that point.
type Option ¶
type Option func(*Source)
func WithChannelBuffer ¶
func WithLogger ¶
func WithReadTimeout ¶
func WithSeeker ¶ added in v1.0.5
func WithSeeker(seeker *OffsetSeeker) Option
WithSeeker gives the source the seeker registered on its client, which is what SeekTo writes durable positions into. The seeker has to be built before the client, because it is a client option.
type Source ¶
type Source struct {
// contains filtered or unexported fields
}
func (*Source) CommitMarks ¶ added in v1.0.4
CommitMarks commits exactly the positions the pipeline has finished with.
Commit above commits everything this source has fetched, and the poll goroutine fetches well ahead of the pipeline: after one 20,000-message batch it had committed offset 70,086. A crash then lost the difference with the consumer group showing no lag. Kafka commits the next offset to read, so a mark at offset N commits N+1.
func (*Source) SeekTo ¶ added in v1.0.5
SeekTo resumes consumption from positions recorded in the pipeline's state database, so a restart picks up where the durable state left off rather than wherever the consumer group happens to sit.
The positions are handed to the seeker, which applies them during the group's join. Committing them here instead does not work: that commit carries an empty member ID and Kafka refuses it unless the group is Empty, which it is not after a crash. See OffsetSeeker.
Empty marks are a no-op rather than a seek to zero. "Nothing recorded" and "recorded position zero" are different facts: the first must leave auto_offset_reset in charge, and seeking to zero would silently replay an entire topic on a pipeline's first run against a fresh state file.
Source Files