[ https://issues.apache.org/jira/browse/KAFKA-7079?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Hashan Gayasri Udugahapattuwa updated KAFKA-7079: ------------------------------------------------- Description: ValueTransformer#transform does not pass the key KStream#transformValues(ValueTransformerWithKeySupplier ..... method is not documented. It might lead to people to use workarounds or fall back to using Transformer. This is very likely if the user is using a wrapper API (i.e: for Scala) as the user would be checking the documentation more than the available API functions in code. ---- *Original issue (as it might be useful as a business requirement)* ValueTransformers' transform method doesn't pass the key to user-code. Reporting this as a bug since it currently requires workarounds. Context: I'm currently in the process of converting two stateful "*aggregate*" DSL operations to the Processor API since the state of those operations are relatively large and takes 99% + of CPU time (when profiled) for serializing and deserializing them via Kryo. Since DSL aggregations use state stores of [Bytes, Array[Byte]]] even when using the in-memory state store, it seems like the only way to reduce the serialization/deserialization overhead is to convert heavy aggregates to *transform*s. In my case, *ValueTransformer* seems to be the option. However, since ValueTransformers' _transform_ method only exposes the _value_, I'd either have to pre-process and add the key to the value or use *Transformer* instead (which is not my intent). As internal _*InternalValueTransformerWithKey*_ already has the readOnlyKey, it seems like a good idea to pass the key to the transform method as well, esp since in a stateful transformation, generally the state store has to be queried by the key. was: ValueTransformers' transform method doesn't pass the key to user-code. Reporting this as a bug since it currently requires workarounds. Context: I'm currently in the process of converting two stateful "*aggregate*" DSL operations to the Processor API since the state of those operations are relatively large and takes 99% + of CPU time (when profiled) for serializing and deserializing them via Kryo. Since DSL aggregations use state stores of [Bytes, Array[Byte]]] even when using the in-memory state store, it seems like the only way to reduce the serialization/deserialization overhead is to convert heavy aggregates to *transform*s. In my case, *ValueTransformer* seems to be the option. However, since ValueTransformers' _transform_ method only exposes the _value_, I'd either have to pre-process and add the key to the value or use *Transformer* instead (which is not my intent). As internal _*InternalValueTransformerWithKey*_ already has the readOnlyKey, it seems like a good idea to pass the key to the transform method as well, esp since in a stateful transformation, generally the state store has to be queried by the key. Summary: ValueTransformerWithKeySupplier is not mentioned in the documentation (was: ValueTransformer#transform does not pass the key) > ValueTransformerWithKeySupplier is not mentioned in the documentation > --------------------------------------------------------------------- > > Key: KAFKA-7079 > URL: https://issues.apache.org/jira/browse/KAFKA-7079 > Project: Kafka > Issue Type: Bug > Components: streams > Affects Versions: 1.1.0 > Environment: Fedora 27 > Reporter: Hashan Gayasri Udugahapattuwa > Priority: Major > > ValueTransformer#transform does not pass the key > KStream#transformValues(ValueTransformerWithKeySupplier ..... method is not > documented. It might lead to people to use workarounds or fall back to using > Transformer. This is very likely if the user is using a wrapper API (i.e: for > Scala) as the user would be checking the documentation more than the > available API functions in code. > > > > > ---- > *Original issue (as it might be useful as a business requirement)* > ValueTransformers' transform method doesn't pass the key to user-code. > Reporting this as a bug since it currently requires workarounds. > Context: > I'm currently in the process of converting two stateful "*aggregate*" DSL > operations to the Processor API since the state of those operations are > relatively large and takes 99% + of CPU time (when profiled) for serializing > and deserializing them via Kryo. > Since DSL aggregations use state stores of [Bytes, Array[Byte]]] even when > using the in-memory state store, it seems like the only way to reduce the > serialization/deserialization overhead is to convert heavy aggregates to > *transform*s. > In my case, *ValueTransformer* seems to be the option. However, since > ValueTransformers' _transform_ method only exposes the _value_, I'd either > have to pre-process and add the key to the value or use *Transformer* instead > (which is not my intent). > > As internal _*InternalValueTransformerWithKey*_ already has the readOnlyKey, > it seems like a good idea to pass the key to the transform method as well, > esp since in a stateful transformation, generally the state store has to be > queried by the key. -- This message was sent by Atlassian JIRA (v7.6.3#76005)