Class SchemaCache

java.lang.Object
io.krabka.streams.schema.SchemaCache
public final class SchemaCache extends Object
Stores schema IDs and writer schemas for synchronous serde operations.

Serdes must never block a Kafka poll loop on registry I/O, so all registry traffic is funneled through this cache:

  1. Each serde calls intern(String, SchemaKind, String, String) (through its registerSubject method) to declare the subjects it will use.
  2. The application calls prewarm() once at startup, which resolves every interned subject to a schema ID with the configured RegisterMode.
  3. Serialization then reads IDs synchronously with idForSubject(String), and deserialization reads writer schemas with writerSchema(int); an unknown ID starts one background fetch and throws the retriable SchemaFetchPendingException.

The cache is thread-safe and can be shared by any number of serdes; sharing one cache per application is the intended usage.

Example

var client = new KrabkaSchemaRegistryClient(URI.create("http://localhost:8081"));
var cache = new SchemaCache(client, RegisterMode.LOOKUP_ONLY, new TopicNameStrategy());

keySerde.registerSubject("orders");
valueSerde.registerSubject("orders");

var report = cache.prewarmReport().join();
if (!report.successful()) {
    report.failures().forEach((subject, error) ->
        log.error("cannot resolve {}", subject, error));
}

  • Constructor Details

    • SchemaCache

      public SchemaCache(KrabkaSchemaRegistryClient client)
      Creates a cache that auto-registers schemas and uses the topic naming rule.
      Parameters:
      client - the registry client used for prewarming and background fetches

    • SchemaCache

      public SchemaCache(KrabkaSchemaRegistryClient client, RegisterMode registerMode, SubjectNameStrategy subjectNameStrategy)
      Creates a fully configured cache.
      Parameters:
      client - the registry client used for prewarming and background fetches
      registerMode - how prewarming resolves each interned subject to an ID
      subjectNameStrategy - the default topic-to-subject mapping

  • Method Details

    • subject

      public String subject(String topic, Role role)
      Maps a topic and role to a subject with the cache's default strategy.
      Parameters:
      topic - the Kafka topic name
      role - whether the schema describes the record key or value
      Returns:
      the registry subject name

    • subject

      public String subject(String topic, Role role, SubjectNameStrategy strategy)
      Maps a topic and role to a subject with an explicit strategy.
      Parameters:
      topic - the Kafka topic name
      role - whether the schema describes the record key or value
      strategy - the strategy to use instead of the cache default
      Returns:
      the registry subject name

    • intern

      public void intern(String subject, SchemaKind kind, String schema, String messageType)
      Adds a local schema to the next prewarm operation. This operation is idempotent by subject.

      Serdes call this through their registerSubject methods; call it directly only when prewarming subjects that no local serde owns.

      Parameters:
      subject - the registry subject to resolve during prewarm
      kind - the schema format
      schema - the local schema text
      messageType - the Protobuf message full name, or null for other formats

    • prewarm

      public CompletableFuture<Void> prewarm()
      Resolves all interned subject IDs with the configured registration mode.

      All subjects are resolved concurrently. The returned future fails when any subject fails, without reporting which one; use prewarmReport() for per-subject outcomes.

      Returns:
      a future that completes when every interned subject has resolved

    • prewarmReport

      public CompletableFuture<SchemaCache.PrewarmReport> prewarmReport()
      Resolves every interned subject and reports successes and failures independently.

      Unlike prewarm(), the returned future always completes normally; inspect the report to find out which subjects failed and why. Successfully resolved subjects are cached even when others fail.

      Returns:
      a future with the per-subject resolution report

    • idForSubject

      public OptionalInt idForSubject(String subject)
      Returns the resolved schema ID for a subject, if prewarming has resolved it.
      Parameters:
      subject - the registry subject name
      Returns:
      the schema ID, or empty when the subject has not been resolved

    • writerSchema

      public String writerSchema(int schemaId)
      Returns a cached writer schema. A cache miss starts one background fetch and throws a retriable error.

      This method never blocks. Only one fetch per schema ID runs at a time; concurrent callers for the same ID all receive SchemaFetchPendingException until the fetch completes.

      Parameters:
      schemaId - the schema ID from a record's frame header
      Returns:
      the writer schema text registered under the ID
      Throws:
      SchemaFetchPendingException - while the schema is being fetched

    • writerMessageType

      public String writerMessageType(int schemaId)
      Returns the cached Protobuf message full name for a schema ID.
      Parameters:
      schemaId - the schema ID from a record's frame header
      Returns:
      the writer's message full name, or null when unknown or not Protobuf

    • writerReferences

      public Map<String,String> writerReferences(int schemaId)
      Returns the cached referenced schemas for a schema ID.
      Parameters:
      schemaId - the schema ID from a record's frame header
      Returns:
      reference name to schema text, empty when the schema has no references

    • seedSubjectId

      public void seedSubjectId(String subject, int schemaId)
      Adds a subject ID directly. This method supports deterministic tests and offline startup.
      Parameters:
      subject - the registry subject name
      schemaId - the schema ID to associate with the subject

    • seedWriterSchema

      public void seedWriterSchema(int schemaId, String schema)
      Adds a writer schema directly. This method supports deterministic tests and offline startup.
      Parameters:
      schemaId - the schema ID to associate with the schema text
      schema - the writer schema text

    • seedWriterMessageType

      public void seedWriterMessageType(int schemaId, String messageType)
      Adds Protobuf message metadata directly.
      Parameters:
      schemaId - the schema ID to associate with the message name
      messageType - the Protobuf message full name

    • seedWriterReferences

      public void seedWriterReferences(int schemaId, Map<String,String> references)
      Adds referenced writer schemas directly. This method supports deterministic tests and offline startup.
      Parameters:
      schemaId - the parent schema ID
      references - reference name to schema text