{"action":"created","issue":{"url":"https://api.github.com/repos/apache/kafka/issues/5619","repository_url":"https://api.github.com/repos/apache/kafka","labels_url":"https://api.github.com/repos/apache/kafka/issues/5619/labels{/name}","comments_url":"https://api.github.com/repos/apache/kafka/issues/5619/comments","events_url":"https://api.github.com/repos/apache/kafka/issues/5619/events","html_url":"https://github.com/apache/kafka/pull/5619","id":357985771,"node_id":"MDExOlB1bGxSZXF1ZXN0MjEzODY1Mjk4","number":5619,"title":"Add a scala-friendlier KStream.transform() variant","user":{"login":"redvasily","id":79625,"node_id":"MDQ6VXNlcjc5NjI1","avatar_url":"https://avatars1.githubusercontent.com/u/79625?v=4","gravatar_id":"","url":"https://api.github.com/users/redvasily","html_url":"https://github.com/redvasily","followers_url":"https://api.github.com/users/redvasily/followers","following_url":"https://api.github.com/users/redvasily/following{/other_user}","gists_url":"https://api.github.com/users/redvasily/gists{/gist_id}","starred_url":"https://api.github.com/users/redvasily/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/redvasily/subscriptions","organizations_url":"https://api.github.com/users/redvasily/orgs","repos_url":"https://api.github.com/users/redvasily/repos","events_url":"https://api.github.com/users/redvasily/events{/privacy}","received_events_url":"https://api.github.com/users/redvasily/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":2,"created_at":"2018-09-07T09:15:35Z","updated_at":"2018-09-10T21:11:15Z","closed_at":null,"author_association":"NONE","pull_request":{"url":"https://api.github.com/repos/apache/kafka/pulls/5619","html_url":"https://github.com/apache/kafka/pull/5619","diff_url":"https://github.com/apache/kafka/pull/5619.diff","patch_url":"https://github.com/apache/kafka/pull/5619.patch"},"body":"This KStream.transform() variant restores the old capability of using\r\nscala-style Transformers that return (key, value) tuple instead of\r\nreturning KeyValue instances.\r\n\r\nThis capability was present in the old Lightbend's kafka-streams-scala\r\npackage and it seems that 2.0.0 API intended to have this capability,\r\nbut it didn't work because of KAKFA-7250\r\n\r\n*More detailed description of your change,\r\nif necessary. The PR title and PR message become\r\nthe squashed commit message, so use a separate\r\ncomment to ping reviewers.*\r\n\r\n*Summary of testing strategy (including rationale)\r\nfor the feature or bug fix. Unit and/or integration\r\ntests are expected for any behaviour change and\r\nsystem tests should be considered for larger changes.*\r\n\r\n### Committer Checklist (excluded from commit message)\r\n- [ ] Verify design and implementation \r\n- [ ] Verify test coverage and CI build status\r\n- [ ] Verify documentation (including upgrade notes)\r\n"},"comment":{"url":"https://api.github.com/repos/apache/kafka/issues/comments/420062446","html_url":"https://github.com/apache/kafka/pull/5619#issuecomment-420062446","issue_url":"https://api.github.com/repos/apache/kafka/issues/5619","id":420062446,"node_id":"MDEyOklzc3VlQ29tbWVudDQyMDA2MjQ0Ng==","user":{"login":"redvasily","id":79625,"node_id":"MDQ6VXNlcjc5NjI1","avatar_url":"https://avatars1.githubusercontent.com/u/79625?v=4","gravatar_id":"","url":"https://api.github.com/users/redvasily","html_url":"https://github.com/redvasily","followers_url":"https://api.github.com/users/redvasily/followers","following_url":"https://api.github.com/users/redvasily/following{/other_user}","gists_url":"https://api.github.com/users/redvasily/gists{/gist_id}","starred_url":"https://api.github.com/users/redvasily/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/redvasily/subscriptions","organizations_url":"https://api.github.com/users/redvasily/orgs","repos_url":"https://api.github.com/users/redvasily/repos","events_url":"https://api.github.com/users/redvasily/events{/privacy}","received_events_url":"https://api.github.com/users/redvasily/received_events","type":"User","site_admin":false},"created_at":"2018-09-10T21:11:15Z","updated_at":"2018-09-10T21:11:15Z","author_association":"NONE","body":"Hi @vvcephei ,\r\n\r\nI tried this approach with two implicit conversions, and it works and it can be done with the code that looks roughly like this:\r\n\r\n`\r\nimport org.apache.kafka.streams.KeyValue\r\nimport org.apache.kafka.streams.kstream.{Transformer, TransformerSupplier}\r\nimport org.apache.kafka.streams.processor.ProcessorContext\r\n\r\nimport scala.language.implicitConversions\r\n\r\nobject TransformerImplicits {\r\n\r\n  implicit def scalaTransformerWrapper[K, V, K1, V1](\r\n    scalaTransformer: Transformer[K, V, (K1, V1)]\r\n  ): Transformer[K, V, KeyValue[K1, V1]] = {\r\n    val transformer = scalaTransformer\r\n    new Transformer[K, V, KeyValue[K1, V1]] {\r\n      override def init(context: ProcessorContext): Unit = transformer.init(context)\r\n      override def transform(key: K, value: V): KeyValue[K1, V1] = {\r\n        transformer.transform(key, value) match {\r\n          case (k1, v1) => KeyValue.pair(k1, v1)\r\n          case _ => null\r\n        }\r\n      }\r\n      override def close(): Unit = transformer.close()\r\n    }\r\n  }\r\n\r\n  implicit def supplierWrapper[K, V, K1, V1](\r\n    supplier: () => Transformer[K, V, KeyValue[K1, V1]]\r\n  ): TransformerSupplier[K, V, KeyValue[K1, V1]] = {\r\n    new TransformerSupplier[K, V, KeyValue[K1, V1]] {\r\n      override def get(): Transformer[K, V, KeyValue[K1, V1]] = supplier()\r\n    }\r\n  }\r\n} `\r\n\r\nBut I really feel that this code feels a bit alien in the current kafka-streams-scala. What I mean is that other scala.kstream.KStreamm methods accept native scala functions and convert them internally into Java objects compatible with java's KStream methods. Take for example KStream.filter(): filter(predicate: (K, V) => Boolean) it accepts a simple Scala function and only internally converts it into a Java compatible Predicate. This way a user is spared from the necessity of importing extra implicit conversions into the scope and kinda follows \"Principle of Least Power\".\r\n\r\nIf anything I would say is that it's the current (https://github.com/apache/kafka/blob/f123d2f18c55b1cf2edb452aeb87e6ad0743c292/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KStream.scala) KStream.transform() doesn't really belong there, as it's the only method (that I can see) that uses Java API directly, simply forwarding it to an underlyling Java KStream object without adapting it to Scala.\r\n\r\nFurthermore, this transform() that this pull requests restores (def transform[K1, V1](transformerSupplier: TransformerSupplier[K, V, KeyValue[K1, V1]], stateStoreNames: String*)) comes directly from the original Lightbend's KStreamS.transform() (and it looks to me that this is where the first version of Apache's kafka-streams-scala came from). Even the original version scala's KStream.transform() in 2.0.0 version tried to make scala-friendly API available, though it didn't accept the function, but the object itself, which lead us to these problems, but intention was definitely there.\r\n\r\nIn the end it looks to me that:\r\n\r\n- providing a pure Java API on the scala wrapper doesn't really mesh well with the rest of kafka-streams-scala API\r\n- having a user bring more implicit conversion into the scope (which already have implicit Serdes/Produced/Consumed) would place unnecesary burden on the user, and also wouldn't follow the pattern of other existing APIs\r\n- doesn't follow the intention of having a scala-friendly wrapper over a Java API\r\n\r\nI don't really know if the existing pure Java .transform() should be kept in the API. On one hand it's probably nice for some extra performance, as there's no need to unpack each tuple and turn it into a KeyValue instance, but on the other hand no other KStream methods have their pure Java API available directly.\r\n\r\nAnother point is that there's no need to convert () => Transformer in Scala 2.12, as Scala 2.12 directly supports SAMs. But again I don't really know how important support for Scala 2.11 is.\r\n\r\nPlease let me know what you think. As I'd like to first come to some conclusion about what kind of code would be preferable for KStream.transform() before actually making further changes and putting any progressing to KIP."}}
{"actor":{"display_login":"redvasily"}}