...

Compatibility, Deprecation, and Migration Plan

• This KIP affects StreamsMetadata and AssignmentInfo. Our changes to StreamsMetadata will be compatible with existing streams apps. AssignmentInfo changes will add a standbyPartitionsByHost map, we would need to upgrade AssignmentInfo version and we expect the existing version probing mechanism to handle the migration path.
• Two methods(KafkaStreams#metadataForKey()) are deprecated and users can migrate their code to use the two new methods(KafkaStreams#queryMetadataForKey()) instead.

Rejected Alternatives

• Adding a StreamsConfig to have a universal enableStandbyServing flag for the application. This would restrict the functionality as there could be multiple stores in an application and we need to have the flexibility to extend different consistency guarantees in such cases, which would be restricted by having a StreamsConfig.
• Making the Streams APIs lag aware e.g only return standbys within a certain time/offset lag. It did not add a lot of value to push this into Streams. Instead, the KIP keeps Streams agnostic of what acceptable values for lag is and provides the application/user flexibility to choose.
• Propagating lag information using the Rebalance protocol. Although it seemed like logical thing to do, with KIP-441 in mind, we decided against it due to various reasons. Foremost, the lag information is quickly outdated
• We decided against making multiple calls to StreamsMetadataState to fetch active tasks, standby tasks and the partition to which key belongs(as it is required to fetch lag) and have a new public class KeyQueryMetadata as it makes the code leaner and returns in a single fetch.

• Since it's sufficient to just support offsetLag initially without needing time-based lag information right away and adding time-based lag needs broker changes to be implemented efficiently, we decided against implementing time-based offsets for now. Two methods(KafkaStreams#metadataForKey()) are deprecated and users can migrate their code to use the two new methods(KafkaStreams#queryMetadataForKey()) instead.