Class BrokerAssociatedNameStrategy

java.lang.Object
io.confluent.kafka.serializers.subject.AssociatedNameStrategy
io.confluent.kafka.schemaregistry.validator.BrokerAssociatedNameStrategy
All Implemented Interfaces:
io.confluent.kafka.serializers.subject.strategy.SubjectNameStrategy, org.apache.kafka.common.Configurable
public class BrokerAssociatedNameStrategy extends io.confluent.kafka.serializers.subject.AssociatedNameStrategy
A broker-side AssociatedNameStrategy that replaces the parent's synchronous Guava cache with a Caffeine cache tuned for broker schema validation. Compared to the parent it:
  • Refreshes instead of expiring on the hot path. Associations are mutable, so a resolved subject is refreshed in the background (refreshAfterWrite) rather than evicted. After the one-time cold load, reads never block on a synchronous Schema Registry call: an entry past the refresh interval is refreshed asynchronously while its existing value keeps being served, and a failed refresh retains that value. Refresh is best-effort — correctness never depends on it keeping up. The parent's expireAfterWrite instead forced a synchronous Schema Registry call on the append thread at every TTL boundary.
  • Serves stale during outages (availability). While Schema Registry is unreachable, refreshes fail and the last-known association keeps being served indefinitely rather than blocking or rejecting; the next successful async refresh restores freshness.
  • Retries only transient failures. IOException and 5xx responses are retried (reusing the validator's retry knobs); a 4xx is a definitive transport error and is surfaced immediately. A legitimate answer — no association (empty list / 404 endpoint), multiple associations, or a changed subject — is cached and refreshed, never retried.
  • Caches definitive negatives. "Multiple associations" and "no association with a NONE fallback" are cached as error BrokerAssociatedNameStrategy.Resolutions for the normal refresh cycle, so they no longer hammer Schema Registry on every record; a later refresh that finds an association replaces the negative.
  • Preserves the original cause. Transient failures are thrown as a SerializationException whose cause is the underlying RestClientException / IOException, so RecordSchemaValidator can classify them onto the right error sensor.

subjectName(String, boolean, ParsedSchema) is fully overridden, so the parent's own cache (built by super.configure) is never consulted.

  • Field Summary

    Fields
    Modifier and Type
    Field
    Description
    static final String
    FAILURE_BACKOFF_SECS
    After exhausting retries, how long (seconds) to short-circuit further lookups for the same key and rethrow the cached failure without calling Schema Registry.
    static final String
    MAX_RETRIES
    Max retries on a transient (IOException / 5xx) association lookup failure.
    static final String
    RETRIES_WAIT_MS
    Wait, in milliseconds, between transient retries.

    Fields inherited from class io.confluent.kafka.serializers.subject.AssociatedNameStrategy

    CACHE_EXPIRY_SECS, CACHE_SIZE, FALLBACK_TYPE, KAFKA_CLUSTER_ID, NAMESPACE_WILDCARD

  • Constructor Summary

    Constructors
    Constructor
    Description
    BrokerAssociatedNameStrategy()
     

  • Method Summary

    Modifier and Type
    Method
    Description
    void
    configure(Map<String,?> configs)
     
    void
    setKafkaClusterId(String clusterId)
     
    void
    setSchemaRegistryClient(io.confluent.kafka.schemaregistry.client.SchemaRegistryClient client)
     
    String
    subjectName(String topic, boolean isKey, io.confluent.kafka.schemaregistry.ParsedSchema schema)
     

    Methods inherited from class io.confluent.kafka.serializers.subject.AssociatedNameStrategy

    usesSchema

    Methods inherited from class Object

    equals, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

  • Field Details

    • MAX_RETRIES

      public static final String MAX_RETRIES
      Max retries on a transient (IOException / 5xx) association lookup failure.
      See Also:

    • RETRIES_WAIT_MS

      public static final String RETRIES_WAIT_MS
      Wait, in milliseconds, between transient retries.
      See Also:

    • FAILURE_BACKOFF_SECS

      public static final String FAILURE_BACKOFF_SECS
      After exhausting retries, how long (seconds) to short-circuit further lookups for the same key and rethrow the cached failure without calling Schema Registry. <= 0 disables backoff.
      See Also:

  • Constructor Details

    • BrokerAssociatedNameStrategy

      public BrokerAssociatedNameStrategy()

  • Method Details

    • configure

      public void configure(Map<String,?> configs)
      Specified by:
      configure in interface org.apache.kafka.common.Configurable
      Overrides:
      configure in class io.confluent.kafka.serializers.subject.AssociatedNameStrategy

    • subjectName

      public String subjectName(String topic, boolean isKey, io.confluent.kafka.schemaregistry.ParsedSchema schema)
      Specified by:
      subjectName in interface io.confluent.kafka.serializers.subject.strategy.SubjectNameStrategy
      Overrides:
      subjectName in class io.confluent.kafka.serializers.subject.AssociatedNameStrategy

    • setSchemaRegistryClient

      public void setSchemaRegistryClient(io.confluent.kafka.schemaregistry.client.SchemaRegistryClient client)
      Specified by:
      setSchemaRegistryClient in interface io.confluent.kafka.serializers.subject.strategy.SubjectNameStrategy
      Overrides:
      setSchemaRegistryClient in class io.confluent.kafka.serializers.subject.AssociatedNameStrategy

    • setKafkaClusterId

      public void setKafkaClusterId(String clusterId)
      Specified by:
      setKafkaClusterId in interface io.confluent.kafka.serializers.subject.strategy.SubjectNameStrategy
      Overrides:
      setKafkaClusterId in class io.confluent.kafka.serializers.subject.AssociatedNameStrategy