Class CpcClusterMetadata

java.lang.Object
io.confluent.kafka.multitenant.BasePhysicalClusterMetadata<io.confluent.kafka.cpc.multitenant.CpcLogicalClusterMetadata>
io.confluent.kafka.multitenant.CommonTopicBasedPhysicalClusterMetadata<io.confluent.kafka.cpc.multitenant.CpcLogicalClusterMetadata>
io.confluent.kafka.cpc.multitenant.CpcClusterMetadata
All Implemented Interfaces:
org.apache.kafka.common.Configurable, org.apache.kafka.common.Reconfigurable, org.apache.kafka.server.multitenant.MultiTenantMetadata
public class CpcClusterMetadata extends io.confluent.kafka.multitenant.CommonTopicBasedPhysicalClusterMetadata<io.confluent.kafka.cpc.multitenant.CpcLogicalClusterMetadata>
Broker-side tenant metadata store for CPC multi-tenancy. Consumes JSON-encoded CpcLogicalClusterMetadata records from a compacted internal topic and maintains an in-memory cache of tenant metadata.

Extends CommonTopicBasedPhysicalClusterMetadata to share the common consume loop, sequence-id dedup (via io.confluent.kafka.multitenant.CommonTopicBasedPhysicalClusterMetadata<io.confluent.kafka.cpc.multitenant.CpcLogicalClusterMetadata>.io.confluent.kafka.multitenant.CommonTopicBasedPhysicalClusterMetadata.LCMPair), and state machine with the Cloud implementation. CPC does not register or emit any metrics. CPC-specific behavior: JSON parsing (vs. protobuf), TopicAdmin-based auto-creation of the metadata topic, tenant lifecycle management via CpcTenantLifecycleManager, and callback-based tenant event notification.

There are two distinct phases of processing:

  1. Startup (STARTING): The entire topic is read to end via KafkaBasedLog.start(). Each record is stored in the cache, but updateTenant(CpcLogicalClusterMetadata, CpcLogicalClusterMetadata) is not called (no callbacks fire per-record). After the initial load, a single batch of bookkeeping runs: rebuild active IDs, invoke register callbacks for all active tenants, then set state to RUNNING.
  2. Running (RUNNING): New records from the topic are consumed one at a time by the KafkaBasedLog WorkThread. Each record triggers updateTenant(CpcLogicalClusterMetadata, CpcLogicalClusterMetadata), which fires the appropriate callback (register, update, or deactivated) and rebuilds active IDs.

  • Constructor Details

    • CpcClusterMetadata

      public CpcClusterMetadata(org.apache.kafka.common.metrics.Metrics metrics)

    • CpcClusterMetadata

      public CpcClusterMetadata(org.apache.kafka.common.metrics.Metrics metrics, org.apache.kafka.common.utils.Time time)

  • 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.multitenant.CommonTopicBasedPhysicalClusterMetadata<io.confluent.kafka.cpc.multitenant.CpcLogicalClusterMetadata>

    • start

      public Map<org.apache.kafka.common.Endpoint, CompletableFuture<Void>> start(Map<String,Object> interBrokerClientConfig, Collection<org.apache.kafka.common.Endpoint> endpoints)

    • startClients

      public void startClients(Function<String, org.apache.kafka.clients.admin.Admin> adminClientSupplier)

    • isUp

      public boolean isUp()

    • logicalClusterIds

      public Set<String> logicalClusterIds()
      Specified by:
      logicalClusterIds in interface org.apache.kafka.server.multitenant.MultiTenantMetadata
      Specified by:
      logicalClusterIds in class io.confluent.kafka.multitenant.CommonTopicBasedPhysicalClusterMetadata<io.confluent.kafka.cpc.multitenant.CpcLogicalClusterMetadata>

    • kafkaLogicalClusterIds

      public Set<String> kafkaLogicalClusterIds()

    • dedicatedLogicalClusterId

      public String dedicatedLogicalClusterId()

    • registerTenantCallback

      public void registerTenantCallback(Function<String,Boolean> callback)

    • registerTenantMetadataUpdateCallback

      public void registerTenantMetadataUpdateCallback(Function<String,Boolean> callback)

    • registerTenantDeactivatedCallback

      public void registerTenantDeactivatedCallback(Function<String,Boolean> callback)

    • consume

      public void consume(org.apache.kafka.clients.consumer.ConsumerRecord<String,byte[]> record)
      Consume a single metadata record, then refresh tenant quotas. This is the single choke point for quota refresh in the RUNNING phase: it covers every mutation kind — register, update, deactivate (all via updateLogicalCluster(String, Long, CpcLogicalClusterMetadata)) and topic tombstone (via the parent's removeLogicalCluster) — because they all flow through consume(). During the STARTING load, per-record refresh is intentionally skipped; startLog() pushes the initial active set once after the full load.
      Overrides:
      consume in class io.confluent.kafka.multitenant.CommonTopicBasedPhysicalClusterMetadata<io.confluent.kafka.cpc.multitenant.CpcLogicalClusterMetadata>