riotpiaole
b704d401a1
Topic naming (kmsvc.{queue}.shard-{id} / .fifo.shard-{id} / .dlq.shard-{id}),
hash-range shard selection/splitting (murmur2-based, matching Kafka's
default partitioner), and topic admin/producer/consumer client
constructors shared by the operator and the message-plane.
78 lines
2.3 KiB
Go
78 lines
2.3 KiB
Go
package kafka
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
|
|
"github.com/twmb/franz-go/pkg/kgo"
|
|
)
|
|
|
|
// Record is a fetched Kafka record relevant to the message plane.
|
|
type Record struct {
|
|
Topic string
|
|
Partition int32
|
|
Offset int64
|
|
Key []byte
|
|
Value []byte
|
|
}
|
|
|
|
// Consumer wraps a kgo.Client consuming a queue's shard topics under the
|
|
// shared consumer group (design.md §3 ConsumerGroup). Autocommit is
|
|
// disabled: offset commits are driven by the watermark logic in
|
|
// internal/core/queue, not by the client's own interval commit, so a
|
|
// crash between receive and ack never advances the committed offset past an
|
|
// unacked message (design.md §3's at-least-once guarantee).
|
|
type Consumer struct {
|
|
client *kgo.Client
|
|
}
|
|
|
|
// NewConsumer subscribes to the given shard topics as a real consumer-group
|
|
// member: Kafka's group-rebalance protocol handles partition ownership
|
|
// across replicas (design.md §9), and the queue-operator's drain check
|
|
// (internal/kafka.Admin.ConsumerLag) reads this same group's committed
|
|
// offsets to know when a `Closing` shard is safe to delete.
|
|
func NewConsumer(brokers []string, group string, topics ...string) (*Consumer, error) {
|
|
cl, err := kgo.NewClient(
|
|
kgo.SeedBrokers(brokers...),
|
|
kgo.ConsumerGroup(group),
|
|
kgo.ConsumeTopics(topics...),
|
|
kgo.DisableAutoCommit(),
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("creating kafka consumer client: %w", err)
|
|
}
|
|
return &Consumer{client: cl}, nil
|
|
}
|
|
|
|
// AddTopics subscribes to additional shard topics created by a split,
|
|
// without losing the group membership/offsets already held for existing
|
|
// topics.
|
|
func (c *Consumer) AddTopics(topics ...string) {
|
|
c.client.AddConsumeTopics(topics...)
|
|
}
|
|
|
|
func (c *Consumer) Close() {
|
|
c.client.Close()
|
|
}
|
|
|
|
// Poll runs one non-blocking fetch iteration and returns whatever records
|
|
// were immediately available. The long-poll wait loop lives in
|
|
// internal/core/queue.ReceiveMessageService, not here.
|
|
func (c *Consumer) Poll(ctx context.Context) ([]Record, error) {
|
|
fetches := c.client.PollFetches(ctx)
|
|
if errs := fetches.Errors(); len(errs) > 0 {
|
|
return nil, fmt.Errorf("poll fetches: %w", errs[0].Err)
|
|
}
|
|
var out []Record
|
|
fetches.EachRecord(func(r *kgo.Record) {
|
|
out = append(out, Record{
|
|
Topic: r.Topic,
|
|
Partition: r.Partition,
|
|
Offset: r.Offset,
|
|
Key: r.Key,
|
|
Value: r.Value,
|
|
})
|
|
})
|
|
return out, nil
|
|
}
|