• Key access to ValueTransformer:  While transforming values via KStream.transformValues and ValueTransformer, the key associated with the value may be needed, even if it is not changed. For instance, it may be used to access stores.

  As of now, the key is not available within these methods and interfaces, leading to the use of KStream.transform and Transformer, and the unnecessary creation of new KeyValue objects.

• Key access to ValueMapper: ValueMapper should have read-only access to the key for the value it is mapping. Sometimes the value transformation will depend on the key.

  It is possible to do this with a full blown KeyValueMapper but that loses the promise that you won't change the key – so you might introduce a re-keying phase that is totally unnecessary.

• Key access to ValueJoiner interface: In working with Kafka Stream joining, it's sometimes the case that a join key is not actually present in the values of the joins themselves (if, for example, a previous transform generated an ephemeral join key.) In such cases, the actual key of the join is not available in the ValueJoiner implementation to be used to construct the final joined value. This can be worked around by explicitly threading the join key into the value if needed, but it seems like extending the interface to pass the join key along as well would be helpful

Public Changes

There are no public changes to ensure the backwards-compatibility. 

Proposed Changes

1. We extend the target interfaces ValueJoiner,  ValueTransformer, and ValueMapper as ValueJoinerWithKey,  ValueTransformerWithKey, and ValueMapperWithKey. In extended abstract classes we have an access to keys.
2. In Processor we check the actual instance of object:
    

   this.valueTransformer = valueTransformer;
   if (valueTransformer instanceof ValueTransformerWithKey) {
       isTransformerWithKey = true;
   } else {
       isTransformerWithKey = false;
   }

   ..............

   ..............

   @Override
   public void process(K key, V value) {
       if (isTransformerWithKey) {
           K keyCopy = (K) Utils.deepCopy(key);
     context.forward(key, ((ValueTransformerWithKey<K, V, R>) valueTransformer).transform(key, value));
     key = keyCopy;
     } else {
           context.forward(key, valueTransformer.transform(value));
     }
   }

   

3. As we can see from the above code snippet, we can guard the key change in Processors by deeply copying the object before calling the apply() method.

The PR can be found here.

Test Plan

The unit tests are changed accordingly to support the changes in core classes.

Rejected Alternatives

1. Not backward-compatible

We propose adding key information for ValueJoiner,  ValueTransformer, and ValueMapper classes and their apply(...) methods.

As a result, we perform the following public changes (and their overloaded versions)