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
83
84
85
86
|
package kafka
import (
"bufio"
"bytes"
"context"
"errors"
"reflect"
"strings"
"testing"
"time"
)
func TestFindCoordinatorResponseV0(t *testing.T) {
item := findCoordinatorResponseV0{
ErrorCode: 2,
Coordinator: findCoordinatorResponseCoordinatorV0{
NodeID: 3,
Host: "b",
Port: 4,
},
}
b := bytes.NewBuffer(nil)
w := &writeBuffer{w: b}
item.writeTo(w)
var found findCoordinatorResponseV0
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()
}
}
func TestClientFindCoordinator(t *testing.T) {
client, shutdown := newLocalClient()
defer shutdown()
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
defer cancel()
resp, err := waitForCoordinatorIndefinitely(ctx, client, &FindCoordinatorRequest{
Addr: client.Addr,
Key: "TransactionalID-1",
KeyType: CoordinatorKeyTypeTransaction,
})
if err != nil {
t.Fatal(err)
}
if resp.Coordinator.Host != "localhost" {
t.Fatal("Coordinator should be found @ localhost")
}
}
// WaitForCoordinatorIndefinitely is a blocking call till a coordinator is found.
func waitForCoordinatorIndefinitely(ctx context.Context, c *Client, req *FindCoordinatorRequest) (*FindCoordinatorResponse, error) {
resp, err := c.FindCoordinator(ctx, req)
for shouldRetryfindingCoordinator(resp, err) && ctx.Err() == nil {
time.Sleep(1 * time.Second)
resp, err = c.FindCoordinator(ctx, req)
}
return resp, err
}
// Should retry looking for coordinator
// Returns true when the test Kafka broker is still setting up.
func shouldRetryfindingCoordinator(resp *FindCoordinatorResponse, err error) bool {
brokerSetupIncomplete := err != nil &&
strings.Contains(
strings.ToLower(err.Error()),
strings.ToLower("unexpected EOF"))
coordinatorNotFound := resp != nil &&
resp.Error != nil &&
errors.Is(resp.Error, GroupCoordinatorNotAvailable)
return brokerSetupIncomplete || coordinatorNotFound
}
|
