Skip to content

Commit b952e63

Browse files
zirkomezirkome
authored
Add CreateAcls Admin API support (#839)
This adds a CreateACLs method to the kafka Client to supports the CreateAcls Admin API. Signed-off-by: Guillaume Fillon <[email protected]>
1 parent 8f2199a commit b952e63

File tree

7 files changed

+273
-17
lines changed

7 files changed

+273
-17
lines changed

‎.circleci/config.yml‎

Lines changed: 28 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -129,7 +129,10 @@ jobs:
129129
ports:
130130
- 9092:9092
131131
- 9093:9093
132-
environment: *environment
132+
environment:
133+
<<: *environment
134+
KAFKA_AUTHORIZER_CLASS_NAME: 'kafka.security.auth.SimpleAclAuthorizer'
135+
KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: 'true'
133136
steps: *steps
134137

135138
kafka-211:
@@ -145,7 +148,10 @@ jobs:
145148
ports:
146149
- 9092:9092
147150
- 9093:9093
148-
environment: *environment
151+
environment:
152+
<<: *environment
153+
KAFKA_AUTHORIZER_CLASS_NAME: 'kafka.security.auth.SimpleAclAuthorizer'
154+
KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: 'true'
149155
steps: *steps
150156

151157
kafka-222:
@@ -161,7 +167,10 @@ jobs:
161167
ports:
162168
- 9092:9092
163169
- 9093:9093
164-
environment: *environment
170+
environment:
171+
<<: *environment
172+
KAFKA_AUTHORIZER_CLASS_NAME: 'kafka.security.auth.SimpleAclAuthorizer'
173+
KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: 'true'
165174
steps: *steps
166175

167176
kafka-231:
@@ -177,7 +186,10 @@ jobs:
177186
ports:
178187
- 9092:9092
179188
- 9093:9093
180-
environment: *environment
189+
environment:
190+
<<: *environment
191+
KAFKA_AUTHORIZER_CLASS_NAME: 'kafka.security.auth.SimpleAclAuthorizer'
192+
KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: 'true'
181193
steps: *steps
182194

183195
kafka-241:
@@ -204,7 +216,10 @@ jobs:
204216
ports:
205217
- 9092:9092
206218
- 9093:9093
207-
environment: *environment
219+
environment:
220+
<<: *environment
221+
KAFKA_AUTHORIZER_CLASS_NAME: 'kafka.security.authorizer.AclAuthorizer'
222+
KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: 'true'
208223
steps: *steps
209224

210225
kafka-260:
@@ -231,7 +246,10 @@ jobs:
231246
ports:
232247
- 9092:9092
233248
- 9093:9093
234-
environment: *environment
249+
environment:
250+
<<: *environment
251+
KAFKA_AUTHORIZER_CLASS_NAME: 'kafka.security.authorizer.AclAuthorizer'
252+
KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: 'true'
235253
steps: *steps
236254

237255
kafka-271:
@@ -258,7 +276,10 @@ jobs:
258276
ports:
259277
- 9092:9092
260278
- 9093:9093
261-
environment: *environment
279+
environment:
280+
<<: *environment
281+
KAFKA_AUTHORIZER_CLASS_NAME: 'kafka.security.authorizer.AclAuthorizer'
282+
KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: 'true'
262283
steps: *steps
263284

264285
workflows:

‎createacl_test.go‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
package kafka
2+
3+
import (
4+
"context"
5+
"testing"
6+
7+
ktesting "github.com/segmentio/kafka-go/testing"
8+
)
9+
10+
func TestClientCreateACLs(t *testing.T) {
11+
if !ktesting.KafkaIsAtLeast("2.0.1") {
12+
return
13+
}
14+
15+
client, shutdown := newLocalClient()
16+
defer shutdown()
17+
18+
res, err := client.CreateACLs(context.Background(), &CreateACLsRequest{
19+
ACLs: []ACLEntry{
20+
{
21+
Principal: "User:alice",
22+
PermissionType: ACLPermissionTypeAllow,
23+
Operation: ACLOperationTypeRead,
24+
ResourceType: ResourceTypeTopic,
25+
ResourcePatternType: PatternTypeLiteral,
26+
ResourceName: "fake-topic-for-alice",
27+
Host: "*",
28+
},
29+
{
30+
Principal: "User:bob",
31+
PermissionType: ACLPermissionTypeAllow,
32+
Operation: ACLOperationTypeRead,
33+
ResourceType: ResourceTypeGroup,
34+
ResourcePatternType: PatternTypeLiteral,
35+
ResourceName: "fake-group-for-bob",
36+
Host: "*",
37+
},
38+
},
39+
})
40+
if err != nil {
41+
t.Fatal(err)
42+
}
43+
44+
for _, err := range res.Errors {
45+
if err != nil {
46+
t.Error(err)
47+
}
48+
}
49+
}

‎createacls.go‎

Lines changed: 108 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,108 @@
1+
package kafka
2+
3+
import (
4+
"context"
5+
"fmt"
6+
"net"
7+
"time"
8+
9+
"github.com/segmentio/kafka-go/protocol/createacls"
10+
)
11+
12+
// CreateACLsRequest represents a request sent to a kafka broker to add
13+
// new ACLs.
14+
type CreateACLsRequest struct {
15+
// Address of the kafka broker to send the request to.
16+
Addr net.Addr
17+
18+
// List of ACL to create.
19+
ACLs []ACLEntry
20+
}
21+
22+
// CreateACLsResponse represents a response from a kafka broker to an ACL
23+
// creation request.
24+
type CreateACLsResponse struct {
25+
// The amount of time that the broker throttled the request.
26+
Throttle time.Duration
27+
28+
// List of errors that occurred while attempting to create
29+
// the ACLs.
30+
//
31+
// The errors contain the kafka error code. Programs may use the standard
32+
// errors.Is function to test the error against kafka error codes.
33+
Errors []error
34+
}
35+
36+
type ACLPermissionType int8
37+
38+
const (
39+
ACLPermissionTypeUnknown ACLPermissionType = 0
40+
ACLPermissionTypeAny ACLPermissionType = 1
41+
ACLPermissionTypeDeny ACLPermissionType = 2
42+
ACLPermissionTypeAllow ACLPermissionType = 3
43+
)
44+
45+
type ACLOperationType int8
46+
47+
const (
48+
ACLOperationTypeUnknown ACLOperationType = 0
49+
ACLOperationTypeAny ACLOperationType = 1
50+
ACLOperationTypeAll ACLOperationType = 2
51+
ACLOperationTypeRead ACLOperationType = 3
52+
ACLOperationTypeWrite ACLOperationType = 4
53+
ACLOperationTypeCreate ACLOperationType = 5
54+
ACLOperationTypeDelete ACLOperationType = 6
55+
ACLOperationTypeAlter ACLOperationType = 7
56+
ACLOperationTypeDescribe ACLOperationType = 8
57+
ACLOperationTypeClusterAction ACLOperationType = 9
58+
ACLOperationTypeDescribeConfigs ACLOperationType = 10
59+
ACLOperationTypeAlterConfigs ACLOperationType = 11
60+
ACLOperationTypeIdempotentWrite ACLOperationType = 12
61+
)
62+
63+
type ACLEntry struct {
64+
ResourceType ResourceType
65+
ResourceName string
66+
ResourcePatternType PatternType
67+
Principal string
68+
Host string
69+
Operation ACLOperationType
70+
PermissionType ACLPermissionType
71+
}
72+
73+
// CreateACLs sends ACLs creation request to a kafka broker and returns the
74+
// response.
75+
func (c *Client) CreateACLs(ctx context.Context, req *CreateACLsRequest) (*CreateACLsResponse, error) {
76+
acls := make([]createacls.RequestACLs, 0, len(req.ACLs))
77+
78+
for _, acl := range req.ACLs {
79+
acls = append(acls, createacls.RequestACLs{
80+
ResourceType: int8(acl.ResourceType),
81+
ResourceName: acl.ResourceName,
82+
ResourcePatternType: int8(acl.ResourcePatternType),
83+
Principal: acl.Principal,
84+
Host: acl.Host,
85+
Operation: int8(acl.Operation),
86+
PermissionType: int8(acl.PermissionType),
87+
})
88+
}
89+
90+
m, err := c.roundTrip(ctx, req.Addr, &createacls.Request{
91+
Creations: acls,
92+
})
93+
if err != nil {
94+
return nil, fmt.Errorf("kafka.(*Client).CreateACLs: %w", err)
95+
}
96+
97+
res := m.(*createacls.Response)
98+
ret := &CreateACLsResponse{
99+
Throttle: makeDuration(res.ThrottleTimeMs),
100+
Errors: make([]error, 0, len(res.Results)),
101+
}
102+
103+
for _, t := range res.Results {
104+
ret.Errors = append(ret.Errors, makeError(t.ErrorCode, t.ErrorMessage))
105+
}
106+
107+
return ret, nil
108+
}

‎describeconfigs.go‎

Lines changed: 0 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -24,15 +24,6 @@ type DescribeConfigsRequest struct {
2424
IncludeDocumentation bool
2525
}
2626

27-
type ResourceType int8
28-
29-
const (
30-
// See https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/config/ConfigResource.java#L36
31-
ResourceTypeUnknown ResourceType = 0
32-
ResourceTypeTopic ResourceType = 2
33-
ResourceTypeBroker ResourceType = 4
34-
)
35-
3627
type DescribeConfigRequestResource struct {
3728
// Resource Type
3829
ResourceType ResourceType
@@ -122,7 +113,6 @@ func (c *Client) DescribeConfigs(ctx context.Context, req *DescribeConfigsReques
122113
IncludeSynonyms: req.IncludeSynonyms,
123114
IncludeDocumentation: req.IncludeDocumentation,
124115
})
125-
126116
if err != nil {
127117
return nil, fmt.Errorf("kafka.(*Client).DescribeConfigs: %w", err)
128118
}

‎docker-compose.yml‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,8 @@ services:
2121
KAFKA_LISTENERS: 'PLAINTEXT://:9092,SASL_PLAINTEXT://:9093'
2222
KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://localhost:9092,SASL_PLAINTEXT://localhost:9093'
2323
KAFKA_SASL_ENABLED_MECHANISMS: 'PLAIN,SCRAM-SHA-256,SCRAM-SHA-512'
24+
KAFKA_AUTHORIZER_CLASS_NAME: 'kafka.security.auth.SimpleAclAuthorizer'
25+
KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: 'true'
2426
KAFKA_OPTS: "-Djava.security.auth.login.config=/opt/kafka/config/kafka_server_jaas.conf"
2527
CUSTOM_INIT_SCRIPT: |-
2628
echo -e 'KafkaServer {\norg.apache.kafka.common.security.scram.ScramLoginModule required\n username="adminscram"\n password="admin-secret";\n org.apache.kafka.common.security.plain.PlainLoginModule required\n username="adminplain"\n password="admin-secret"\n user_adminplain="admin-secret";\n };' > /opt/kafka/config/kafka_server_jaas.conf;

‎protocol/createacls/createacls.go‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
package createacls
2+
3+
import "github.com/segmentio/kafka-go/protocol"
4+
5+
func init() {
6+
protocol.Register(&Request{}, &Response{})
7+
}
8+
9+
type Request struct {
10+
// We need at least one tagged field to indicate that v2+ uses "flexible"
11+
// messages.
12+
_ struct{} `kafka:"min=v2,max=v2,tag"`
13+
14+
Creations []RequestACLs `kafka:"min=v0,max=v2"`
15+
}
16+
17+
func (r *Request) ApiKey() protocol.ApiKey { return protocol.CreateAcls }
18+
19+
func (r *Request) Broker(cluster protocol.Cluster) (protocol.Broker, error) {
20+
return cluster.Brokers[cluster.Controller], nil
21+
}
22+
23+
type RequestACLs struct {
24+
ResourceType int8 `kafka:"min=v0,max=v2"`
25+
ResourceName string `kafka:"min=v0,max=v2"`
26+
ResourcePatternType int8 `kafka:"min=v0,max=v2"`
27+
Principal string `kafka:"min=v0,max=v2"`
28+
Host string `kafka:"min=v0,max=v2"`
29+
Operation int8 `kafka:"min=v0,max=v2"`
30+
PermissionType int8 `kafka:"min=v0,max=v2"`
31+
}
32+
33+
type Response struct {
34+
// We need at least one tagged field to indicate that v2+ uses "flexible"
35+
// messages.
36+
_ struct{} `kafka:"min=v2,max=v2,tag"`
37+
38+
ThrottleTimeMs int32 `kafka:"min=v0,max=v2"`
39+
Results []ResponseACLs `kafka:"min=v0,max=v2"`
40+
}
41+
42+
func (r *Response) ApiKey() protocol.ApiKey { return protocol.CreateAcls }
43+
44+
type ResponseACLs struct {
45+
ErrorCode int16 `kafka:"min=v0,max=v2"`
46+
ErrorMessage string `kafka:"min=v0,max=v2,nullable"`
47+
}
48+
49+
var _ protocol.BrokerMessage = (*Request)(nil)

‎resource.go‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
package kafka
2+
3+
// https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/resource/ResourceType.java
4+
type ResourceType int8
5+
6+
const (
7+
ResourceTypeUnknown ResourceType = 0
8+
ResourceTypeAny ResourceType = 1
9+
ResourceTypeTopic ResourceType = 2
10+
ResourceTypeGroup ResourceType = 3
11+
// See https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/config/ConfigResource.java#L36
12+
ResourceTypeBroker ResourceType = 4
13+
ResourceTypeCluster ResourceType = 4
14+
ResourceTypeTransactionalID ResourceType = 5
15+
ResourceTypeDelegationToken ResourceType = 6
16+
)
17+
18+
// https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/resource/PatternType.java
19+
type PatternType int8
20+
21+
const (
22+
// PatternTypeUnknown represents any PatternType which this client cannot
23+
// understand.
24+
PatternTypeUnknown PatternType = 0
25+
// PatternTypeAny matches any resource pattern type.
26+
PatternTypeAny PatternType = 1
27+
// PatternTypeMatch perform pattern matching.
28+
PatternTypeMatch PatternType = 2
29+
// PatternTypeLiteral represents a literal name.
30+
// A literal name defines the full name of a resource, e.g. topic with name
31+
// 'foo', or group with name 'bob'.
32+
PatternTypeLiteral PatternType = 3
33+
// PatternTypePrefixed represents a prefixed name.
34+
// A prefixed name defines a prefix for a resource, e.g. topics with names
35+
// that start with 'foo'.
36+
PatternTypePrefixed PatternType = 4
37+
)

0 commit comments

Comments
 (0)