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:

• API Request Charges: Cloud providers charge per 1,000 GET requests (e.g., $0.0004 on AWS S3)
• Data Egress Charges: Cross-AZ or cross-region transfers incur substantial fees (e.g., $0.01-$0.09 per GB)

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:

• KIP-963 metrics aggregate at the topic level (e.g., RemoteFetchBytesPerSec per topic)
• No visibility into which client-id or consumer-group is driving remote fetch costs
• Impossible to implement chargeback or showback models in multi-tenant environment

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

• Type: Rate metric (bytes/second) with total count
• Tags: client-id, topic, partition (optional)
• Description: Tracks bytes transferred from remote storage per client
• Use: Calculate data egress costs

2.RemoteFetchRequestsPerSec

• Type: Rate metric (requests/second) with total count
• Tags: client-id, topic
• Description: Tracks number of remote fetch operations per client
• Use: Calculate API request costs; identify inefficient fetch patterns

3.RemoteFetchLatency

• Type: Histogram (p50, p95, p99)
• Tags: client-id, topic
• Description: Measures remote fetch operation latency
• Use: Differentiate between slow storage and slow consumers

JMX ObjectName Structure

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

Configuration Parameters

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:

• Context Propagation (ReplicaManager): Extract clientId from FetchParams and inject into RemoteStorageFetchInfo
• Sensor Management (RemoteLogManager): Maintain ConcurrentHashMap with LRU eviction
• Metric Recording (RemoteLogManager): Record metrics in the async fetch callback upon successful completion

Key Implementation Points

1.ReplicaManager Modification

• When constructing RemoteStorageFetchInfo for remote fetches, extract and pass the clientId from the incoming FetchRequest.

2,RemoteLogManager Instrumentation

• Check remote.log.metrics.cost.attribution.enabled configuration
• Lookup or create sensor for (clientId, topic) tuple (subject to LRU limits)
• Record request count immediately upon task submission
• Record byte count in the completion callback after successful fetch

3.Thread Safety

• Use Kafka's thread-safe Metrics library (atomic accumulators)
• Metric recording occurs on RemoteLogManager's thread pool (not network I/O threads)
• ConcurrentHashMap for sensor cache with LRU eviction

Performance Characteristics

• CPU Overhead: < 1.2% (hash map lookup + metric recording)
• Latency Impact: ~0.3ms added to fetch path (negligible compared to 50-500ms S3 latency)
• Memory Footprint: ~5KB per sensor × 1000 max sensors = ~5MB heap

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

• Mechanism: Dashboards displaying top consumers by cost
• Goal: Visibility and awareness
• Implementation: Grafana dashboards with no billing system integration required
• Outcome: Teams voluntarily optimize consumption patterns based on visibility

Level 2: Chargeback Model

• Mechanism: Automated monthly cost reports per consumer group
• Goal: Cost recovery and accountability
• Implementation: Export metrics to billing systems for internal invoicing
• Outcome: Platform teams can allocate actual costs to consuming business units, transforming Kafka from a cost center to a cost pass-through service

Level 3: Real-Time Cost Enforcement

• Mechanism: Stream processing of KIP-1261 metrics with automated quota application
• Goal: Cost prevention
• Implementation: Monitor RemoteFetchBytesPerSec per client-id and trigger quota enforcement when thresholds are exceeded
• Example: If client-id=team-a exceeds $50/hour in remote fetch costs, automatically apply throughput quotas to prevent runaway spending
• Outcome: Prevents unexpected cost spikes before monthly billing cycles complete

Enterprise Adoption Impact

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

• Implement fair billing in multi-tenant clusters
• Detect and prevent cost anomalies in real-time
• Optimize consumption patterns based on financial impact
• Meet compliance requirements for data access auditing

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:

• Kafka Quotas: Use cost metrics to inform quota policies
• ACLs: Combine with access control for comprehensive governance
• Monitoring Systems: Export to Prometheus, Datadog, or CloudWatch for alerting
• FinOps Platforms: Feed into enterprise cost management tools
• Compatibility, Deprecation, and Migration Plan

Backward Compatibility

• Wire Protocol: No changes
• Client Libraries: No changes required
• Storage Plugins: No changes to RemoteStorageManager interface
• Existing Metrics: KIP-963 metrics remain unchanged; KIP-1261 metrics are additive

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

• Verify exact byte count attribution for mock remote fetches
• Test LRU eviction with max.consumer.groups limit
• Validate context propagation from FetchRequest to RemoteStorageFetchInfo

Integration Tests

• Multi-tenant simulation: verify independent attribution for concurrent consumers
• Fault tolerance: ensure failed fetches don't increment byte metrics
• Cardinality safety: verify sensor map size limits

Performance Tests

• Benchmark throughput degradation (target: < 1%)
• Measure CPU overhead (target: < 2%)
• Validate latency impact (target: < 1ms)

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:

• Financial billing cannot rely on self-reported data (trust issue)
• Requires client upgrades (KIP-1261 works with all existing clients)
• Broker is the authoritative source for billing

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