Status

Current state: DraftUnder Discussion

Discussion thread: Mailing list discussion - to be updated]

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.

...

• 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)

Sequential Fetch Limitation

Due to KAFKA-14915, the broker currently fetches only one partition per remote fetch request, rather than batching across partitions. This architectural constraint means consumers reading from multi-partition topics generate a higher volume of individual API GET requests than theoretically necessary. This amplifies the financial impact of remote storage access and makes precise per-client cost attribution even more critical for identifying which consumers are most affected by this sequential fetch behavior.

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.

...

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

4. RemoteFetchErrorsPerSec

• Type: Rate metric (errors/second) with total count
• Tags: client-id, topic, error-type (optional)
• Description: Tracks failed remote fetch attempts per client
• Use: Attribute API request costs for failed operations; identify problematic consumers

Note: Cloud providers often charge for API GET requests even when requests fail due to timeouts or storage errors. This metric is essential for complete financial attribution.

JMX ObjectName Structure

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

...

remote.log.metrics.max.

client.

sensors

Relationship to KIP-963

KIP-963 introduced RemoteFetchRequestsPerSec and RemoteFetchBytesPerSec at the topic level under BrokerTopicMetrics. KIP-1267 does not replace these metrics. Instead, it provides a high-resolution, opt-in view of the same data with client-level attribution. The KIP-963 metrics remain the primary operational health indicators, while KIP-1267 metrics enable FinOps and chargeback use cases that require identity context.

Public Interfaces

Modified Classes

RemoteStorageFetchInfo

Add optional clientId field to propagate request context:

...

• 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
• Record error count for failed fetch attempts

Billing Accuracy: RemoteFetchBytesPerSec is recorded in the async completion callback only after successful data transfer. This ensures the metric reflects actual payload bytes that match cloud provider billing (e.g., AWS S3 does not charge egress for failed transfers). Failed fetches increment RemoteFetchErrorsPerSec but not RemoteFetchBytesPerSec.

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

4. MBean Lifecycle Management

When a sensor is evicted from the LRU cache due to the remote.log.metrics.max.client.sensors limit, its corresponding JMX MBean will be immediately unregistered from the JMX registry. This ensures bounded memory usage and prevents MBean leaks as client populations churn over time.

Performance Characteristics

...

Where:
• VEgress = Total bytes from RemoteFetchBytesPerConsumerGroup RemoteFetchBytesPerSec (aggregated count)

• REgress = Cloud provider's egress rate (e.g., $0.09/GB for Internet, $0.01/GB for Inter-AZ)
• NRequests = Total count from RemoteFetchRequestsPerConsumerGroup RemoteFetchRequestsPerSec + RemoteFetchErrorsPerSec

• RAPI = Cloud provider's API rate (e.g., $0.0004 per 1,000 GET requests)

...

• 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

Security Considerations

The metrics introduced by this KIP expose client-id information through JMX endpoints. Organizations should:

• Restrict JMX access to authorized operators only
• Consider implementing authentication for metric exporters (Prometheus, etc.)
• Be aware that client-id values may contain sensitive team/application identifiers
• Use Kafka ACLs to control which principals can view cost attribution dashboards

No changes to Kafka's authorization model are required; existing JMX security practices apply.

Compatibility, Deprecation, and Migration Plan

...

• Verify exact byte count attribution for mock remote fetches
• Test LRU eviction with max.consumerclient.groups limitsensors  limit
• Validate context propagation from FetchRequest to RemoteStorageFetchInfo
• Verify MBean unregistration upon sensor eviction

...

• 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
• Error attribution: validate that failed S3 requests are correctly attributed to client-id

Performance Tests

• Test Environment:

  • Cluster: 3 brokers, 32GB RAM, 8 vCPUs each
  • Workload: 10,000 msg/sec, 1KB message size
  • Consumers: 50 concurrent consumer groups
  • Test duration: 24 hours
  • Remote storage: AWS S3 Standard

  Targets:

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

  ---

Operational Guide

Prometheus Integration

...

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

Estimated hourly cost:Promql

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

Fetch efficiency (bytes per request):Promql

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

Error rate by client:Promql

sum(rate(kafka_server_remote_fetch_errors_total[1h])) by (client_id)

...