Fixes #663. Updated from abandoned PR #673: rebased to target 3.x and improved retry interface. Restructured PR description for readability

Goal

Prevent transient schema registry API errors from blowing up Vulcan serialisation/deserialisation by adding retries with sensible default config.

Implementation

• add a new trait SchemaRegistryClientRetry[F[_]] allowing a retry strategy on an arbitrary computation that uses the schema registry, F[A], to be specified.
• add default implementation that retries on schema registry exceptions identified in the underlying Java code (see "What Throwables should we retry?" below).
  • the default strategy uses a retry time backoff, requiring Sync to be upgraded to Async on some AvroSettings constructors; therefore this change is breaking and targets 3.x

Retry Interface Design

Feedback is very welcome here. I took care in designing this, drawing significant influence from discussions on issue #623 about commit retry.

Proposed Interface

trait SchemaRegistryClientRetry[F[_]] {

  def withRetry[A](action: F[A]): F[A]
}

// AvroSettings.scala
sealed abstract class AvroSettings[F[_]] {
  ...
  def schemaRegistryClientRetry: SchemaRegistryClientRetry[F]
}

Benefits: very flexible and composable; users can bring their own F to perform logging etc. Easy interop with e.g. cats-retry library. Robust against future implementation changes in Vulcan or the underlying Java libraries; the retry can be applied to any computation thanks to the type param.
Cons: required an intermediate trait to get the type parameter [A] because an anonymous function can't have a type param, e.g. we can't do def schemaRegistryClientRetry: F[A] => F[A].

To try and get rid of the need of the type param and new trait, I also considered a variation of this proposal where the three Vulcan call sites that use the schema registry (directly or indirectly) could take one non-type-paramaterised anonymous function each as retry strategies (e.g. def schemaRegistryGetSchemaRetry: F[ParsedSchema] => F[ParsedSchema]), but this would leave the retry config extremely coupled to the implementation details of the existing ser/des.

Alternative 1

(this was my first iteration)

// AvroSettings.scala
sealed abstract class AvroSettings[F[_]] {
  ...
  def schemaRegistryRetryMaxAttempts: Int
  def schemaRegistryRetryBackoffStrategy: Int => FiniteDuration
}

Benefits: simple interface with no new traits.
Cons: inflexible; no provision to add logging etc. No cats-retry interop.

Alternative 2

Implement tagless final traits as facades over the relevant Java schema registry classes, allow retries to be specified on their implementations.

Cons: this requires huge change.

What Throwables should we retry?

Some code analysis was required to determine where the schema registry is used and what exceptions it can throw. I wanted to avoid retrying on every error, which would include deterministic failures.

• in general, schema registry client operations can throw IOException or RestClientException
• for serialization, the relevant Java KafkaAvroSerializer error handling maps both IOException and RestClientException (if non-client http error code) to SerializationException
• for deserialization, the vulcan code makes a call using the schema registry client itself and further calls are made by the underlying Java code (1, 2, 3) which are mapped again to SerializationException

TL;DR this new helper function catches the relevant exceptions:

  val isRetriable: Throwable => Boolean = {
    case _: SerializationException     => true
    case _: IOException                => true
    case apiError: RestClientException => apiError.getErrorCode >= 500
    case _                             => false
  }

I made the function public so users of cats-retry etc. have it available to them for easy re-use.

TODO

Once the interface is agreed, I need to add unit tests and docs.