Merge pull request #13 from zerodha/develop

Merge develop to master

Author: kalbhor

README.md

@@ -1,19 +1,19 @@
 <a href="https://zerodha.tech"><img src="https://zerodha.tech/static/images/github-badge.svg" align="right" /></a>
-## Relay
+## kaf-relay
 
-Relay is an opinionated program designed to replicate messages on topics from one Kafka cluster to another Kafka cluster.
+kaf-relay is an opinionated, high performance program for keeping Kafka clusters in sync by replicating topics. It is specfically designed for high-availability with background healthchecks, offset tracking, and topic lag checks.
 
 ### Features
 
 * Topic Forwarding: Relay consumes messages from topics in one Kafka cluster and forwards them to topics in another Kafka cluster.
 * Authentication: PLAIN, SCRAM-SHA-256, SCRAM-SHA-512
 * Topic Remapping: Relay allows you to specify topic remappings, enabling you to map a topic from the source cluster to a different topic in the destination cluster.
-* Consumer group failover: Assuming we have multiple identical kafkas (separate nodes 1...N) at the upstream side, this mode allows us to fallback to the next kafka in a round-robin fashion if current broker goes down. This allows us to deduplicate messages downstream without using any external stores.
-* Topic lag failover: In addition to consumer group failover, if there is a lag between node1 and node2, we initiate an immediate switch-over to node2.
+* Consumer group failover: Given identical Kafka instances (separate nodes 1...N) at the upstream, instantly switch over to the next node in a round-robin fashion on current node failure. Offset tracking between source and target nodes allows de-duplication without external stores.
+* Topic lag failover: Monitors offsets amongst N identical nodes to detect lags and to instantly switch upstream consumer nodes.
 * Stop at end: Flag `--stop-at-end` allows the program to stop after reaching the end of consumer topic offsets that was picked up on boot.
 * Filter messages using go plugins: Flag `--filter` allows the program to filter messages based on the logic in plugin code.
 
-#### relay in different modes
+#### kaf-relay in different modes
 
 ![image](./screenshots/relay.png)
 
@@ -25,7 +25,7 @@ Relay is an opinionated program designed to replicate messages on topics from on
 ## Installation
 
 ```bash
-git clone https://github.com/joeirimpan/kaf-relay.git
+git clone https://github.com/zerodha/kaf-relay.git
 cd kaf-relay
 make dist
 ```
@@ -92,4 +92,4 @@ kafka_relay_msg_count{source="topicA", destination="machineX_topicA", partition=
 kafka_relay_msg_count{source="topicA", destination="machineX_topicA", partition="3"} 44
 kafka_relay_msg_count{source="topicA", destination="machineX_topicA", partition="4"} 44
 kafka_relay_msg_count{source="topicA", destination="machineX_topicA", partition="5"} 100
-```
+```

config.go

@@ -73,6 +73,7 @@ type ConsumerGroupCfg struct {
 	ClientCfg `koanf:",squash"`
 
 	GroupID     string        `koanf:"group_id"`
+	InstanceID  string        `koanf:"instance_id"`
 	MaxWaitTime time.Duration `koanf:"max_wait_time"`
 
 	MaxFailovers int `koanf:"max_failovers"`

config.toml.sample

@@ -34,6 +34,7 @@ max_wait_time = "10ms"
 offset_commit_interval = "100ms"
 initial_offset = "start"
 group_id = "relay_group1"
+instance_id = "kaf_relay"
 
 max_failovers = -1 # infinite
 
@@ -57,6 +58,7 @@ max_wait_time = "10ms"
 offset_commit_interval = "100ms"
 initial_offset = "start"
 group_id = "relay_group2"
+instance_id = "kaf_relay"
 
 max_failovers = -1 # infinite
 

go.mod

@@ -1,4 +1,4 @@
-module github.com/joeirimpan/kaf-relay
+module github.com/zerodha/kaf-relay
 
 go 1.21
 

initz.go

@@ -10,14 +10,9 @@ import (
 	"strings"
 
 	"github.com/VictoriaMetrics/metrics"
-	"github.com/joeirimpan/kaf-relay/filter"
 	"github.com/knadh/koanf/v2"
 	"github.com/twmb/franz-go/pkg/kgo"
-)
-
-var (
-	// instanceID is for the static consumer group memeber.
-	instanceID = "kaf-relay"
+	"github.com/zerodha/kaf-relay/filter"
 )
 
 // getProducerClient returns a kafka producer client.
@@ -154,7 +149,7 @@ func initConsumerGroup(ctx context.Context, cfg ConsumerGroupCfg, l *slog.Logger
 		kgo.FetchMaxWait(cfg.MaxWaitTime),
 		kgo.ConsumeTopics(cfg.Topics...),
 		kgo.ConsumerGroup(cfg.GroupID),
-		kgo.InstanceID(instanceID),
+		kgo.InstanceID(cfg.InstanceID),
 		kgo.SessionTimeout(cfg.SessionTimeout),
 		kgo.DisableAutoCommit(),
 		kgo.OnPartitionsAssigned(onAssigned),

relay.go

@@ -9,9 +9,9 @@ import (
 	"time"
 
 	"github.com/VictoriaMetrics/metrics"
-	"github.com/joeirimpan/kaf-relay/filter"
 	"github.com/twmb/franz-go/pkg/kadm"
 	"github.com/twmb/franz-go/pkg/kgo"
+	"github.com/zerodha/kaf-relay/filter"
 )
 
 // relay represents the input, output kafka and the remapping necessary to forward messages from one topic to another.

utils.go

@@ -185,7 +185,7 @@ func appendSASL(opts []kgo.Opt, cfg ClientCfg) []kgo.Opt {
 // leaveAndResetOffsets leaves the current consumer group and resets its offset if given.
 func leaveAndResetOffsets(ctx context.Context, cl *kgo.Client, cfg ConsumerGroupCfg, offsets map[string]map[int32]kgo.Offset, l *slog.Logger) error {
 	// leave group; mark the group as `Empty` before attempting to reset offsets.
-	l.Debug("leaving group", "group_id", cfg.GroupID)
+	l.Debug("leaving group", "group_id", cfg.GroupID, "instance_id", cfg.InstanceID)
 	if err := leaveGroup(ctx, cl, cfg); err != nil {
 		return err
 	}
@@ -203,12 +203,20 @@ func leaveAndResetOffsets(ctx context.Context, cl *kgo.Client, cfg ConsumerGroup
 
 // leaveGroup leaves the consumer group with our instance id
 func leaveGroup(ctx context.Context, cl *kgo.Client, cfg ConsumerGroupCfg) error {
-	req := kmsg.NewPtrLeaveGroupRequest()
-	req.Group = cfg.GroupID
+	l := kadm.LeaveGroup(cfg.GroupID).
+		Reason("resetting offsets").
+		InstanceIDs(cfg.InstanceID)
 
-	req.Members = []kmsg.LeaveGroupRequestMember{{InstanceID: &instanceID}}
-	_, err := req.RequestWith(ctx, cl)
-	return err
+	resp, err := kadm.NewClient(cl).LeaveGroup(ctx, l)
+	if err != nil {
+		return err
+	}
+
+	if err := resp.Error(); err != nil {
+		return err
+	}
+
+	return nil
 }
 
 // resetOffsets resets the consumer group with the given offsets map.