mjsax commented on code in PR #20910:
URL: https://github.com/apache/kafka/pull/20910#discussion_r2536880272
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableProcessValues.java:
##########
@@ -90,80 +88,69 @@ public boolean enableSendingOldValues(final boolean
forceMaterialization) {
return sendOldValues;
}
- private class KTableTransformValuesProcessor extends
ContextualProcessor<K, Change<V>, K, Change<VOut>> {
- private final ValueTransformerWithKey<? super K, ? super V, ? extends
VOut> valueTransformer;
- private KeyValueStoreWrapper<K, VOut> store;
- private TimestampedTupleForwarder<K, VOut> tupleForwarder;
Review Comment:
We move both `KeyValueStoreWrapper store` and `TimestampedTupleForwarder
tupleForwarder` into the newly created `ForwardCaptureProcessorContext`.
The reason is, that when `forward` is called inside `FixedKeyProcessor`, we
want to execute the "depth first" forward directly, so we need to do it inside
`ForwardCaptureProcessorContext`. For the old "valueTransformer", the
"valueTransformer" just terminated, returning the single result `value` back.
This change in control flow actually raises a user facing API question (ie,
when we change the public user API from "transform" to "process"): in the old
code, it was ensured that `transformValues` would return a single result per
input record. Using the new `processValues` interface, user invoke
`context.forward()` and it's possible to call it more than ones. Of course,
because the key cannot be modified, the latest `forward()` call would win
(assuming the user does not use the `child` parameter; enabling the usage of
`child` parameter is a second change we would get down the line).
So the question arises, if we would really want to move to `processValues`?
Or maybe introduce something entirely new? It some new variation of the old
"transformValues" but base on the new PAPI, and tailored to the KTable case.
Asking this question right away, because it could change how we want to
setup the internal code to begin with, and could make this refactoring obsolete.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
