Select keys from value exclusively (#364)

Author: philipp94831

streams-bootstrap-core/src/main/java/com/bakdata/kafka/streams/kstream/KStreamX.java

@@ -83,6 +83,32 @@ public interface KStreamX<K, V> extends KStream<K, V> {
     @Override
     <KR> KStreamX<KR, V> selectKey(KeyValueMapper<? super K, ? super V, ? extends KR> mapper, Named named);
 
+    /**
+     * Set a new key (with possibly new type) for each input record. The provided {@link ValueMapper} is applied to each
+     * input record and computes a new key for it. Thus, an input record {@code <K,V>} can be transformed into an output
+     * record {@code <K':V>}. This is a stateless record-by-record operation.
+     *
+     * @param mapper a {@link ValueMapper} that computes a new key for each record
+     * @param <KR> the new key type of the result stream
+     * @return a {@code KStream} that contains records with new key (possibly of different type) and unmodified value
+     * @see #selectKey(KeyValueMapper)
+     */
+    <KR> KStreamX<KR, V> selectKey(ValueMapper<? super V, ? extends KR> mapper);
+
+    /**
+     * Set a new key (with possibly new type) for each input record.
+     * The provided {@link ValueMapper} is applied to each input record and computes a new key for it.
+     * Thus, an input record {@code <K,V>} can be transformed into an output record {@code <K':V>}.
+     * This is a stateless record-by-record operation.
+     *
+     * @param mapper a {@link ValueMapper} that computes a new key for each record
+     * @param named  a {@link Named} config used to name the processor in the topology
+     * @param <KR>   the new key type of the result stream
+     * @return a {@code KStream} that contains records with new key (possibly of different type) and unmodified value
+     * @see #selectKey(KeyValueMapper, Named)
+     */
+    <KR> KStreamX<KR, V> selectKey(ValueMapper<? super V, ? extends KR> mapper, Named named);
+
     @Override
     <KR, VR> KStreamX<KR, VR> map(
             KeyValueMapper<? super K, ? super V, ? extends KeyValue<? extends KR, ? extends VR>> mapper);

streams-bootstrap-core/src/main/java/com/bakdata/kafka/streams/kstream/KStreamXImpl.java

@@ -102,6 +102,16 @@ public <KR> KStreamX<KR, V> selectKey(final KeyValueMapper<? super K, ? super V,
         return this.context.wrap(this.wrapped.selectKey(mapper, named));
     }
 
+    @Override
+    public <KR> KStreamX<KR, V> selectKey(final ValueMapper<? super V, ? extends KR> mapper) {
+        return this.selectKey((k, v) -> mapper.apply(v));
+    }
+
+    @Override
+    public <KR> KStreamX<KR, V> selectKey(final ValueMapper<? super V, ? extends KR> mapper, final Named named) {
+        return this.selectKey((k, v) -> mapper.apply(v), named);
+    }
+
     @Override
     public <KR, VR> KStreamX<KR, VR> map(
             final KeyValueMapper<? super K, ? super V, ? extends KeyValue<? extends KR, ? extends VR>> mapper) {

streams-bootstrap-core/src/test/java/com/bakdata/kafka/streams/kstream/KStreamXTest.java

@@ -898,6 +898,46 @@ public void buildTopology(final StreamsBuilderX builder) {
         }
     }
 
+    @Test
+    void shouldSelectKeyFromValue() {
+        final StringApp app = new StringApp() {
+            @Override
+            public void buildTopology(final StreamsBuilderX builder) {
+                final KStreamX<String, String> input = builder.stream("input");
+                input.selectKey(v -> v).to("output");
+            }
+        };
+        try (final TestTopology<String, String> topology = app.startApp()) {
+            topology.input()
+                    .add("foo", "bar");
+            topology.streamOutput()
+                    .expectNextRecord()
+                    .hasKey("bar")
+                    .hasValue("bar")
+                    .expectNoMoreRecord();
+        }
+    }
+
+    @Test
+    void shouldSelectKeyFromValueNamed() {
+        final StringApp app = new StringApp() {
+            @Override
+            public void buildTopology(final StreamsBuilderX builder) {
+                final KStreamX<String, String> input = builder.stream("input");
+                input.selectKey(v -> v, Named.as("select")).to("output");
+            }
+        };
+        try (final TestTopology<String, String> topology = app.startApp()) {
+            topology.input()
+                    .add("foo", "bar");
+            topology.streamOutput()
+                    .expectNextRecord()
+                    .hasKey("bar")
+                    .hasValue("bar")
+                    .expectNoMoreRecord();
+        }
+    }
+
     @Test
     void shouldMapCapturingErrors() {
         final KeyValueMapper<String, String, KeyValue<String, String>> mapper = mock();