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
|
package kafka
import (
"context"
"net"
)
// The Resolver interface is used as an abstraction to provide service discovery
// of the hosts of a kafka cluster.
type Resolver interface {
// LookupHost looks up the given host using the local resolver.
// It returns a slice of that host's addresses.
LookupHost(ctx context.Context, host string) (addrs []string, err error)
}
// BrokerResolver is an interface implemented by types that translate host
// names into a network address.
//
// This resolver is not intended to be a general purpose interface. Instead,
// it is tailored to the particular needs of the kafka protocol, with the goal
// being to provide a flexible mechanism for extending broker name resolution
// while retaining context that is specific to interacting with a kafka cluster.
//
// Resolvers must be safe to use from multiple goroutines.
type BrokerResolver interface {
// Returns the IP addresses of the broker passed as argument.
LookupBrokerIPAddr(ctx context.Context, broker Broker) ([]net.IPAddr, error)
}
// NewBrokerResolver constructs a Resolver from r.
//
// If r is nil, net.DefaultResolver is used instead.
func NewBrokerResolver(r *net.Resolver) BrokerResolver {
return brokerResolver{r}
}
type brokerResolver struct {
*net.Resolver
}
func (r brokerResolver) LookupBrokerIPAddr(ctx context.Context, broker Broker) ([]net.IPAddr, error) {
ipAddrs, err := r.LookupIPAddr(ctx, broker.Host)
if err != nil {
return nil, err
}
if len(ipAddrs) == 0 {
return nil, &net.DNSError{
Err: "no addresses were returned by the resolver",
Name: broker.Host,
IsTemporary: true,
IsNotFound: true,
}
}
return ipAddrs, nil
}
|
