func (cl *Client) fetchTopicMetadata(all bool, reqTopics []string) (map[string]*topicPartitionsData, error) {
	_, meta, err := cl.fetchMetadataForTopics(cl.ctx, all, reqTopics)
	if err != nil {
		return nil, err
	}
	topics := make(map[string]*topicPartitionsData, len(meta.Topics))
	useLeaderEpoch := cl.supportsOffsetForLeaderEpoch()
	for i := range meta.Topics {
		topicMeta := &meta.Topics[i]
		if topicMeta.Topic == nil {
			cl.cfg.logger.Log(LogLevelWarn, "metadata response contained nil topic name even though we did not request with topic IDs, skipping")
			continue
		}
		topic := *topicMeta.Topic
		parts := &topicPartitionsData{
			loadErr:            kerr.ErrorForCode(topicMeta.ErrorCode),
			isInternal:         topicMeta.IsInternal,
			partitions:         make([]*topicPartition, 0, len(topicMeta.Partitions)),
			writablePartitions: make([]*topicPartition, 0, len(topicMeta.Partitions)),
		}
		topics[topic] = parts
		if parts.loadErr != nil {
			continue
		}
		if len(topic) > 249 {
			parts.loadErr = fmt.Errorf("invalid long topic name of (len %d) greater than max allowed 249", len(topic))
			continue
		}
		sort.Slice(topicMeta.Partitions, func(i, j int) bool {
			return topicMeta.Partitions[i].Partition < topicMeta.Partitions[j].Partition
		})
		for i := range topicMeta.Partitions {
			if got := topicMeta.Partitions[i].Partition; got != int32(i) {
				parts.loadErr = fmt.Errorf("kafka did not reply with a comprensive set of partitions for a topic; we expected partition %d but saw %d", i, got)
				break
			}
		}
		if parts.loadErr != nil {
			continue
		}
		for i := range topicMeta.Partitions {
			partMeta := &topicMeta.Partitions[i]
			leaderEpoch := partMeta.LeaderEpoch
			if meta.Version < 7 || !useLeaderEpoch {
				leaderEpoch = -1
			}
			p := &topicPartition{
				loadErr: kerr.ErrorForCode(partMeta.ErrorCode),
				topicPartitionData: topicPartitionData{
					leader:      partMeta.Leader,
					leaderEpoch: leaderEpoch,
				},
				records: &recBuf{
					cl: cl,
					topic:     topic,
					partition: partMeta.Partition,
					maxRecordBatchBytes: cl.maxRecordBatchBytesForTopic(topic),
					recBufsIdx: -1,
					failing:    partMeta.ErrorCode != 0,
				},
				cursor: &cursor{
					topic:       topic,
					topicID:     topicMeta.TopicID,
					partition:   partMeta.Partition,
					keepControl: cl.cfg.keepControl,
					cursorsIdx:  -1,
					cursorOffset: cursorOffset{
						offset:            -1,  
						lastConsumedEpoch: -1,  
					},
				},
			}
			if p.loadErr != nil {
				p.leader = unknownSeedID(0)
			}
			p.cursor.topicPartitionData = p.topicPartitionData
			p.records.topicPartitionData = p.topicPartitionData
			cl.sinksAndSourcesMu.Lock()
			sns, exists := cl.sinksAndSources[p.leader]
			if !exists {
				sns = sinkAndSource{
					sink:   cl.newSink(p.leader),
					source: cl.newSource(p.leader),
				}
				cl.sinksAndSources[p.leader] = sns
			}
			cl.sinksAndSourcesMu.Unlock()
			p.records.sink = sns.sink
			p.cursor.source = sns.source
			parts.partitions = append(parts.partitions, p)
			if p.loadErr == nil {
				parts.writablePartitions = append(parts.writablePartitions, p)
			}
		}
	}
	return topics, nil
}
fetchTopicMetadata fetches metadata for all reqTopics and returns new
topicPartitionsData for each topic.
package kgo
import (
	"context"
	"errors"
	"fmt"
	"sort"
	"sync"
	"time"
	"github.com/twmb/franz-go/pkg/kerr"
)
const minRefreshTrigger = 5 * time.Second / 2
type metawait struct {
	mu         sync.Mutex
	c          *sync.Cond
	lastUpdate time.Time
}
func (m *metawait) init() { m.c = sync.NewCond(&m.mu) }
func (m *metawait) signal() {
	m.mu.Lock()
	m.lastUpdate = time.Now()
	m.mu.Unlock()
	m.c.Broadcast()
}
func (cl *Client) waitmeta(ctx context.Context, wait time.Duration, why string) {
	now := time.Now()
	cl.metawait.mu.Lock()
	if now.Sub(cl.metawait.lastUpdate) < minRefreshTrigger {
		cl.metawait.mu.Unlock()
		return
	}
	cl.metawait.mu.Unlock()
	cl.triggerUpdateMetadataNow(why)
	quit := false
	done := make(chan struct{})
	timeout := time.NewTimer(wait)
	defer timeout.Stop()
	go func() {
		defer close(done)
		cl.metawait.mu.Lock()
		defer cl.metawait.mu.Unlock()
		for !quit {
			if now.Sub(cl.metawait.lastUpdate) < minRefreshTrigger {
				return
			}
			cl.metawait.c.Wait()
		}
	}()
	select {
	case <-done:
		return
	case <-timeout.C:
	case <-ctx.Done():
	case <-cl.ctx.Done():
	}
	cl.metawait.mu.Lock()
	quit = true
	cl.metawait.mu.Unlock()
	cl.metawait.c.Broadcast()
}
func (cl *Client) triggerUpdateMetadata(must bool, why string) bool {
	if !must {
		cl.metawait.mu.Lock()
		defer cl.metawait.mu.Unlock()
		if time.Since(cl.metawait.lastUpdate) < minRefreshTrigger {
			return false
		}
	}
	select {
	case cl.updateMetadataCh <- why:
	default:
	}
	return true
}
func (cl *Client) triggerUpdateMetadataNow(why string) {
	select {
	case cl.updateMetadataNowCh <- why:
	default:
	}
}
func (cl *Client) updateMetadataLoop() {
	defer close(cl.metadone)
	var consecutiveErrors int
	var lastAt time.Time
	ticker := time.NewTicker(cl.cfg.metadataMaxAge)
	defer ticker.Stop()
	for {
		var now bool
		select {
		case <-cl.ctx.Done():
			return
		case <-ticker.C:
		case why := <-cl.updateMetadataCh:
			cl.cfg.logger.Log(LogLevelInfo, "metadata update triggered", "why", why)
		case why := <-cl.updateMetadataNowCh:
			cl.cfg.logger.Log(LogLevelInfo, "immediate metadata update triggered", "why", why)
			now = true
		}
		var nowTries int
	start:
		nowTries++
		if !now {
			if wait := cl.cfg.metadataMinAge - time.Since(lastAt); wait > 0 {
				timer := time.NewTimer(wait)
				select {
				case <-cl.ctx.Done():
					timer.Stop()
					return
				case why := <-cl.updateMetadataNowCh:
					timer.Stop()
					cl.cfg.logger.Log(LogLevelInfo, "immediate metadata update triggered, bypassing normal wait", "why", why)
				case <-timer.C:
				}
			}
		}
		time.Sleep(time.Until(lastAt.Add(10 * time.Millisecond)))
	out:
		for {
			select {
			case <-cl.updateMetadataCh:
			case <-cl.updateMetadataNowCh:
			default:
				break out
			}
		}
		again, err := cl.updateMetadata()
		if again || err != nil {
			if now && nowTries < 3 {
				goto start
			}
			if err != nil {
				cl.triggerUpdateMetadata(true, fmt.Sprintf("re-updating metadata due to err: %s", err))
			} else {
				cl.triggerUpdateMetadata(true, "re-updating metadata due inner topic or partition error")
			}
		}
		if err == nil {
			lastAt = time.Now()
			consecutiveErrors = 0
			continue
		}
		consecutiveErrors++
		after := time.NewTimer(cl.cfg.retryBackoff(consecutiveErrors))
		select {
		case <-cl.ctx.Done():
			after.Stop()
			return
		case <-after.C:
		}
	}
}
func (cl *Client) updateMetadata() (needsRetry bool, err error) {
	defer cl.metawait.signal()
	defer cl.consumer.doOnMetadataUpdate()
	var (
		tpsProducerLoad = cl.producer.topics.load()
		tpsConsumer     *topicsPartitions
		all             = cl.cfg.regex
		reqTopics       []string
	)
	c := &cl.consumer
	switch {
	case c.d != nil:
		tpsConsumer = c.d.tps
	case c.g != nil:
		tpsConsumer = c.g.tps
	}
	if !all {
		reqTopicsSet := make(map[string]struct{})
		for _, m := range []map[string]*topicPartitions{
			tpsProducerLoad,
			tpsConsumer.load(),
		} {
			for topic := range m {
				reqTopicsSet[topic] = struct{}{}
			}
		}
		reqTopics = make([]string, 0, len(reqTopicsSet))
		for topic := range reqTopicsSet {
			reqTopics = append(reqTopics, topic)
		}
	}
	latest, err := cl.fetchTopicMetadata(all, reqTopics)
	if err != nil {
		cl.bumpMetadataFailForTopics(  
			tpsProducerLoad,
			err,
		)
		return true, err
	}
	tpsConsumerLoad := tpsConsumer.load()
	if all && len(latest) > 0 {
		allTopics := make([]string, 0, len(latest))
		for topic := range latest {
			allTopics = append(allTopics, topic)
		}
		tpsConsumerLoad = tpsConsumer.ensureTopics(allTopics)
		defer tpsConsumer.storeData(tpsConsumerLoad)
	}
	var (
		consumerSessionStopped bool
		reloadOffsets          listOrEpochLoads
		tpsPrior               *topicsPartitions
	)
	stopConsumerSession := func() {
		if consumerSessionStopped {
			return
		}
		consumerSessionStopped = true
		loads, tps := cl.consumer.stopSession()
		reloadOffsets.mergeFrom(loads)
		tpsPrior = tps
	}
	defer func() {
		if consumerSessionStopped {
			session := cl.consumer.startNewSession(tpsPrior)
			defer session.decWorker()
			reloadOffsets.loadWithSession(session, "resuming reload offsets after session stopped for cursor migrating in metadata")
		}
	}()
	var missingProduceTopics []string
	for _, m := range []struct {
		priors    map[string]*topicPartitions
		isProduce bool
	}{
		{tpsProducerLoad, true},
		{tpsConsumerLoad, false},
	} {
		for topic, priorParts := range m.priors {
			newParts, exists := latest[topic]
			if !exists {
				if m.isProduce {
					missingProduceTopics = append(missingProduceTopics, topic)
				}
				continue
			}
			needsRetry = needsRetry || cl.mergeTopicPartitions(
				topic,
				priorParts,
				newParts,
				m.isProduce,
				&reloadOffsets,
				stopConsumerSession,
			)
		}
	}
	if len(missingProduceTopics) > 0 {
		cl.bumpMetadataFailForTopics(
			tpsProducerLoad,
			errors.New("metadata request did not return this topic"),
			missingProduceTopics...,
		)
	}
	return needsRetry, nil
}