This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.

Status

Current state: DraftUnder Discussion

Discussion thread: https://lists.apache.org/thread/7y7svyp3f560fzv1bgcr893vn258cn06

JIRA:

...

We propose adding a new method to the StreamPartitioner interface. To maintain backward compatibility, this will be a default method that delegates to the existing partitions method.

org.apache.kafka.streams.processor.StreamPartitioner

...

    /**
     * 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 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}
     */
    default Optional<Set<Integer>> partitions(String topic, K key, V value, Headers headers, int numPartitions) {
        return partitions(topic, key, value, numPartitions);
    }
}
```

Proposed Changes

Describe the new thing you want to do in appropriate detail. This may be fairly extensive and have large subsections of its own. Or it may be a few sentences. Use judgement based on the scope of the change.

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

...

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},
     * 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);
    }
}

    // already existing method
    public <K> KeyQueryMetadata queryMetadataForKey(final String storeName,
                                                    final K key,
                                                    final Serializer<K> keySerializer) {
        return queryMetadataForKey(storeName, key, new RecordHeaders(), keySerializer);
    }

    // already existing method 
    public <K> KeyQueryMetadata queryMetadataForKey(final String storeName,
                                                    final K key,
                                                    final StreamPartitioner<? super K, ?> partitioner) {
        return queryMetadataForKey(storeName, key, new RecordHeaders(), partitioner);
    }

    /**
     * Finds the metadata containing the active hosts and standby hosts where the key being queried would reside.
     *
     * @param storeName     the {@code storeName} to find metadata for
     * @param key           the key to find metadata for
     * @param headers       the record headers
     * @param keySerializer serializer for the key
     * @param <K>           key type
     * Returns {@link KeyQueryMetadata} containing all metadata about hosting the given key for the given store,
     * or {@code null} if no matching metadata could be found.
     */
    public <K> KeyQueryMetadata queryMetadataForKey(final String storeName,
                                                    final K key,
                                                    final Headers headers,
                                                    final Serializer<K> keySerializer) {
        validateIsRunningOrRebalancing();
        return streamsMetadataState.keyQueryMetadataForKey(storeName, key, headers, keySerializer);
    }

    /**
     * Finds the metadata containing the active hosts and standby hosts where the key being queried would reside.
     *
     * @param storeName     the {@code storeName} to find metadata for
     * @param key           the key to find metadata for
	 * @param headers       the record headers
     * @param partitioner   the partitioner to be used to locate the host for the key
     * @param <K>           key type
     * Returns {@link KeyQueryMetadata} containing all metadata about hosting the given key for the given store, using
     * the supplied partitioner, or {@code null} if no matching metadata could be found.
     */
    public <K> KeyQueryMetadata queryMetadataForKey(final String storeName,
                                                    final K key,
												    final Headers headers,
                                                    final StreamPartitioner<? super K, ?> partitioner) {
        validateIsRunningOrRebalancing();
        return streamsMetadataState.keyQueryMetadataForKey(storeName, key, headers, partitioner);
    }

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. No changes are planed for KafkaStreamsNamedTopologyWrapper. This class and entire feature are deprecated, so no changes are planed there
4. Internal Callers and Propagation:
   1. RecordCollectorImpl#send to pass record headers when calling the partitioner.
   2. StreamsMetadataState to support the new header-aware lookups.
5. 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

• 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.