forked from segmentio/kafka-go
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathconn_test.go
147 lines (121 loc) · 2.67 KB
/
conn_test.go
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
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
package kafka
import (
"context"
"fmt"
"math/rand"
"net"
"strconv"
"testing"
"time"
)
func TestConn(t *testing.T) {
tests := []struct {
scenario string
function func(*testing.T, *Conn)
}{
{
scenario: "close right away",
function: testConnClose,
},
{
scenario: "ensure the initial offset of a connection is the first offset",
function: testConnFirstOffset,
},
{
scenario: "write a single message to kafka",
function: testConnWrite,
},
{
scenario: "ensure the connection can seek to the first offset",
function: testConnSeekFirstOffset,
},
{
scenario: "ensure the connection can seek to the last offset",
function: testConnSeekLastOffset,
},
{
scenario: "ensure the connection can seek to a random offset",
function: testConnSeekRandomOffset,
},
}
for _, test := range tests {
testFunc := test.function
t.Run(test.scenario, func(t *testing.T) {
t.Parallel()
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
topic := fmt.Sprintf("kafka-go-%d", rand.Int63())
conn, err := (&Dialer{
Resolver: &net.Resolver{},
}).DialLeader(ctx, "tcp", "localhost:9092", topic, 0)
if err != nil {
t.Fatal("failed to open a new kafka connection:", err)
}
defer conn.Close()
testFunc(t, conn)
})
}
}
func testConnClose(t *testing.T, conn *Conn) {
if err := conn.Close(); err != nil {
t.Error(err)
}
}
func testConnFirstOffset(t *testing.T, conn *Conn) {
offset, whence := conn.Offset()
if offset != 0 && whence != 0 {
t.Error("bad first offset:", offset, whence)
}
}
func testConnWrite(t *testing.T, conn *Conn) {
b := []byte("Hello World!")
n, err := conn.Write(b)
if err != nil {
t.Error(err)
}
if n != len(b) {
t.Error("bad length returned by (*Conn).Write:", n)
}
}
func testConnSeekFirstOffset(t *testing.T, conn *Conn) {
for i := 0; i != 10; i++ {
if _, err := conn.Write([]byte(strconv.Itoa(i))); err != nil {
t.Fatal(err)
}
}
offset, err := conn.Seek(0, 0)
if err != nil {
t.Error(err)
}
if offset != 0 {
t.Error("bad offset:", offset)
}
}
func testConnSeekLastOffset(t *testing.T, conn *Conn) {
for i := 0; i != 10; i++ {
if _, err := conn.Write([]byte(strconv.Itoa(i))); err != nil {
t.Fatal(err)
}
}
offset, err := conn.Seek(0, 2)
if err != nil {
t.Error(err)
}
if offset != 10 {
t.Error("bad offset:", offset)
}
}
func testConnSeekRandomOffset(t *testing.T, conn *Conn) {
for i := 0; i != 10; i++ {
if _, err := conn.Write([]byte(strconv.Itoa(i))); err != nil {
t.Fatal(err)
}
}
offset, err := conn.Seek(3, 1)
if err != nil {
t.Error(err)
}
if offset != 3 {
t.Error("bad offset:", offset)
}
}