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
|
package kafka
import (
"context"
"fmt"
"net"
"time"
"github.com/segmentio/kafka-go/protocol/describeuserscramcredentials"
)
// DescribeUserScramCredentialsRequest represents a request sent to a kafka broker to
// describe user scram credentials.
type DescribeUserScramCredentialsRequest struct {
// Address of the kafka broker to send the request to.
Addr net.Addr
// List of Scram users to describe
Users []UserScramCredentialsUser
}
type UserScramCredentialsUser struct {
Name string
}
// DescribeUserScramCredentialsResponse represents a response from a kafka broker to a describe user
// credentials request.
type DescribeUserScramCredentialsResponse struct {
// The amount of time that the broker throttled the request.
Throttle time.Duration
// Top level error that occurred while attempting to describe
// the user scram credentials.
//
// The errors contain the kafka error code. Programs may use the standard
// errors.Is function to test the error against kafka error codes.
Error error
// List of described user scram credentials.
Results []DescribeUserScramCredentialsResponseResult
}
type DescribeUserScramCredentialsResponseResult struct {
User string
CredentialInfos []DescribeUserScramCredentialsCredentialInfo
Error error
}
type DescribeUserScramCredentialsCredentialInfo struct {
Mechanism ScramMechanism
Iterations int
}
// DescribeUserScramCredentials sends a user scram credentials describe request to a kafka broker and returns
// the response.
func (c *Client) DescribeUserScramCredentials(ctx context.Context, req *DescribeUserScramCredentialsRequest) (*DescribeUserScramCredentialsResponse, error) {
users := make([]describeuserscramcredentials.RequestUser, len(req.Users))
for userIdx, user := range req.Users {
users[userIdx] = describeuserscramcredentials.RequestUser{
Name: user.Name,
}
}
m, err := c.roundTrip(ctx, req.Addr, &describeuserscramcredentials.Request{
Users: users,
})
if err != nil {
return nil, fmt.Errorf("kafka.(*Client).DescribeUserScramCredentials: %w", err)
}
res := m.(*describeuserscramcredentials.Response)
responseResults := make([]DescribeUserScramCredentialsResponseResult, len(res.Results))
for responseIdx, responseResult := range res.Results {
credentialInfos := make([]DescribeUserScramCredentialsCredentialInfo, len(responseResult.CredentialInfos))
for credentialInfoIdx, credentialInfo := range responseResult.CredentialInfos {
credentialInfos[credentialInfoIdx] = DescribeUserScramCredentialsCredentialInfo{
Mechanism: ScramMechanism(credentialInfo.Mechanism),
Iterations: int(credentialInfo.Iterations),
}
}
responseResults[responseIdx] = DescribeUserScramCredentialsResponseResult{
User: responseResult.User,
CredentialInfos: credentialInfos,
Error: makeError(responseResult.ErrorCode, responseResult.ErrorMessage),
}
}
ret := &DescribeUserScramCredentialsResponse{
Throttle: makeDuration(res.ThrottleTimeMs),
Error: makeError(res.ErrorCode, res.ErrorMessage),
Results: responseResults,
}
return ret, nil
}
|
