Pankraz76 opened a new pull request, #23947: URL: https://github.com/apache/kafka/pull/23947
### Problem Since the `KRaftMetadataCache` rewrite, `getPartitionLeaderEndpoint` returns `Optional.empty()` when a partition exists but its leader is not a registered broker, most importantly when the partition is offline (leader `-1`). `TransactionMarkerChannelManager` interprets `Optional.empty()` as "partition deleted": it removes the partition from the transaction and completes the transaction without writing the marker. A commit/abort marker for an offline partition, including one re-enqueued after its leader failed, is therefore silently dropped. The partition is left with a hanging transaction and the commit marker is lost, so `read_committed` consumers stall on it. ### Root cause A regression from the `KRaftMetadataCache` rewrite in [KAFKA-17573](https://issues.apache.org/jira/browse/KAFKA-17573). The previous Scala implementation returned `Node.noNode()` when the leader had no registered broker: ```scala case Some(partition) => Option(image.cluster().broker(partition.leader)) match { case None => Some(Node.noNode) case Some(broker) => Some(broker.node(listenerName.value()).orElse(Node.noNode())) } ``` The Java port mapped that case to `Optional.empty()`: ```java .flatMap(partition -> Optional.ofNullable(image.cluster().broker(partition.leader)) .map(broker -> broker.node(listenerName.value()).orElse(Node.noNode()))); ``` ### Fix If the partition exists, `getPartitionLeaderEndpoint` returns the leader's endpoint for the listener, or `Node.noNode()` if there is none (leader offline, not registered, or without that listener). It returns `Optional.empty()` only if the topic or partition does not exist. The Javadoc on `MetadataCache` and `KRaftMetadataCache` is updated accordingly. ### Testing - `MetadataCacheTest#testGetPartitionLeaderEndpoint`: leader with endpoint, leader without the listener, offline partition, unregistered leader, unknown partition, unknown topic. - `TransactionMarkerChannelManagerTest#shouldSaveForLaterWhenPartitionIsOffline`: with a real `KRaftMetadataCache`, the marker for an offline partition is kept in the unknown-broker queue, survives a retry, and the transaction is not completed. - `TransactionsTest#testCommitMarkerIsWrittenAfterPartitionComesBackOnline` (integration): commits while a partition is offline, brings it back, and verifies a `read_committed` consumer sees the committed record. All three fail without the fix and pass with it. ### Compatibility / risk This restores the pre-KAFKA-17573 contract. All callers were reviewed: - `TransactionMarkerChannelManager`: markers for offline partitions are parked and sent once a leader is available, instead of being dropped. `Node.noNode()` is never used as a destination. - `NetworkPartitionMetadataClient`: already treats empty and `Node.noNode()` identically (`LEADER_NOT_AVAILABLE`). - `ShareCoordinatorMetadataCacheHelperImpl` / share-group DLQ: already rejects both `null` and `Node.noNode()` leaders. While a partition stays offline, its parked marker is re-checked on every send-thread iteration. This is the existing unknown-broker queue behavior, the same as before KAFKA-17573 and the same as today for a leader without the listener. JIRA: [KAFKA-20734](https://issues.apache.org/jira/browse/KAFKA-20734) Reported by Alyssa Huang. The issue was identified, and the fix sketched, by Artem Livshits. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
