...

public interface StreamPartitioner<K, V> {

       Optional<Set<Integer>> partitions(String topic, K key, V value, int numPartitions);

    /**
        * Determine the number(s) of the partition(s) to which a record with the given key and value should be sent,
        * for the given topic and current partition count.
      * 
     * @param topic the topic name this record is sent to
        * @param key the key of the record
        * @param value the value of the record
     * @param headers the record headers
        * @param numPartitions the total number of partitions
        * @return an Optional of Set of integers between 0 and {@code numPartitions-1},
     */
    Optional<Set<Integer>> partitions(String topic, K key, V value, int numPartitions);

    /**
     * Determine the number(s) of the partition(s) to which a record with the given key, value, and headers should be sent,
     * for the given topic and current partition count.
     * 
     * @param topic the topic name this record is sent to
     * @param key the key of the record
     * @param value the value of the record
     * @param headers the headers of the record
     * @param numPartitions the total number of partitions
     * @return an Optional of Set of integers between 0 and {@code numPartitions-1}
     */
   * Empty optional means using default partitioner
     * Optional of an empty set means the record won't be sent to any partitions i.e drop it.
     * Optional of Set of integers means the partitions to which the record should be sent to.
     * */
    default Optional<Set<Integer>> partitions(String topic, K key, V value, Headers headers, int numPartitions) {
            return partitions(topic, key, value, numPartitions);
      }
}

Proposed Changes

...

1. Interface Enhancement: Add the headers-aware partitions method to StreamPartitioner. This is a default method to ensure binary compatibility.
2. Public API for Interactive Queries: Add new overloads to KafkaStreams#queryMetadataForKey that accept `Headers`. This allows users to perform metadata lookups for header-dependent partitioning strategies
3. Internal Callers and Propagation:
   1. RecordCollectorImpl#send to pass record headers when calling the partitioner.
   2. StreamsMetadataState to support the new header-aware lookups. Existing header-less methods will propagate empty headers to the underlying partitioner logic
   3. Existing public methods in KafkaStreams and StreamsMetadataState that do not accept headers will remain. They will propagate empty headers to the underlying header-aware partitioner methods to ensure consistent interface usage.
4. Built-in Implementations:

   1. WindowedStreamPartitioner: Override the new method to propagate headers to the underlying WindowedSerializer#serializeBaseKey.

   2. DefaultStreamPartitioner: Override the new method to propagate headers to the keySerializer#serialize

Compatibility, Deprecation, and Migration Plan

• What impact (if any) will there be on existing users?
• If we are changing behavior how will we phase out the older behavior?
• If we need special migration tools, describe them here.
• When will we remove the existing behavior?

Test Plan

Describe in few sentences how the KIP will be tested. We are mostly interested in system tests (since unit-tests are specific to implementation details). How will we know that the implementation works as expected? How will we know nothing broke?

Rejected Alternatives

...

• Backward Compatibility: change is fully backward compatible. Existing custom StreamPartitioner implementations will continue to work as they will use the default implementation of the new method.
• Deprecation: No methods are being deprecated in this KIP.
• Migration: Users wishing to leverage headers in their custom partitioning logic should migrate their StreamPartitioner implementations to override the new `partitions` method.

Test Plan

• Unit Tests and Integration tests:
  • Verify Partitioners correctly propagate headers.
  • Verify propagated headers are used to determine target partition

Rejected Alternatives

• Updated the existing method without default: Rejected as it would break all existing user implementations.