Status

Current state: Draft

Discussion thread: Mailing list discussion

JIRA: [To Be Created]

Released: XXX

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

The Economic Challenge of Tiered Storage

KIP-405 (Tiered Storage) transformed Kafka's architecture by decoupling compute from storage, enabling infinite retention through remote object storage (S3, GCS, Azure Blob). While this reduces storage costs by 30-40%, it introduces a critical operational challenge: variable operational expenses for data access.

In traditional Kafka deployments, read operations have zero marginal cost once infrastructure is provisioned. With Tiered Storage, remote fetch operations incur direct costs:

The Visibility Gap

Despite KIP-963 providing broker-level metrics for Tiered Storage health monitoring, a critical gap remains: operators cannot attribute remote storage costs to specific consumer applications.

Current limitations:

Real-world impact: A misconfigured consumer performing a full historical scan can generate thousands of dollars in S3 costs without detection until the monthly bill arrives.

Use Cases Requiring Cost Attribution

  1. Multi-Tenant Chargeback: Enterprise clusters serving 100+ teams need to bill specific cost centers for their remote storage consumption
  2. Rogue Consumer Detection: Identify consumers with auto.offset.reset=earliest causing unexpected cost spikes
  3. Optimization Guidance: Detect inefficient fetch patterns (small fetch sizes causing high API request counts)
  4. Compliance Auditing: Track which applications accessed historical archives for regulatory requirements
  5. Financial Quotas: Implement real-time cost-based throttling to prevent bill shock

Business Significance

From a business perspective, KIP-1261 represents the foundational architecture for financial governance in streaming data. It transitions Kafka from a "black box" of infrastructure spend to a transparent, auditable platform compliant with enterprise FinOps standards. This contribution enables organizations to:

  1. Implement Granular Chargeback: Accurately bill specific cost centers for their historical data consumption
  2. Enforce Financial Quotas: Detect and throttle "rogue" consumers based on cost velocity rather than just bandwidth
  3. Optimize Cloud Spend: Identify inefficient consumption patterns (e.g., small fetch sizes causing high API costs) and drive architectural improvements

The following sections detail the technical specification, implementation strategy, and validation plans for this enhancement, serving as a guide for architecting financial accountability in multi-tenant Kafka ecosystems.

Proposed Changes

New Metrics

This KIP proposes a new JMX metric group RemoteFetchMetrics with client-level attribution:

1.RemoteFetchBytesPerSec

2.RemoteFetchRequestsPerSec

3.RemoteFetchLatency


JMX ObjectName Structure

kafka.server:type=RemoteFetchMetrics,name=RemoteFetchBytesPerSec ,client-id={client_id},topic={topic_name}

Configuration Parameters

ConfigurationTypeDefaultDescription
remote.log.metrics.cost.attribution.enabledBooleanfalseMaster switch to enable client-level metrics
remote.log.metrics.max.consumer.groupsInt1000Maximum unique client-ids tracked (LRU eviction)
remote.log.metrics.include.partitionBooleanfalseInclude partition tag (increases cardinality)


Public Interfaces

Modified Classes

RemoteStorageFetchInfo


Add optional clientId field to propagate request context:

public class RemoteStorageFetchInfo {
 private final Optional<String> clientId;
 
 public RemoteStorageFetchInfo(..., Optional<String> clientId) {
 this.clientId = clientId;
 }
 
 public Optional<String> clientId() {
 return clientId;
 }
}


New Metrics Registry

New metric group: kafka.server:type=RemoteFetchMetrics

This is separate from BrokerTopicMetrics to isolate high-cardinality client-level data.

Proposed Implementation

Architecture Overview

The implementation follows a "surgical instrumentation" approach with minimal changes to the data path:

Key Implementation Points

1.ReplicaManager Modification

2,RemoteLogManager Instrumentation

3.Thread Safety

Performance Characteristics

Financial Governance Framework

This section demonstrates how KIP-1261 enables financial accountability in multi-tenant Kafka environments.

Cost Attribution Model

KIP-1261 provides the telemetry needed to calculate per-client costs using the following formula:

Cost_Client = (V_Egress × R_Egress) + (N_Requests × R_API)

Where:
• VEgress = Total bytes from RemoteFetchBytesPerConsumerGroup
• REgress = Cloud provider's egress rate (e.g., $0.09/GB for Internet, $0.01/GB for Inter-AZ)
• NRequests = Total count from RemoteFetchRequestsPerConsumerGroup
• RAPI = Cloud provider's API rate (e.g., $0.0004 per 1,000 GET requests)

Governance Models Enabled

Organizations can implement three levels of financial maturity:

Level 1: Showback Model

Level 2: Chargeback Model

Level 3: Real-Time Cost Enforcement

Enterprise Adoption Impact

KIP-1261 addresses a critical barrier to Tiered Storage adoption in enterprise environments. Without cost attribution, organizations cannot:

By providing granular cost visibility, this KIP transforms Tiered Storage from a feature with uncertain cost implications into a governable, predictable storage strategy suitable for regulated industries requiring infinite retention capabilities.

Integration with Existing Kafka Features

The metrics provided by KIP-1261 can be integrated with:

Backward Compatibility

Migration Strategy

Phase 1 - Preparation:

  1. Upgrade brokers to version containing KIP-1261
  2. Ensure remote.log.metrics.cost.attribution.enabled=false (default)
  3. Verify cluster stability

Phase 2 - Canary Activation:

  1. Enable on single broker: remote.log.metrics.cost.attribution.enabled=true
  2. Monitor JMX endpoints for new metrics
  3. Validate metric accuracy

Phase 3 - Observability Integration:

  1. Configure Prometheus JMX Exporter
  2. Deploy Grafana dashboards
  3. Validate metric aggregation

Phase 4 - Full Rollout:

  1. Enable on all brokers
  2. Begin chargeback data collection

Rollback Plan

Instant rollback via dynamic configuration: set remote.log.metrics.cost.attribution.enabled=false. This immediately stops metric recording without requiring broker restart.

Test Plan

Unit Tests

Integration Tests

Performance Tests

Operational Guide

Prometheus Integration

JMX Exporter configuration:

rules:
 - pattern: kafka.server<type=RemoteFetchMetrics, name=RemoteFetchBytesPerSec, client-id=(.+), topic=(.+)><>Count
 name: kafka_server_remote_fetch_bytes_total
 labels:
 client_id: "$1"
 topic: "$2"
 type: COUNTER

Example PromQL Queries

Total bytes by client (30 days):

sum(increase(kafka_server_remote_fetch_bytes_total[30d])) by (client_id)

Estimated hourly cost:

sum(rate(kafka_server_remote_fetch_bytes_total[1h])) by (client_id) * 0.00000000009

Fetch efficiency (bytes per request):

sum(rate(kafka_server_remote_fetch_bytes_total[1h])) by (client_id) 
/ 
sum(rate(kafka_server_remote_fetch_requests_total[1h])) by (client_id)

Rejected Alternatives

Topic-Level Only Attribution

Approach: Map topics to teams via external CMDB.

Rejection Reason: Fails for shared topics consumed by multiple teams. Cannot determine which consumer is driving costs.

Client-Side Telemetry (KIP-714)
Approach: Have clients report their own fetch statistics.

Rejection Reasons:

S3 Access Log Parsing

Approach: Parse cloud provider access logs to attribute costs.

Rejection Reason: S3 logs only contain broker IP addresses, not client-ids. Correlation is impossible without broker-side instrumentation.

References

• KIP-405: Kafka Tiered Storage
• KIP-963: Additional metrics in Tiered Storage
• KIP-714: Client Metrics and Observability