Versions Compared

Old Version 7

changes.mady.by.user Uladzislau Blok

Saved on

compared with

New Version 8

changes.mady.by.user Uladzislau Blok

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

Code Block
languagejava
titleStreamPartitioner
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);
    }
}


Code Block
languagejava
titleKafkaStreams
    // already existing method
    public <K> KeyQueryMetadata queryMetadataForKey(final String storeName,
                                                    final K key,
                                                    final Serializer<K> keySerializer) {
        validateIsRunningOrRebalancing();
        return streamsMetadataState.keyQueryMetadataForKey(storeName, key, new RecordHeaders(), keySerializer);
    }

    // already existing method 
    public <K> KeyQueryMetadata queryMetadataForKey(final String storeName,
                                                    final K key,
                                                    final StreamPartitioner<? super K, ?> partitioner) {
        validateIsRunningOrRebalancing();
        return streamsMetadataState.keyQueryMetadataForKey(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. Existing header-less methods will propagate empty headers to the underlying partitioner logicExisting 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.
  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

...