Open arjunnair1997 opened 1 month ago
I'm on board, though I'd skip (2).
I think this is the full patch:
diff --git a/pkg/kgo/record_and_fetch.go b/pkg/kgo/record_and_fetch.go
index 4f1ebe6f..55315644 100644
--- a/pkg/kgo/record_and_fetch.go
+++ b/pkg/kgo/record_and_fetch.go
@@ -274,6 +274,8 @@ func (p *FetchPartition) EachRecord(fn func(*Record)) {
type FetchTopic struct {
// Topic is the topic this is for.
Topic string
+ // TopicID is the ID of the topic, if your cluster supports topic IDs.
+ TopicID [16]byte
// Partitions contains individual partitions in the topic that were
// fetched.
Partitions []FetchPartition
@@ -560,6 +562,7 @@ func (fs Fetches) EachTopic(fn func(FetchTopic)) {
for topic, partitions := range topics {
fn(FetchTopic{
topic,
+ [16]byte{},
partitions,
})
}
diff --git a/pkg/kgo/source.go b/pkg/kgo/source.go
index 7a55ef0c..cddb4189 100644
--- a/pkg/kgo/source.go
+++ b/pkg/kgo/source.go
@@ -979,6 +979,7 @@ func (s *source) handleReqResp(br *broker, req *fetchRequest, resp *kmsg.FetchRe
fetchTopic := FetchTopic{
Topic: topic,
+ TopicID: rt.TopicID,
Partitions: make([]FetchPartition, 0, len(rt.Partitions)),
}
I can push this after I cut a patch release. There aren't many new features at the moment, so I may wait on releasing this for a little bit -- is that alright? (potentially a month?)
Hey, that's great, thanks. If you decide to merge it in now, I can pin franz-go master, and start using it. Otherwise, I can fork franz go for a bit until you decide to merge it in.
I was burned in the past by merging a feature then having to revert it to issue a quick bugfix release; if you're open to using the branch for now, please do that 😅
I'll leave this issue open, it'll be closed once I merge the PR -- which I'll do right as I'm about to cut a release.
Hello,
I'm trying to use the franz go kafka client to consume records from a source cluster. I'm using a direct consumer, because the workload is to consume a list of arbitrary (topic id, partition id, offset) periodically.
I'm trying to achieve this by keeping a single client alive. When it receives the list of (topic id, partition id, offset) to consume from, it performs remove consume topic partitions using
RemoveConsumePartitions
or evenPurgeTopicsFromClient
, and adds the new consume topic partitions usingAddConsumePartitions
.However, the above APIs use the topic name string and not the topic ids. I think I can make my use case of ensuring I'm fetching from the correct topic work if the
FetchTopic
struct also has the topic id in it and it's returned from thePollRecords
function with the topic ids also set.I think it's possible to do this by:
FetchTopic
.FetchTopic
.What do you think? If you're open to this, then I can implement this.