// If len(Pkgs) == 0 && len(Errs) == 0, the pattern is well-formed but did not
	// match any packages.
}

// NewMatch returns a Match describing the given pattern,
// without resolving its packages or errors.
func NewMatch(pattern string) *Match {
	return &Match{pattern: pattern}
}

// Pattern returns the pattern to be matched.
func (m *Match) Pattern() string { return m.pattern }

// against the standard library (std and cmd) in GOROOT/src.
func MatchInModule(ctx context.Context, pattern string, m module.Version, tags map[string]bool) *search.Match {
	match := search.NewMatch(pattern)
	if m == (module.Version{}) {
		matchPackages(ctx, match, tags, includeStd, nil)
	}

	LoadModFile(ctx) // Sets Target, needed by fetch and matchPackages.

	if !match.IsLiteral() {

		}
		match = func(mod module.Version, roots []string, isLocal bool) *search.Match {
			m := search.NewMatch(pattern)
			matchPackages(ctx, m, imports.AnyTags(), omitStd, []module.Version{mod})
			return m
		}
	} else {
		match = func(mod module.Version, roots []string, isLocal bool) *search.Match {
			m := search.NewMatch(pattern)
			prefix := mod.Path
			if MainModules.Contains(mod.Path) {

	}

	patterns = search.CleanPatterns(patterns)
	matches = make([]*search.Match, 0, len(patterns))
	allPatternIsRoot := false
	for _, pattern := range patterns {
		matches = append(matches, search.NewMatch(pattern))
		if pattern == "all" {
			allPatternIsRoot = true
		}
	}

	updateMatches := func(rs *Requirements, ld *loader) {
		for _, m := range matches {
			switch {
			case m.IsLocal():

func (b *Batch[K, T]) IsFull() bool {
	b.Lock()
	defer b.Unlock()

	return b.isFull()
}

func (b *Batch[K, T]) isFull() bool {
	return len(b.items) >= int(b.limit)
}

// NewBatch creates a new batch
func NewBatch[K key, T any](limit uint32) *Batch[K, T] {
	return &Batch[K, T]{
		keys:  make([]K, 0, limit),
		items: make(map[K]T, limit),
		limit: limit,
	}

// along with this program.  If not, see <http://www.gnu.org/licenses/>.

package store

import (
	"errors"
	"sync"
	"testing"
)

func TestBatch(t *testing.T) {
	var limit uint32 = 100
	batch := NewBatch[int, int](limit)
	for i := 0; i < int(limit); i++ {
		if err := batch.Add(i, i); err != nil {
			t.Fatalf("failed to add %v; %v", i, err)
		}
		if _, ok := batch.GetByKey(i); !ok {

	// Later, you can issue a watch with the REST apis list.RV and end up earlier than the storage cacher.
	// The general working model is that if you get a "resourceVersion too old" message, you re-list and rewatch.
	// For this test, we'll actually cycle, "list/watch/create/delete" until we get an RV from list that observes the create and not an error.

		args:       args,
		store:      queueStore,
		loggerOnce: loggerOnce,
		quitCh:     make(chan struct{}),
	}

	if target.store != nil {
		if args.BatchSize > 1 {
			target.batch = store.NewBatch[string, *sarama.ProducerMessage](args.BatchSize)
		}
		store.StreamItems(target.store, target, target.quitCh, target.loggerOnce)
	}

	return target, nil
}

func isKafkaConnErr(err error) bool {

- Fixes a bug that when there is a "connection refused" error, the reflector's ListAndWatch func will return directly but what expected is that sleep 1 second and rewatch since the specified resourceVersion.
  ([#81634](https://github.com/kubernetes/kubernetes/pull/81634), [@likakuli](https://github.com/likakuli))