Current state: Draft
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).
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:
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.
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.
From a business perspective, KIP-1267 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:
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.
This KIP proposes a new JMX metric group RemoteFetchMetrics with client-level attribution:
client-id, topic, error-type (optional)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.
kafka.server:type=RemoteFetchMetrics,name=RemoteFetchBytesPerSec,client-id={client_id},topic={topic_name}
|
| Configuration | Type | Default | Description |
|---|---|---|---|
| remote.log.metrics.cost.attribution.enabled | Boolean | false | Master switch to enable client-level metrics |
| Int | 1000 | Maximum unique client-ids tracked (LRU eviction) |
| remote.log.metrics.include.partition | Boolean | false | Include partition tag (increases cardinality) |
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.
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 metric group: kafka.server:type=RemoteFetchMetrics
This is separate from BrokerTopicMetrics to isolate high-cardinality client-level data.
The implementation follows a "surgical instrumentation" approach with minimal changes to the data path:
1.ReplicaManager Modification
2,RemoteLogManager Instrumentation
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
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.
This section demonstrates how KIP-1267 enables financial accountability in multi-tenant Kafka environments.
KIP-1267 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 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 RemoteFetchRequestsPerSec + RemoteFetchErrorsPerSec
• RAPI = Cloud provider's API rate (e.g., $0.0004 per 1,000 GET requests)
Organizations can implement three levels of financial maturity:
Level 1: Showback Model
Level 2: Chargeback Model
Level 3: Real-Time Cost Enforcement
KIP-1267 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.
The metrics provided by KIP-1267 can be integrated with:
The metrics introduced by this KIP expose client-id information through JMX endpoints. Organizations should:
client-id values may contain sensitive team/application identifiersNo changes to Kafka's authorization model are required; existing JMX security practices apply.
Instant rollback via dynamic configuration: set remote.log.metrics.cost.attribution.enabled=false. This immediately stops metric recording without requiring broker restart.
Unit Tests
Integration Tests
Performance Tests
Test Environment:
Targets:
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
- pattern: kafka.server<type=RemoteFetchMetrics, name=RemoteFetchRequestsPerSec, client-id=(.+), topic=(.+)><>Count
name: kafka_server_remote_fetch_requests_total
labels:
client_id: "$1"
topic: "$2"
type: COUNTER
- pattern: kafka.server<type=RemoteFetchMetrics, name=RemoteFetchErrorsPerSec, client-id=(.+), topic=(.+)><>Count
name: kafka_server_remote_fetch_errors_total
labels:
client_id: "$1"
topic: "$2"
type: COUNTER
|
Total bytes by client (30 days):
sum(increase(kafka_server_remote_fetch_bytes_total[30d])) by (client_id) |
sum(rate(kafka_server_remote_fetch_bytes_total[1h])) by (client_id) * 0.00000000009 |
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) |
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:
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.
• KIP-405: Kafka Tiered Storage
• KIP-963: Additional metrics in Tiered Storage
• KIP-714: Client Metrics and Observability