configLogIf(ctx, fmt.Errorf("Unable to initialize remote webhook DNS config %w", err))
	}
	if err == nil && dnsURL != "" {
		bootstrapTraceMsg("initialize remote bucket DNS store")
		globalDNSConfig, err = dns.NewOperatorDNS(dnsURL,
			dns.Authentication(dnsUser, dnsPass),
			dns.RootCAs(globalRootCAs))
		if err != nil {
			configLogIf(ctx, fmt.Errorf("Unable to initialize remote webhook DNS config %w", err))
		}
	}

// expand and then we continue to rebalance.
func (z *erasureServerPools) updateRebalanceStats(ctx context.Context) error {
	var ok bool
	for i := range z.serverPools {
		if z.findIndex(i) == -1 {
			// Also ensure to initialize rebalanceStats to indicate
			// its a new pool that can receive rebalanced data.
			z.rebalMeta.PoolStats = append(z.rebalMeta.PoolStats, &rebalanceStats{})
			ok = true
		}
	}
	if ok {

		queueStore = store.NewQueueStore[event.Event](queueDir, args.QueueLimit, event.StoreExtension)
		if err := queueStore.Open(); err != nil {
			return nil, fmt.Errorf("unable to initialize the queue store of Kafka `%s`: %w", id, err)
		}
	}

	target := &KafkaTarget{
		id:         event.TargetID{ID: id, Name: "kafka"},
		args:       args,
		store:      queueStore,
		loggerOnce: loggerOnce,

func mustNewRequest(method string, urlStr string, contentLength int64, body io.ReadSeeker, t *testing.T) *http.Request {
	req, err := newTestRequest(method, urlStr, contentLength, body)
	if err != nil {
		t.Fatalf("Unable to initialize new http request %s", err)
	}
	return req
}

// This is similar to mustNewRequest but additionally the request
// is signed with AWS Signature V4, fails if not able to do so.

		queueStore = store.NewQueueStore[event.Event](queueDir, args.QueueLimit, event.StoreExtension)
		if err := queueStore.Open(); err != nil {
			return nil, fmt.Errorf("unable to initialize the queue store of NATS `%s`: %w", id, err)
		}
	}

	target := &NATSTarget{
		id:         event.TargetID{ID: id, Name: "nats"},
		args:       args,
		loggerOnce: loggerOnce,
		store:      queueStore,

	case authTypeStreamingSigned, authTypeStreamingSignedTrailer:
		// Initialize stream signature verifier.
		reader, s3Error = newSignV4ChunkedReader(r, rAuthType == authTypeStreamingSignedTrailer)
		if s3Error != ErrNone {
			writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
			return
		}
	case authTypeStreamingUnsignedTrailer:
		// Initialize stream signature verifier.

	return types.TargetStats{
		TotalMessages:  atomic.LoadInt64(&h.totalMessages),
		FailedMessages: atomic.LoadInt64(&h.failedMessages),
		QueueLength:    queueLength,
	}
}

// Init initialize kafka target
func (h *Target) Init(ctx context.Context) error {
	if !h.kconfig.Enabled {
		return nil
	}
	if err := h.validate(); err != nil {
		return err
	}
	if h.kconfig.QueueDir != "" {

	HTTP         map[string]http.Config  `json:"http"`
	AuditWebhook map[string]http.Config  `json:"audit"`
	AuditKafka   map[string]kafka.Config `json:"audit_kafka"`
}

// NewConfig - initialize new logger config.
func NewConfig() Config {
	cfg := Config{
		// Console logging is on by default
		Console: Console{
			Enabled: true,
		},
		HTTP:         make(map[string]http.Config),

		Scheme:   ep.Scheme,
		Online:   true,
	}
	sys.hMutex.Unlock()
}

// newHCClient initializes an anonymous client for performing health check on the remote endpoints
func newHCClient() *madmin.AnonymousClient {
	clnt, e := madmin.NewAnonymousClientNoEndpoint()
	if e != nil {
		bugLogIf(GlobalContext, errors.New("Unable to initialize health check client"))
		return nil
	}
	clnt.SetCustomTransport(globalRemoteTargetTransport)

func TestDeleteObjectsVersioned(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	obj, fsDirs, err := prepareErasure(ctx, 16)
	if err != nil {
		t.Fatal("Unable to initialize 'Erasure' object layer.", err)
	}
	// Remove all dirs.
	for _, dir := range fsDirs {
		defer os.RemoveAll(dir)
	}

	type testCaseType struct {
		bucket string
		object string
	}