Interface OdysseyRequestManager

All Superinterfaces:
AutoCloseable
public interface OdysseyRequestManager extends AutoCloseable
Interface for managing asynchronous requests to Odyssey (Schema Registry). Implementation ensures Controller threads are not blocked waiting for external service calls.

When Odyssey is disabled, the integration manager is null and the legacy code path is used instead - no request manager is instantiated.

For each Odyssey request (which may be a batch), a single SchemaRegistryClient should be obtained using

invalid reference 
#getClient(String, Map)
and passed to both validate and createAssociations. This ensures connection reuse within a request while avoiding global client caching.

Implementations:

  • DefaultOdysseyRequestManager: Full implementation with HTTP calls to Schema Registry

  • Method Details

    • acquireToken

      CompletableFuture<OdysseyRequestManager.TokenAcquisitionResult> acquireToken(org.apache.kafka.common.security.auth.KafkaPrincipal principal)
      Acquires a Logical Access Token (LAT) from Token Platform for SR authentication.

      This method runs asynchronously on the request manager's executor to avoid blocking controller threads. The returned future completes with:

      • Success result with token if acquisition succeeded
      • NotConfigured result if Token Platform is not configured (permanent)
      • PermanentError if principal metadata is missing or permissions are denied
      • TransientError if Token Platform call failed (may succeed on retry)
      Parameters:
      principal - The Kafka principal for the request
      Returns:
      CompletableFuture with the token acquisition result

    • getClient

      io.confluent.kafka.schemaregistry.client.SchemaRegistryClient getClient(String schemaRegistryUrl, Optional<String> lsrcId, Map<String,String> httpHeaders)
      Gets a SchemaRegistryClient for the given URL with HTTP headers and (for cloud) LSRC id. The client should be reused for all operations within a single Odyssey request.

      Note: For production Odyssey calls, the httpHeaders map should contain the Authorization header with a LAT token from Token Platform. Passing null or empty headers will result in unauthenticated SR calls that will be rejected.

      For cloud deployments, lsrcId should be Optional.of("lsrc-..."). It is set on the client as bearer.auth.logical.cluster, which causes the SR client to populate the target-sr-cluster HTTP header on every request — required by the multi-tenant Cloud SR. Pass Optional.empty() on platform deployments where there is no LSRC concept.

      Parameters:
      schemaRegistryUrl - The Schema Registry endpoint URL
      lsrcId - The Logical Schema Registry Cluster id; Optional.empty() on platform deployments
      httpHeaders - HTTP headers to include in requests (e.g., Authorization). May be null in test contexts, but required for production.
      Returns:
      A configured SchemaRegistryClient (with cacheCapacity=0 for minimal overhead)

    • validate

      CompletableFuture<Map<String, OdysseyResult>> validate(Map<String, ParsedAssociations> topicAssociations, String namespace, io.confluent.kafka.schemaregistry.client.SchemaRegistryClient client)
      Validates a batch against SR (dry-run=true). Used by the CreateTopics flow.

    • validateMixedBatch

      CompletableFuture<Map<String, OdysseyResult>> validateMixedBatch(List<TopicAssociationInfo> upserts, List<DisassociateInfo> disassociates, String namespace, io.confluent.kafka.schemaregistry.client.SchemaRegistryClient client)
      Validates a mixed upsert+disassociate batch against SR (dry-run=true). Used by the IncrementalAlterConfigs flow. Topics may appear in both lists; results are merged with failure-wins semantics — one failure for a topic marks the whole topic failed. TODO(DGS-24351): consolidate with validate(Map, String, SchemaRegistryClient) into a single batch API; the "Mixed" prefix becomes redundant once there's only one batch shape.

    • updateAssociations

      CompletableFuture<Map<String, OdysseyResult>> updateAssociations(List<TopicAssociationInfo> upserts, List<DisassociateInfo> disassociates, io.confluent.kafka.schemaregistry.client.SchemaRegistryClient client)
      Executes a mixed upsert+disassociate SR update (used by IncrementalAlterConfigs flow). Same structure as validateMixedBatch(List, List, String, SchemaRegistryClient) but dryRun=false.

    • createAssociations

      CompletableFuture<Map<String, OdysseyResult>> createAssociations(List<TopicAssociationInfo> topics, boolean topicAlreadyExists, io.confluent.kafka.schemaregistry.client.SchemaRegistryClient client)
      Creates associations with Odyssey.
      Parameters:
      topics - List of topics with their association information
      topicAlreadyExists - If true, uses INSERT-only semantics (topic already exists)
      client - The SchemaRegistryClient
      Returns:
      CompletableFuture with Map of topic name to result

    • fetchAssociations

      CompletableFuture<Map<String, OdysseyResult>> fetchAssociations(List<String> topicNames, String namespace, io.confluent.kafka.schemaregistry.client.SchemaRegistryClient client)
      Fetches existing schema associations (without their schemas) for the given topics from Schema Registry. Used by the DescribeConfigs Odyssey flow to enrich describe responses with association data. Schemas are never returned via DescribeConfigs.
      Parameters:
      topicNames - List of topic names to fetch associations for
      namespace - The cluster namespace
      client - The SchemaRegistryClient to use for this request
      Returns:
      CompletableFuture with Map of topic name to result; never throws — errors are captured per-topic as failed OdysseyResult instances

    • deleteAssociations

      CompletableFuture<Map<String, OdysseyResult>> deleteAssociations(Map<String, org.apache.kafka.common.Uuid> topicNamesAndIds, String namespace, io.confluent.kafka.schemaregistry.client.SchemaRegistryClient client)
      Deletes associations for the given topics from Schema Registry. Used by the DeleteTopics Odyssey flow to remove associations before topic deletion.

      Each topic's key and/or value associations are deleted via AssociationDeleteOp. No dry-run is performed — this is a real mutation.

      Parameters:
      topicNamesAndIds - Map of topic names to their Uuid ids whose associations should be deleted; iteration order determines request order
      namespace - The cluster namespace (LKC ID)
      client - The SchemaRegistryClient to use
      Returns:
      CompletableFuture with Map of topic name to result; errors captured per-topic

    • close

      void close()
      Gracefully shuts down the manager, cleaning up resources.
      Specified by:
      close in interface AutoCloseable