Prefix          string                 `yaml:"prefix" json:"prefix"`
	NotificationCfg BatchJobNotification   `yaml:"notify" json:"notify"`
	Retry           BatchJobRetry          `yaml:"retry" json:"retry"`
	Rules           []BatchJobExpireFilter `yaml:"rules" json:"rules"`
}

var _ yaml.Unmarshaler = &BatchJobExpire{}

// - filter
// - notify
// - retry
type BatchJobKeyRotateFlags struct {
	Filter BatchKeyRotateFilter `yaml:"filter" json:"filter"`
	Notify BatchJobNotification `yaml:"notify" json:"notify"`
	Retry  BatchJobRetry        `yaml:"retry" json:"retry"`
}

// BatchJobKeyRotateV1 v1 of batch key rotation job
type BatchJobKeyRotateV1 struct {

	return nil
}

// compareConfResult does a string compare of 2 test files.
func compareConfResult(result, expected string, t *testing.T) {
	t.Helper()
	retry.UntilSuccessOrFail(t, func() error {
		return checkResult(result, expected)
	}, retry.Delay(time.Millisecond*10), retry.Timeout(time.Second*3))
}

// checkBinDir verifies the presence/absence of test files.

				return
			}
		case "Notify":
			err = z.Notify.DecodeMsg(dc)
			if err != nil {
				err = msgp.WrapError(err, "Notify")
				return
			}
		case "Retry":
			err = z.Retry.DecodeMsg(dc)
			if err != nil {
				err = msgp.WrapError(err, "Retry")
				return
			}
		default:
			err = dc.Skip()
			if err != nil {
				err = msgp.WrapError(err)
				return
			}
		}
	}
	return
}

	// These settings are needed to ensure that kafka client doesn't hang on brokers
	// refer https://github.com/IBM/sarama/issues/765#issuecomment-254333355
	sconfig.Producer.Retry.Max = 2
	sconfig.Producer.Retry.Backoff = (10 * time.Second)
	sconfig.Producer.Return.Successes = true
	sconfig.Producer.Return.Errors = true
	sconfig.Producer.RequiredAcks = 1
	sconfig.Producer.Timeout = (10 * time.Second)

	// These settings are needed to ensure that kafka client doesn't hang on brokers
	// refer https://github.com/IBM/sarama/issues/765#issuecomment-254333355
	config.Producer.Retry.Max = 2
	config.Producer.Retry.Backoff = (1 * time.Second)
	config.Producer.Return.Successes = true
	config.Producer.Return.Errors = true
	config.Producer.RequiredAcks = 1
	config.Producer.Timeout = (5 * time.Second)

	go func() {
		for {
			client.doListen(ctx, listenCh, v)
			select {
			case <-ctx.Done():
				return
			default:
				// There was error in the REST request, retry after sometime as probably the peer is down.
				time.Sleep(5 * time.Second)
			}
		}
	}()
}

// Trace - send http trace request to peer nodes

			c.updateState(StateConnectionError)
			time.Sleep(sleep)
		}
		if err != nil {
			retry(err)
			continue
		}
		// Send connect message.
		m := message{
			Op: OpConnect,
		}
		req := connectReq{
			Host: c.Local,
			ID:   c.id,
		}
		err = c.sendMsg(conn, m, &req)
		if err != nil {
			retry(err)
			continue
		}
		// Wait for response
		var r connectResp

	rng := rand.New(rand.NewSource(time.Now().UTC().UnixNano()))
	var d net.Dialer
	var retry int
	for retry < len(hosts) {
		ctx, cancel := context.WithTimeout(GlobalContext, 300*time.Millisecond)

		host = hosts[rng.Intn(len(hosts))]
		conn, err := d.DialContext(ctx, "tcp", host)
		cancel()
		if err != nil {
			retry++
			continue
		}
		conn.Close()
		break
	}

	return host
}

func writeErrorResponse(ctx context.Context, w http.ResponseWriter, err APIError, reqURL *url.URL) {
	if err.HTTPStatusCode == http.StatusServiceUnavailable {
		// Set retry-after header to indicate user-agents to retry request after 120secs.
		// https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Retry-After
		w.Header().Set(xhttp.RetryAfter, "120")
	}

	switch err.Code {
	case "InvalidRegion":