1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
|
package kafka
import (
"bufio"
"bytes"
"context"
"log"
"os"
"reflect"
"testing"
"time"
)
func TestClientHeartbeat(t *testing.T) {
client, topic, shutdown := newLocalClientAndTopic()
defer shutdown()
ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
defer cancel()
groupID := makeGroupID()
group, err := NewConsumerGroup(ConsumerGroupConfig{
ID: groupID,
Topics: []string{topic},
Brokers: []string{"localhost:9092"},
HeartbeatInterval: 2 * time.Second,
RebalanceTimeout: 2 * time.Second,
RetentionTime: time.Hour,
Logger: log.New(os.Stdout, "cg-test: ", 0),
})
if err != nil {
t.Fatal(err)
}
defer group.Close()
gen, err := group.Next(ctx)
if err != nil {
t.Fatal(err)
}
ctx, cancel = context.WithTimeout(context.Background(), time.Second*30)
defer cancel()
resp, err := client.Heartbeat(ctx, &HeartbeatRequest{
GroupID: groupID,
GenerationID: gen.ID,
MemberID: gen.MemberID,
})
if err != nil {
t.Fatal(err)
}
if resp.Error != nil {
t.Error(resp.Error)
}
}
func TestHeartbeatRequestV0(t *testing.T) {
item := heartbeatResponseV0{
ErrorCode: 2,
}
b := bytes.NewBuffer(nil)
w := &writeBuffer{w: b}
item.writeTo(w)
var found heartbeatResponseV0
remain, err := (&found).readFrom(bufio.NewReader(b), b.Len())
if err != nil {
t.Error(err)
t.FailNow()
}
if remain != 0 {
t.Errorf("expected 0 remain, got %v", remain)
t.FailNow()
}
if !reflect.DeepEqual(item, found) {
t.Error("expected item and found to be the same")
t.FailNow()
}
}
|
