rpt.SetStatus(bucket, fileName, nil)
		}
	}

	enc := func(b []byte) *string {
		if b == nil {
			return nil
		}
		v := base64.StdEncoding.EncodeToString(b)
		return &v
	}

	for bucket, meta := range bucketMap {
		err := globalBucketMetadataSys.save(ctx, *meta)
		if err != nil {
			rpt.SetStatus(bucket, "", err)
			continue
		}

		}

		// close when we know this is a network error.
		target.conn.Close()
	}

	conn, err = amqp091.Dial(target.args.URL.String())
	if err != nil {
		if xnet.IsConnRefusedErr(err) {
			return nil, nil, store.ErrNotConnected
		}
		return nil, nil, err
	}

	ch, err = conn.Channel()
	if err != nil {
		return nil, nil, err
	}

	target.conn = conn

	if err != nil {
		return nil, err
	}

	for _, pod := range pods {
		fw, err := kubeClient.NewPortForwarder(pod.Name, pod.Namespace, "localhost", 0, centralOpts.XdsPodPort)
		if err != nil {
			return nil, err
		}
		err = fw.Start()
		if err != nil {
			return nil, err
		}
		defer fw.Close()
		xdsOpts.Xds = fw.Address()

		if err != nil {
			return nil, err
		}
		return objectKey[:], nil
	case crypto.SSEC:
		sealedKey, err := crypto.SSEC.ParseMetadata(metadata)
		if err != nil {
			return nil, err
		}
		var objectKey crypto.ObjectKey
		if err = objectKey.Unseal(key, sealedKey, crypto.SSEC.String(), bucket, object); err != nil {
			return nil, err
		}
		return objectKey[:], nil
	default:
		return nil, errObjectTampered

	var rights []byte
	if fd != nil {
		rights = unix.UnixRights(*fd)
	}
	err := z.u.SetWriteDeadline(time.Now().Add(readWriteDeadline))
	if err != nil {
		return nil, err
	}

	_, _, err = z.u.WriteMsgUnix(data, rights, nil)
	if err != nil {
		return nil, err
	}

	// wait for ack
	return z.readMessage(readWriteDeadline)
}

	v := BatchJobReplicateCredentials{}
	bts, err := v.MarshalMsg(nil)
	if err != nil {
		t.Fatal(err)
	}
	left, err := v.UnmarshalMsg(bts)
	if err != nil {
		t.Fatal(err)
	}
	if len(left) > 0 {
		t.Errorf("%d bytes left over after UnmarshalMsg(): %q", len(left), left)
	}

	left, err = msgp.Skip(bts)
	if err != nil {
		t.Fatal(err)
	}
	if len(left) > 0 {

			port, err := parsePort(portStr)
			if err != nil {
				return nil, err
			}
			ports = append(ports, int(port))
		}
	}
	return ports, nil
}

func validatePortList(ports string) error {
	if _, err := parsePorts(ports); err != nil {
		return fmt.Errorf("portList %q invalid: %v", ports, err)
	}
	return nil
}

func validatePortListWithWildcard(ports string) error {

	v := BatchJobKV{}
	bts, err := v.MarshalMsg(nil)
	if err != nil {
		t.Fatal(err)
	}
	left, err := v.UnmarshalMsg(bts)
	if err != nil {
		t.Fatal(err)
	}
	if len(left) > 0 {
		t.Errorf("%d bytes left over after UnmarshalMsg(): %q", len(left), left)
	}

	left, err = msgp.Skip(bts)
	if err != nil {
		t.Fatal(err)
	}
	if len(left) > 0 {

	if err != nil {
		return nil, nil, err
	}

	version, err := formatGetBackendErasureVersion(formatData)
	if err != nil {
		return nil, nil, fmt.Errorf("Drive %s: %w", export, err)
	}

	migrate := func(formatPath string, formatData []byte) ([]byte, fs.FileInfo, error) {
		if err = os.WriteFile(formatPath, formatData, 0o666); err != nil {
			return nil, nil, err
		}

	if err != nil {
		return nil, fmt.Errorf("error %s: %s", err, manifest)
	}
	return NewObjectSet(getAllIstioObjects()), nil
}

// fakeApplyExtraResources applies any extra resources for the given test name.
func fakeApplyExtraResources(inFile string) error {
	reconciler, err := helmreconciler.NewHelmReconciler(testClient, nil, nil, nil)
	if err != nil {
		return err
	}