kadm: use cached metadata where possible

As a follow up to #800, we convert kadm to using cached metadata
everywhere except for the actual Metadata function.

With a quick local test using `rpk group describe`, this brings the
prior 4 metadata requests down to 1.

pkg/kadm/go.mod

@@ -4,6 +4,8 @@ go 1.21
 
 toolchain go1.22.0
 
+replace github.com/twmb/franz-go => ../../
+
 require (
 	github.com/twmb/franz-go v1.18.1
 	github.com/twmb/franz-go/pkg/kmsg v1.9.0

pkg/kadm/metadata.go

@@ -200,7 +200,7 @@ func int32s(is []int32) []int32 {
 // ListBrokers issues a metadata request and returns BrokerDetails. This
 // returns an error if the request fails to be issued, or an *AuthError.
 func (cl *Client) ListBrokers(ctx context.Context) (BrokerDetails, error) {
-	m, err := cl.Metadata(ctx)
+	m, err := cl.BrokerMetadata(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -212,7 +212,7 @@ func (cl *Client) ListBrokers(ctx context.Context) (BrokerDetails, error) {
 //
 // This returns an error if the request fails to be issued, or an *AuthErr.
 func (cl *Client) BrokerMetadata(ctx context.Context) (Metadata, error) {
-	return cl.metadata(ctx, true, nil)
+	return cl.metadata(ctx, true, nil, true)
 }
 
 // Metadata issues a metadata request and returns it. Specific topics to
@@ -224,10 +224,10 @@ func (cl *Client) Metadata(
 	ctx context.Context,
 	topics ...string,
 ) (Metadata, error) {
-	return cl.metadata(ctx, false, topics)
+	return cl.metadata(ctx, false, topics, false)
 }
 
-func (cl *Client) metadata(ctx context.Context, noTopics bool, topics []string) (Metadata, error) {
+func (cl *Client) metadata(ctx context.Context, noTopics bool, topics []string, useCache bool) (Metadata, error) {
 	req := kmsg.NewPtrMetadataRequest()
 	req.IncludeClusterAuthorizedOperations = true
 	req.IncludeTopicAuthorizedOperations = true
@@ -239,7 +239,13 @@ func (cl *Client) metadata(ctx context.Context, noTopics bool, topics []string)
 	if noTopics {
 		req.Topics = []kmsg.MetadataRequestTopic{}
 	}
-	resp, err := req.RequestWith(ctx, cl.cl)
+	var resp *kmsg.MetadataResponse
+	var err error
+	if useCache {
+		resp, err = cl.cl.RequestCachedMetadata(ctx, req, -1)
+	} else {
+		resp, err = req.RequestWith(ctx, cl.cl)
+	}
 	if err != nil {
 		return Metadata{}, err
 	}

pkg/kadm/topics.go

@@ -34,7 +34,7 @@ func (cl *Client) ListTopicsWithInternal(
 	ctx context.Context,
 	topics ...string,
 ) (TopicDetails, error) {
-	m, err := cl.Metadata(ctx, topics...)
+	m, err := cl.metadata(ctx, false, topics, true)
 	if err != nil {
 		return nil, err
 	}

pkg/kadm/txn.go

@@ -228,13 +228,13 @@ func (ds DescribedProducersTopics) EachProducer(fn func(DescribedProducer)) {
 // This may return *ShardErrors or *AuthError.
 func (cl *Client) DescribeProducers(ctx context.Context, s TopicsSet) (DescribedProducersTopics, error) {
 	if len(s) == 0 {
-		m, err := cl.Metadata(ctx)
+		m, err := cl.metadata(ctx, false, nil, true)
 		if err != nil {
 			return nil, err
 		}
 		s = m.Topics.TopicsSet()
 	} else if e := s.EmptyTopics(); len(e) > 0 {
-		m, err := cl.Metadata(ctx, e...)
+		m, err := cl.metadata(ctx, false, e, true)
 		if err != nil {
 			return nil, err
 		}