// Admin ACL operations (Phase 5): DescribeAcls v2-v3, CreateAcls v2-v3,
// and DeleteAcls v2-v3. Field order is pinned from the message schemas
// in the Kafka 4.3 tree under clients/src/main/resources/common/message/;
// the enum byte values from ResourceType.java, PatternType.java,
// AclOperation.java, and AclPermissionType.java. v3 is wire-identical
// to v2 (it only adds the USER resource type), so one shape serves both.
// Per-item error codes travel as values; the Admin retry policy
// re-issues retriable ones.
///|
/// ACL resource types (ResourceType byte values).
pub const ACL_RESOURCE_UNKNOWN : Int = 0
///|
pub const ACL_RESOURCE_ANY : Int = 1
///|
pub const ACL_RESOURCE_TOPIC : Int = 2
///|
pub const ACL_RESOURCE_GROUP : Int = 3
///|
pub const ACL_RESOURCE_CLUSTER : Int = 4
///|
pub const ACL_RESOURCE_TRANSACTIONAL_ID : Int = 5
///|
pub const ACL_RESOURCE_DELEGATION_TOKEN : Int = 6
///|
/// v3 and newer brokers.
pub const ACL_RESOURCE_USER : Int = 7
///|
/// Resource pattern types (PatternType byte values).
pub const ACL_PATTERN_UNKNOWN : Int = 0
///|
pub const ACL_PATTERN_ANY : Int = 1
///|
/// MATCH in filters; never describes a stored binding.
pub const ACL_PATTERN_MATCH : Int = 2
///|
pub const ACL_PATTERN_LITERAL : Int = 3
///|
pub const ACL_PATTERN_PREFIXED : Int = 4
///|
/// ACL operations (AclOperation byte values).
pub const ACL_OPERATION_UNKNOWN : Int = 0
///|
pub const ACL_OPERATION_ANY : Int = 1
///|
pub const ACL_OPERATION_ALL : Int = 2
///|
pub const ACL_OPERATION_READ : Int = 3
///|
pub const ACL_OPERATION_WRITE : Int = 4
///|
pub const ACL_OPERATION_CREATE : Int = 5
///|
pub const ACL_OPERATION_DELETE : Int = 6
///|
pub const ACL_OPERATION_ALTER : Int = 7
///|
pub const ACL_OPERATION_DESCRIBE : Int = 8
///|
pub const ACL_OPERATION_CLUSTER_ACTION : Int = 9
///|
pub const ACL_OPERATION_DESCRIBE_CONFIGS : Int = 10
///|
pub const ACL_OPERATION_ALTER_CONFIGS : Int = 11
///|
pub const ACL_OPERATION_IDEMPOTENT_WRITE : Int = 12
///|
pub const ACL_OPERATION_CREATE_TOKENS : Int = 13
///|
pub const ACL_OPERATION_DESCRIBE_TOKENS : Int = 14
///|
/// KIP-939 era; supported by 4.3 brokers.
pub const ACL_OPERATION_TWO_PHASE_COMMIT : Int = 15
///|
/// ACL permission types (AclPermissionType byte values).
pub const ACL_PERMISSION_UNKNOWN : Int = 0
///|
pub const ACL_PERMISSION_ANY : Int = 1
///|
pub const ACL_PERMISSION_DENY : Int = 2
///|
pub const ACL_PERMISSION_ALLOW : Int = 3
///|
/// One ACL binding: who may do what with which resource from where.
pub(all) struct AclBinding {
resource_type : Int
resource_name : String
pattern_type : Int
principal : String
host : String
operation : Int
permission_type : Int
} derive(@debug.Debug)
///|
/// One ACL to create.
pub(all) struct AclCreation {
resource_type : Int
resource_name : String
pattern_type : Int
principal : String
host : String
operation : Int
permission_type : Int
} derive(@debug.Debug)
///|
/// Filter describing which ACLs to list or delete: a None field matches
/// anything; ANY/MATCH enum values widen the match.
pub(all) struct AclFilter {
resource_type : Int
resource_name : String?
pattern_type : Int
principal : String?
host : String?
operation : Int
permission_type : Int
} derive(@debug.Debug)
///|
fn encode_acl_creation(body : @buf.Encoder, creation : AclCreation) -> Unit {
body.write_i8(creation.resource_type)
body.write_compact_string(creation.resource_name)
body.write_i8(creation.pattern_type)
body.write_compact_string(creation.principal)
body.write_compact_string(creation.host)
body.write_i8(creation.operation)
body.write_i8(creation.permission_type)
body.write_tag_buffer()
}
///|
fn encode_acl_filter(body : @buf.Encoder, filter : AclFilter) -> Unit {
body.write_i8(filter.resource_type)
body.write_compact_nullable_string(filter.resource_name)
body.write_i8(filter.pattern_type)
body.write_compact_nullable_string(filter.principal)
body.write_compact_nullable_string(filter.host)
body.write_i8(filter.operation)
body.write_i8(filter.permission_type)
body.write_tag_buffer()
}
///|
fn decode_acl_binding(d : @buf.Decoder) -> AclBinding raise {
let resource_type = d.read_i8()
let resource_name = d.read_compact_string()
let pattern_type = d.read_i8()
let principal = d.read_compact_string()
let host = d.read_compact_string()
let operation = d.read_i8()
let permission_type = d.read_i8()
d.skip_tag_buffer()
{
resource_type,
resource_name,
pattern_type,
principal,
host,
operation,
permission_type,
}
}
///|
/// One ACL creation result.
pub struct CreateAclsResult {
error_code : Int
error_message : String?
} derive(@debug.Debug)
///|
/// Encode a CreateAcls v2/v3 request body.
pub fn encode_create_acls_request(creations : Array[AclCreation]) -> Bytes {
let body = @buf.Encoder::new()
body.write_compact_len(creations.length())
for creation in creations {
encode_acl_creation(body, creation)
}
body.write_tag_buffer()
body.to_bytes()
}
///|
/// Decode a CreateAcls v2/v3 response body: one result per creation.
pub fn decode_create_acls_response(
d : @buf.Decoder,
) -> (Array[CreateAclsResult], Int) raise {
let throttle = d.read_i32()
let out : Array[CreateAclsResult] = []
let n = d.read_compact_len()
for _ in 0.. Array[CreateAclsResult] {
let d = self.request(
API_CREATE_ACLS,
self.api_version(API_CREATE_ACLS),
encode_create_acls_request(creations),
timeout_ms~,
)
let (results, throttle) = decode_create_acls_response(d)
self.note_throttle(throttle)
results
}
///|
/// One ACL of a described resource.
pub(all) struct AclDescription {
principal : String
host : String
operation : Int
permission_type : Int
} derive(@debug.Debug)
///|
pub struct DescribedAclResource {
resource_type : Int
resource_name : String
pattern_type : Int
acls : Array[AclDescription]
} derive(@debug.Debug)
///|
pub struct DescribeAclsResult {
error_code : Int
error_message : String?
resources : Array[DescribedAclResource]
} derive(@debug.Debug)
///|
/// Encode a DescribeAcls v2/v3 request body.
pub fn encode_describe_acls_request(filter : AclFilter) -> Bytes {
let body = @buf.Encoder::new()
encode_acl_filter(body, filter)
body.write_tag_buffer()
body.to_bytes()
}
///|
/// Decode a DescribeAcls v2/v3 response body.
pub fn decode_describe_acls_response(
d : @buf.Decoder,
) -> (DescribeAclsResult, Int) raise {
let throttle = d.read_i32()
let error_code = d.read_i16()
let error_message = d.read_compact_nullable_string()
let resources : Array[DescribedAclResource] = []
let n = d.read_compact_len()
for _ in 0.. DescribeAclsResult {
let d = self.request(
API_DESCRIBE_ACLS,
self.api_version(API_DESCRIBE_ACLS),
encode_describe_acls_request(filter),
timeout_ms~,
)
let (result, throttle) = decode_describe_acls_response(d)
self.note_throttle(throttle)
result
}
///|
/// One ACL a delete filter matched, with its own deletion error.
pub struct DeletedAcl {
error_code : Int
error_message : String?
binding : AclBinding
} derive(@debug.Debug)
///|
pub struct DeleteAclsFilterResult {
error_code : Int
error_message : String?
matching_acls : Array[DeletedAcl]
} derive(@debug.Debug)
///|
/// Encode a DeleteAcls v2/v3 request body.
pub fn encode_delete_acls_request(filters : Array[AclFilter]) -> Bytes {
let body = @buf.Encoder::new()
body.write_compact_len(filters.length())
for filter in filters {
encode_acl_filter(body, filter)
}
body.write_tag_buffer()
body.to_bytes()
}
///|
/// Decode a DeleteAcls v2/v3 response body: one result per filter, each
/// carrying the ACLs it matched and deleted.
pub fn decode_delete_acls_response(
d : @buf.Decoder,
) -> (Array[DeleteAclsFilterResult], Int) raise {
let throttle = d.read_i32()
let out : Array[DeleteAclsFilterResult] = []
let n = d.read_compact_len()
for _ in 0.. Array[DeleteAclsFilterResult] {
let d = self.request(
API_DELETE_ACLS,
self.api_version(API_DELETE_ACLS),
encode_delete_acls_request(filters),
timeout_ms~,
)
let (results, throttle) = decode_delete_acls_response(d)
self.note_throttle(throttle)
results
}
///|
/// List the ACLs matching the filter (an all-ANY filter with null
/// fields lists everything). Retriable errors re-issue under the policy.
pub async fn Admin::describe_acls(
self : Admin,
filter : AclFilter,
) -> DescribeAclsResult {
self.with_retries(
fn(result : DescribeAclsResult) { admin_error_retriable(result.error_code) },
async fn() {
let conn = self.any_conn()
conn.describe_acls(filter, timeout_ms=self.request_timeout_ms)
},
)
}
///|
/// Create ACL bindings; one result per creation.
pub async fn Admin::create_acls(
self : Admin,
creations : Array[AclCreation],
) -> Array[CreateAclsResult] {
self.with_retries(
fn(results : Array[CreateAclsResult]) {
let mut retry = false
for result in results {
if admin_error_retriable(result.error_code) {
retry = true
}
}
retry
},
async fn() {
let conn = self.any_conn()
conn.create_acls(creations, timeout_ms=self.request_timeout_ms)
},
)
}
///|
/// Delete every ACL matching each filter; one result per filter with
/// the matched bindings.
pub async fn Admin::delete_acls(
self : Admin,
filters : Array[AclFilter],
) -> Array[DeleteAclsFilterResult] {
self.with_retries(
fn(results : Array[DeleteAclsFilterResult]) {
let mut retry = false
for result in results {
if admin_error_retriable(result.error_code) {
retry = true
}
for acl in result.matching_acls {
if admin_error_retriable(acl.error_code) {
retry = true
}
}
}
retry
},
async fn() {
let conn = self.any_conn()
conn.delete_acls(filters, timeout_ms=self.request_timeout_ms)
},
)
}