fix(inkless:consolidation): route DeleteRecords to the real leader and report cross-tier low watermark [KC-332]#710
Conversation
There was a problem hiding this comment.
Pull request overview
Fixes DeleteRecords behavior for switched/consolidating diskless topics by (1) forwarding the leader-only local-log portion to the partition’s real KRaft leader when clients are routed to a follower, and (2) reporting a broker-agnostic low watermark based on the control-plane cross-tier earliest offset so DeleteRecords and ListOffsets(EARLIEST) agree across brokers.
Changes:
- Add
DisklessDeleteRecordsForwarderand wire it throughBrokerServer→KafkaApisto fan outDeleteRecordsto the real leader when needed. - Update
ReplicaManager.deleteRecordsto avoid ISR-gated completion for consolidating partitions and to advance/report the control-plane cross-tier earliest as the low watermark. - Add/extend unit tests covering cross-tier low watermark behavior and forwarding routing.
Reviewed changes
Copilot reviewed 7 out of 7 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| core/src/main/scala/kafka/server/ReplicaManager.scala | Advances cross-tier earliest on successful consolidating deletes, overrides reported low watermark, and adds cross-tier earliest accessors. |
| core/src/main/scala/kafka/server/KafkaApis.scala | Routes DeleteRecords locally vs forwarded-to-leader buckets, executes them in parallel, and merges results. |
| core/src/main/scala/kafka/server/DisklessDeleteRecordsForwarder.scala | New inter-broker forwarder to send leader-only DeleteRecords leg to the real KRaft leader. |
| core/src/main/scala/kafka/server/BrokerServer.scala | Instantiates/starts/stops the forwarder and passes it into KafkaApis. |
| core/src/main/java/kafka/server/builders/KafkaApisBuilder.java | Updates constructor call to include the new optional forwarder parameter. |
| core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala | Adds unit tests for consolidating delete low watermark behavior and cross-tier earliest cache/control-plane fallback semantics. |
| core/src/test/scala/unit/kafka/server/DisklessDeleteRecordsForwarderTest.scala | New unit tests validating routing/forwarding behavior and error handling. |
de0693d to
2549301
Compare
2549301 to
634299e
Compare
| requests | ||
| } | ||
|
|
||
| private def buildRequest(pending: PendingForward, now: Long): RequestAndCompletionHandler = { |
There was a problem hiding this comment.
We don't preserve the principal when we forward these requests. This may result in TOPIC_AUTHORIZATION_FAILED when the request is performed from the broker. AFAIU Kafka has the envelop mechanism for preserving the original principal
There was a problem hiding this comment.
Well, we use this class in KafkaApis.handleDeleteRecordsRequest, which does an authorization, so it should be gated. Regarding the TOPIC_AUTHORIZATION_FAILED error I'm not sure how can we get this if we forward with the brokers own ID? Should we assume that broker principals may not have permission to delete records?
There was a problem hiding this comment.
Should we assume that broker principals may not have permission to delete records?
I assumed this 🤔 I mean, it's possible to configure brokers to have only CLUSTER_ACTION, for example, which won't allow DeleteRecords.
There was a problem hiding this comment.
We can address this as a followup, though
There was a problem hiding this comment.
Yes, that's probably a bigger change, we can keep this PR to the minimal.
634299e to
bbebd3d
Compare
…d report cross-tier low watermark [KC-332] For a switched consolidating diskless topic the metadata transformer advertises an AZ-selected replica (a follower) as the client-facing leader, so a DeleteRecords request looped on NOT_LEADER_OR_FOLLOWER and timed out. DisklessDeleteRecordsForwarder forwards the leader-only leg to the partition's real KRaft leader; KafkaApis fans the request out into local and forwarded legs and combines the results (KafkaApisBuilder / BrokerServer wire the forwarder). Completion no longer waits on the ISR low watermark for consolidating partitions (their follower log starts are frozen at the switch). Instead ReplicaManager.deleteRecords advances the control-plane cross-tier earliest and reports it as the low watermark, so DeleteRecords and a subsequent ListOffsets(EARLIEST) agree on every broker. Co-authored-by: Cursor <cursoragent@cursor.com>
bbebd3d to
e4eac32
Compare
For a switched consolidating diskless topic the metadata transformer advertises an AZ-selected replica (a follower) as the client-facing leader, so a DeleteRecords request looped on NOT_LEADER_OR_FOLLOWER and timed out. DisklessDeleteRecordsForwarder forwards the leader-only leg to the partition's real KRaft leader; KafkaApis fans the request out into local and forwarded legs and combines the results (KafkaApisBuilder / BrokerServer wire the forwarder).
Completion no longer waits on the ISR low watermark for consolidating partitions (their follower log starts are frozen at the switch). Instead ReplicaManager.deleteRecords advances the control-plane cross-tier earliest and reports it as the low watermark, so DeleteRecords and a subsequent ListOffsets(EARLIEST) agree on every broker. Adds the shared crossTierEarliestOffset / isConsolidatingDisklessPartition accessors.