rs.Completed = true
	case replication.Pending:
		rs.Pending = true
	case replication.Failed:
		rs.Failed = true
		rs.Err = err
	}
}

// GetAll returns replication metrics for all buckets at once.
func (r *ReplicationStats) GetAll() map[string]BucketReplicationStats {
	if r == nil {
		return map[string]BucketReplicationStats{}
	}

	r.RLock()

func (s *peerRESTServer) GetAllBucketStatsHandler(mss *grid.MSS) (*BucketStatsMap, *grid.RemoteErr) {
	replicationStats := globalReplicationStats.GetAll()
	bucketStatsMap := make(map[string]BucketStats, len(replicationStats))
	for k, v := range replicationStats {
		bucketStatsMap[k] = BucketStats{
			ReplicationStats: v,

		return ErrBatchFull
	}

	if _, ok := b.items[key]; !ok {
		b.keys = append(b.keys, key)
	}
	b.items[key] = item

	return nil
}

// GetAll fetches the items and resets the batch
// Returned items are not referenced by the batch
func (b *Batch[K, T]) GetAll() (orderedKeys []K, orderedItems []T, err error) {
	b.Lock()
	defer b.Unlock()

	orderedKeys = append([]K(nil), b.keys...)
	for _, key := range orderedKeys {

			wantOk:  false,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			v := Value{
				value: tt.fields.value,
			}
			gotVal, gotOk := v.bytesToBool()
			if gotVal != tt.wantVal {
				t.Errorf("bytesToBool() gotVal = %v, want %v", gotVal, tt.wantVal)
			}
			if gotOk != tt.wantOk {
				t.Errorf("bytesToBool() gotOk = %v, want %v", gotOk, tt.wantOk)
			}
		})
	}

	if key.IsLast || target.batch.IsFull() {
		return target.commitBatch()
	}
	return nil
}

func (target *KafkaTarget) commitBatch() error {
	keys, msgs, err := target.batch.GetAll()
	if err != nil {
		return err
	}
	if err = target.producer.SendMessages(msgs); err != nil {
		if isKafkaConnErr(err) {
			return store.ErrNotConnected
		}
		return err
	}

		t.Fatal("Expected batch.IsFull to be true but got false")
	}
	batchLen := batch.Len()
	if batchLen != int(limit) {
		t.Fatalf("expected batch length to be %v but got %v", limit, batchLen)
	}
	keys, items, err := batch.GetAll()
	if err != nil {
		t.Fatalf("unable to get the items from the batch; %v", err)
	}
	if len(items) != int(limit) {
		t.Fatalf("Expected length of the batch items to be %v but got %v", limit, len(items))
	}

	}
	// Generate a new UUID for the key.
	key, err := uuid.NewRandom()
	if err != nil {
		return err
	}
	return store.write(key.String(), item)
}

// GetRaw - gets an item from the store.
func (store *QueueStore[I]) GetRaw(key string) (raw []byte, err error) {
	store.RLock()

	defer func(store *QueueStore[I]) {
		store.RUnlock()
		if err != nil {
			// Upon error we remove the entry.

	Name() string
	SendFromStore(key Key) error
}

// Store - Used to persist items.
type Store[I any] interface {
	Put(item I) error
	PutMultiple(item []I) error
	Get(key string) (I, error)
	GetRaw(key string) ([]byte, error)
	Len() int
	List() ([]string, error)
	Del(key string) error
	DelList(key []string) error
	Open() error
	Delete() error
	Extension() string
}

			if err != nil {
				return err
			}
			retval := inject.IntoResourceFile(injector, templs, vc, rev, meshConfig,
				reader, writer, func(warning string) {
					warnings = append(warnings, warning)
				})
			if len(warnings) > 0 {
				fmt.Fprintln(c.ErrOrStderr())
			}
			for _, warning := range warnings {
				fmt.Fprintln(c.ErrOrStderr(), warning)
			}
			return retval
		},

		if nErr.Err != nil {
			peersLogOnceIf(logger.SetReqInfo(ctx, reqInfo), nErr.Err, nErr.Host.String())
		}
	}

	replicationStatsList := globalReplicationStats.GetAll()
	bucketStatsMap := BucketStatsMap{
		Stats:     make(map[string]BucketStats, len(replicationStatsList)),
		Timestamp: UTCNow(),
	}
	for k, replicationStats := range replicationStatsList {