…なるほど、理由が分かるとスッキリしますね。
read がない connection reset をリトライするのは、おそらく書き込み側（リクエストを送信する時）にエラーになったケースなので、その場合は処理が成功することはありえないので、リトライを行うということだったようです。

その上で今回はリトライすべきかどうか

今回の構築したサービスの仕様だと、Kinesis Data Streamsをサブスクライブしているアプリは 冪等 であることを期待しているので、重複してputすることを許容し、そのままリトライさせることにします（ていうかKinesisであればそもそもサービスとしてAt Least Onceなので、SDK側の判断で重複リトライしてもよいのでは..という気もしましたが、ダメなケースがあるのかな）。

カスタムリトライの実装

aws/aws-sdk-go のリポジトリの exampleフォルダ にカスタムリトライのサンプルコードがあり参考にできます。

実装を見ると、500番台のエラーは常に リトライしない という拡張なようです。

type CustomRetryer struct {
	client.DefaultRetryer
}

func (r CustomRetryer) ShouldRetry(req *request.Request) bool {
	if req.HTTPResponse.StatusCode >= 500 {
		// Don't retry any 5xx status codes.
		return false
	}

	// Fallback to SDK's built in retry rules
	return r.DefaultRetryer.ShouldRetry(req)
}

今回私が実装したいのは、 read: connection reset の時も リトライを行いたい ということなのでその条件のときに return true という、ほぼ同じ考えが適用できるステキなサンプルでした。

次に実装をあげますが、元のDefaultRetryerがtemporaryというインタフェースでスイッチしていた実装なのでそれを切り貼りしています。

type CustomRetryer struct {
	client.DefaultRetryer
}

type temporary interface {
	Temporary() bool
}

func (r CustomRetryer) ShouldRetry(req *request.Request) bool {
	if isErrReadConnectionReset(req.Error); {
		return true
	}
	return r.DefaultRetryer.ShouldRetry(req)
}

func isErrReadConnectionReset(err error) bool {
	switch e := err.(type) {
	case awserr.Error:
		origErr := e.OrigErr()
		if origErr != nil {
			return isErrReadConnectionReset(origErr)
		}
	case temporary:
		if strings.Contains(err.Error(), "read: connection reset by peer") {
			return true
		}
	}
	return false
}

上記の実装だと、 read: connection reset が発生した場合に規定の回数より多くリトライをしてしまうのでは？ という懸念が浮かびましたが、ドキュメントを読むと最大リトライの配慮は別処理でなされるので問題ないようです。

// Implementations may consider request attempt count when determining if a
// request is retryable, but the SDK will use MaxRetries to limit the
// number of attempts a request are made
ShouldRetry(*Request) bool
https://docs.aws.amazon.com/sdk-for-go/api/aws/request/#Retryer

それ以外の判定はDefaultRetryerに最終的な判断を移譲させます。

実装したカスタムリトライは aws.session で設定できます。

var kc = kinesis.New(session.Must(session.NewSession(&aws.Config{
			Retryer: CustomRetryer{},
		}),
	))

DefaultRetryer側の設定を変えたい場合は、埋め込んでいるためそのまま設定できます。

var kc = kinesis.New(session.Must(
		session.NewSession(&aws.Config{
			Retryer: CustomRetryer{
				DefaultRetryer: client.DefaultRetryer{
					NumMaxRetries: client.DefaultRetryerMaxNumRetries,

					// その他のデフォルト値も必要であれば設定しておく
					MinRetryDelay:    client.DefaultRetryerMinRetryDelay,
					MinThrottleDelay: client.DefaultRetryerMinThrottleDelay,
					MaxRetryDelay:    client.DefaultRetryerMaxRetryDelay,
					MaxThrottleDelay: client.DefaultRetryerMaxThrottleDelay,
				},
			},
		}),
	))

既存のパッケージの機能をそのまま使えるのは安心感があると思います。こういう薄いラッパーが作りやすいのは嬉しい仕組みですね。

errors.Is() で判定できるのでは？

私は試していませんが、 syscall.ECONNRESET を用いて判定もできるかと思います。先ほどのisErrReadConnectionReset()を少し書き換えます。

func isErrReadConnectionReset(err error) bool {
	switch e := err.(type) {
	case awserr.Error:
		origErr := e.OrigErr()
		if origErr != nil {
			return isErrReadConnectionReset(origErr)
		}
	case temporary:
-		if strings.Contains(err.Error(), "read: connection reset by peer") {
+		if errors.Is(err, syscall.ECONNRESET) {
			return true
		}
	}
	return false
}

さいごに

今まであまり深く気に留めていなかった read: connection reset by peer といったエラーに関しても、SDK実装者側の設計や配慮を抑え、アプリ開発に活かすと不明瞭な点が減り、より自信を持ったコードを書けるようになりました。