# No repartitioning with \`selectKey\`

**URL:** <https://forum.confluent.io/t/no-repartitioning-with-selectkey/9837>\
**Category:** Kafka Streams\
**Created:** [17 January 2024 10:32 UTC](https://forum.confluent.io/t/no-repartitioning-with-selectkey/9837 "2024-01-17T10:32:23Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![tharindg](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/tharindg/32/3456_2.png) [@tharindg](https://forum.confluent.io/u/tharindg)\
**Post date:** [17 January 2024 10:32 UTC](https://forum.confluent.io/t/no-repartitioning-with-selectkey/9837/1 "2024-01-17T10:32:23Z")

</div>

Consider 2 input topics `A` and `B` with the same number of partitions in each. Topic `A` gets data through a custom partitioner.

One stream will read data from `A` and write to a `persistentWindowStore`.  
Another Stream will read data from `B`, read a value based on the timestamp from the `persistentWindowStore` and publish an event to the output topic.

Due to the custom partitioning in `A` I tried to use `selectKey` to get data repartitioned but it doesn’t seem to trigger a `repartition`. This causes my logic to fail.

When I use `repartition(Repartitioned.streamPartitioner(...))` it does the repartitioning and things work properly.

I wonder why the `selectKey` doesn’t trigger a repartitioning in this case.

Code

```auto
    val stateStore = "state-store"
    val punctuatorStore = "punctuate-store"

    val builder = StreamsBuilder()

    builder.addStateStore(
        Stores.windowStoreBuilder(
            Stores.persistentWindowStore(
                stateStore,
                Duration.ofDays(2),
                Duration.ofMinutes(1),
                false,
            ),
            Serdes.String(),
            serde.stateStoreSerde,
        ).withLoggingEnabled(changeLogConfigs()),
    )

    builder.addStateStore(
        Stores.keyValueStoreBuilder(
            Stores.persistentKeyValueStore(punctuatorStore),
            Serdes.String(),
            serde.stateStoreSerde,
        ),
    )

    builder.stream(
        "A", // topic A
        Consumed.with<String, B>(Topology.AutoOffsetReset.LATEST)
            .withKeySerde(serde.string)
            .withValueSerde(serde.serdeA),
    ).selectKey { _, b -> b.name }
    .process(
        // store data into the state store
        ProcessorSupplier { StateStoreStorageProcessor(stateStore, serde.serdeA) },
        Named.`as`("storage-flow"),
        stateStore
    )

    builder.stream(
        "B", // topic B
        Consumed.with<String, A>(Topology.AutoOffsetReset.LATEST)
            .withKeySerde(serde.string)
            .withValueSerde(serde.serdeB),
    ).selectKey { _, a -> a.name }
    .process(
        // create the output with data from B and state store
        ProcessorSupplier { OutputCreateProcessor(stateStore, serde.serdeA) },
        Named.`as`("output-calculator"),
        stateStore
    ).to(
        "Output",
        Produced.with(serde.string, serde.outputSerde)
    )

    builder.build()

```

Topology description

```auto
Topologies:
   Sub-topology: 0
    Source: KSTREAM-SOURCE-0000000002 (topics: [B])
      --> KSTREAM-KEY-SELECT-0000000003
    Processor: KSTREAM-KEY-SELECT-0000000003 (stores: [])
      --> output-calculator
      <-- KSTREAM-SOURCE-0000000002
    Source: KSTREAM-SOURCE-0000000000 (topics: [A])
      --> KSTREAM-KEY-SELECT-0000000001
    Processor: KSTREAM-KEY-SELECT-0000000001 (stores: [])
      --> storage-flow
      <-- KSTREAM-SOURCE-0000000000
    Processor: output-calculator (stores: [state-store])
      --> KSTREAM-SINK-0000000004
      <-- KSTREAM-KEY-SELECT-0000000003
    Sink: KSTREAM-SINK-0000000004 (topic: Output)
      <-- output-calculator
    Processor: storage-flow (stores: [state-store])
      --> none
      <-- KSTREAM-KEY-SELECT-0000000001

```

---

<div class="post-metadata">

**Author:** ![mjsax](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mjsax/32/3113_2.png) [@mjsax](https://forum.confluent.io/u/mjsax)\
**Post date:** [18 January 2024 00:11 UTC](https://forum.confluent.io/t/no-repartitioning-with-selectkey/9837/2 "2024-01-18T00:11:54Z")

</div>

That’s by-design.

In general, `selectKey()` does not trigger a repartition directly, but it only marks the `KStream` for a potential downstream repartitioning to avoid unnecessary repartitioning. For example, you don’t want a repartition step if you do:

```auto
stream.selectKey().flatMapValue().to("outputTopic");

```

Ie, repartitioning is “lazy”. If a `KStream` is marked for repartitioning, a downstream DSL operator like `groupByKey()`, or `join()` would trigger the actual repartitioning, because they require data to be partitioned correctly.

However, PAPI-integration operators like `process()` or `transformXxx()` do not evaluate the “repartition flag” and don’t insert the repartition step. The semantics of the provided/custom `Processor/Transformer` are unknown and thus it might be unnecessary to repartition. Thus, it’s left to the user’s responsibility to `repartition()` expliclity if necessary for this case.

This behavior should actually be explained in the JavaDocs? If not, would you like to do a PR to improve the JavaDocs accordingly (or file a Jira about this so we don’t drop it on the floor)?

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/flex019/uploads/confluentcommunity/original/1X/c49438c90c9df282e9996fdf6971be890c71b65a.svg) [@system](https://forum.confluent.io/u/system)\
**Post date:** [25 January 2024 00:12 UTC](https://forum.confluent.io/t/no-repartitioning-with-selectkey/9837/3 "2024-01-25T00:12:46Z")

</div>

This topic was automatically closed 7 days after the last reply. New replies are no longer allowed.