[v24.3.x] rpk: improve error messages in security/acl

src/go/rpk/pkg/adminapi/admin.go

@@ -131,6 +131,22 @@ func NewHostClient(fs afero.Fs, p *config.RpkProfile, host string) (*rpadmin.Adm
 	return rpadmin.NewClient(addrs, tc, auth, p.FromCloud)
 }
 
+// TryDecodeMessageFromErr tries to decode the message if it's a
+// rpadmin.HTTPResponseError and logs the full error. Otherwise, it returns
+// the original error string.
+func TryDecodeMessageFromErr(err error) string {
+	if err == nil {
+		return ""
+	}
+	if he := (*rpadmin.HTTPResponseError)(nil); errors.As(err, &he) {
+		zap.L().Sugar().Debugf("got admin API error: %v", strings.TrimSpace(err.Error()))
+		if body, err := he.DecodeGenericErrorBody(); err == nil {
+			return body.Message
+		}
+	}
+	return strings.TrimSpace(err.Error())
+}
+
 // licenseFeatureChecks checks if the user is talking to a cluster that has
 // enterprise features enabled without a license and returns a message with a
 // warning.

src/go/rpk/pkg/cli/security/acl/create.go

@@ -69,6 +69,10 @@ Allow write permissions to user buzz to transactional ID "txn":
 			tw := out.NewTable(headersWithError...)
 			defer tw.Flush()
 			for _, c := range results {
+				errMsg := kafka.ErrMessage(c.Err)
+				if c.ErrMessage != "" {
+					errMsg = fmt.Sprintf("%v: %v", errMsg, c.ErrMessage)
+				}
 				tw.PrintStructFields(aclWithMessage{
 					c.Principal,
 					c.Host,
@@ -77,7 +81,7 @@ Allow write permissions to user buzz to transactional ID "txn":
 					c.Pattern,
 					c.Operation,
 					c.Permission,
-					kafka.ErrMessage(c.Err),
+					errMsg,
 				})
 			}
 		},

src/go/rpk/pkg/cli/security/role/create.go

@@ -46,7 +46,7 @@ flag in the 'rpk security acl create' command.`,
 
 			roleName := args[0]
 			_, err = cl.CreateRole(cmd.Context(), roleName)
-			out.MaybeDie(err, "unable to create role %q: %v", roleName, err)
+			out.MaybeDie(err, "unable to create role %q: %v", roleName, adminapi.TryDecodeMessageFromErr(err))
 
 			if isText, _, s, err := f.Format(createResponse{[]string{roleName}}); !isText {
 				out.MaybeDie(err, "unable to print in the required format %q: %v", f.Kind, err)

tests/rptest/tests/acls_test.py

@@ -307,7 +307,8 @@ def test_invalid_acl_topic_name(self):
 
         acl = [acl for acl in acls if acl.resource_name == resource]
         assert len(acl) == 1, f'Expected match for {resource} not found'
-        assert acl[0].error == 'INVALID_REQUEST'
+        assert 'INVALID_REQUEST' in acl[
+            0].error, f'expected INVALID_REQUEST to be in {acl[0].error}'
 
     '''
     The old config style has use_sasl at the top level, which enables