{"action":"created","issue":{"url":"https://api.github.com/repos/akka/akka/issues/25394","repository_url":"https://api.github.com/repos/akka/akka","labels_url":"https://api.github.com/repos/akka/akka/issues/25394/labels{/name}","comments_url":"https://api.github.com/repos/akka/akka/issues/25394/comments","events_url":"https://api.github.com/repos/akka/akka/issues/25394/events","html_url":"https://github.com/akka/akka/issues/25394","id":345110060,"node_id":"MDU6SXNzdWUzNDUxMTAwNjA=","number":25394,"title":"Carrying element context through streams","user":{"login":"jroper","id":105833,"node_id":"MDQ6VXNlcjEwNTgzMw==","avatar_url":"https://avatars0.githubusercontent.com/u/105833?v=4","gravatar_id":"","url":"https://api.github.com/users/jroper","html_url":"https://github.com/jroper","followers_url":"https://api.github.com/users/jroper/followers","following_url":"https://api.github.com/users/jroper/following{/other_user}","gists_url":"https://api.github.com/users/jroper/gists{/gist_id}","starred_url":"https://api.github.com/users/jroper/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jroper/subscriptions","organizations_url":"https://api.github.com/users/jroper/orgs","repos_url":"https://api.github.com/users/jroper/repos","events_url":"https://api.github.com/users/jroper/events{/privacy}","received_events_url":"https://api.github.com/users/jroper/received_events","type":"User","site_admin":false},"labels":[{"id":243500627,"node_id":"MDU6TGFiZWwyNDM1MDA2Mjc=","url":"https://api.github.com/repos/akka/akka/labels/discuss","name":"discuss","color":"bfe5bf","default":false},{"id":93144690,"node_id":"MDU6TGFiZWw5MzE0NDY5MA==","url":"https://api.github.com/repos/akka/akka/labels/t:stream","name":"t:stream","color":"cccccc","default":false}],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":2,"created_at":"2018-07-27T06:47:20Z","updated_at":"2018-08-01T14:08:24Z","closed_at":null,"author_association":"CONTRIBUTOR","body":"This issue is intended as a discussion about whether it makes sense to have a per element context, and how that might be implemented. I don't know if it would be a good idea to provide this or not, but I was thinking about it while riding to work today so I thought I'd record it somewhere.\r\n\r\nConsider Kafka streams. I might be doing a stream transformation that looks like this:\r\n\r\n```java\r\nKStream<String, Message> stream = ...;\r\nKStream<String, Transformed> transformed = stream\r\n  .filter((key, message) -> message.isInteresting())\r\n  .map((key, message) -> KeyValue.pair(key, transform(message)));\r\n```\r\n\r\nNow the important thing to realise here is that when the transformed messages are emitted, they are associated with the original message that came in, and this is so that once the transformed messages are emitted, the original message that caused it can be acknowledged, giving at least once delivery guarantees. This only works because Kafka streams tracks which incoming message is associated with each outgoing message - it carries this context through the stream processing. In Akka streams we can't do this, we only have two options, either require users to explicitly acknowledge errors (which in practice is error prone because you can't for example handle and then acknowledge a message in the same mapAsync since it could allow an acknowledgement of a later message to overtake the failure of an earlier message), or we can offer an API that allows you to do 1:1 transformations on messages, and carry that context around the transformation making the assumption that we can zip it back together because it's a 1:1 transformation, but as soon as the user does a transformation that isn't 1:1 like `filter` or `mapConcat`, it breaks down. We already know from experience offering APIs like this in Lagom that this causes confusion and bugs for users.\r\n\r\nSo the question is, can we implement some form of message context, which would then allow us to offer streaming APIs for message broker transformation that are as simple and intuitive to use as Kafka Streams? I think we can.\r\n\r\nNow of course, there are times when it's impossible to carry this context. For example, the moment you insert a reactive streams processor into the flow, you're going to lose the context. There's nothing we can do about that. But, from code that I've seen out in the wild, I've actually never even seen someone do that. It's all `map`, `mapAsync`, `filter`, `grouped`, `mapConcat`, all operations that can carry context (`grouped` is an interesting one that I'll get to later). So, I think it's fine to have such a limitation, and to just allow the context to be dropped in certain situations.\r\n\r\nSo let's take a first shot at how such an API might look:\r\n\r\n```scala\r\ntrait FlowOps[In] {\r\n  def viaWithContext[Out](flow: Flow[In, Out, _]): Flow[In, (Option[In], Out), NotUsed]\r\n}\r\n```\r\n\r\nThis method would be used by a library, that took a user supplied `flow`, and injected it into its stream. For each element emitted by the user supplied flow, it will then, if it can determine the incoming message context, associate it with the output message, and emit a tuple of the two. How might this be implemented in practice? Considering a graph stage logic, whenever an `onPush` callback was invoked, the current context message (if such context exists) that caused the push would be set as the context for that stage, and each `push` that that stage did from then until the next `onPush` was invoked would have the current context message for that stage set as the context for the next down stream `onPush`. For simple stages like `map`/`filter`, this will just work. For more complex stages, such as `mapAsync`, the stage will need to be aware of context - it will need to capture the current context when it dispatches the processing, and then re-associate the captured context for that operation with the stage when it handles the asynchronous callbacks. For something like `grouped`, it means that only the context that triggered the group to be emitted will be carried as the context - for the Kafka streams use case, this is exactly what we want, as when we group, we will end up only wanting to commit the last message in the group. For merges (eg `flatMapConcat` or `flatMapMerge`), each input context will need to be associated with its corresponding sub streams.\r\n\r\nNested `viaWithContext`'s would be a challenge, we could probably just disallow them.\r\n\r\nNow there is one problem with the above in practice when applied to our use case of providing a Kafka Streams like API - that is, when a stage emits multiple elements per input element, you only want to acknowledge the incoming element for the last element associated with that incoming element from the stage, but there's no way to know when you receive an element whether any further elements are going to be received for that context. One possible solution to this might be to allow stages to signal, when they push, whether the this terminates the incoming context or not. They would only signal termination if the incoming context had the termination signal attached to it. Sometimes, there might be a message to go with the terminating push (eg, 1:1 stages like `map`, stages where all output elements are known like `mapConcat`), but at other times, like `flatMapConcat`, this signal will have to be pushed without a corresponding output element. Also, a stage like `filter` will need to push the terminate signal if it filters out a message when the current incoming context has the terminate signal associated with it, otherwise the terminate signal will be dropped. An API for this might look like this:\r\n\r\n```scala\r\nsealed trait Context[+T]\r\ncase object NoContext extends Context[Nothing]\r\ncase class TerminateContext[+T](element: T) extends Context[T]\r\ncase class ContinueContext[+T](element: T) extends Context[T]\r\n\r\ntrait FlowOps[In] {\r\n  def viaWithContext[Out](flow: Flow[In, Out, _]): Flow[In, (Context[In], Option[Out]), NotUsed]\r\n}\r\n```\r\n\r\nNow all of this is very complex, there is another completely different possibility for how this might be implemented underneath, and that is to use a different materializer for the nested stages. I don't understand the materialization process of Akka streams well enough to be able to know how feasible this is, but the idea would be essentially to walk through the graph, and manually transform each stage so that it carries context. Support would need to be implemented specific to each stage, but an advantage here is that it could fail if there was a stage that didn't support carrying context (and in this case, no custom graph stage would support carrying context). So a transformation for `Map` might look like this:\r\n\r\n```scala\r\ncase Map(f) => Map {\r\n  case (ctx, None) => (ctx, None)\r\n  case (ctx, Some(t)) => (ctx, Some(f(t))))\r\n}\r\n```\r\n\r\nWhile a transformation of `Filter` might look like this:\r\n\r\n```scala\r\ncase Filter(predicate) => Collect {\r\n  case (ctx, None) => (ctx, None)\r\n  case (ctx: TerminateContext, t) => (ctx, t.filter(predicate))\r\n  case (ctx, o @ Some(t)) if predicate(t) => (ctx, o)\r\n}\r\n```\r\n\r\nA hybrid of the two approaches could also be done where graph stages could declare a `viaWithContext` version of themselves."},"comment":{"url":"https://api.github.com/repos/akka/akka/issues/comments/409587677","html_url":"https://github.com/akka/akka/issues/25394#issuecomment-409587677","issue_url":"https://api.github.com/repos/akka/akka/issues/25394","id":409587677,"node_id":"MDEyOklzc3VlQ29tbWVudDQwOTU4NzY3Nw==","user":{"login":"jrudolph","id":9868,"node_id":"MDQ6VXNlcjk4Njg=","avatar_url":"https://avatars1.githubusercontent.com/u/9868?v=4","gravatar_id":"","url":"https://api.github.com/users/jrudolph","html_url":"https://github.com/jrudolph","followers_url":"https://api.github.com/users/jrudolph/followers","following_url":"https://api.github.com/users/jrudolph/following{/other_user}","gists_url":"https://api.github.com/users/jrudolph/gists{/gist_id}","starred_url":"https://api.github.com/users/jrudolph/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jrudolph/subscriptions","organizations_url":"https://api.github.com/users/jrudolph/orgs","repos_url":"https://api.github.com/users/jrudolph/repos","events_url":"https://api.github.com/users/jrudolph/events{/privacy}","received_events_url":"https://api.github.com/users/jrudolph/received_events","type":"User","site_admin":false},"created_at":"2018-08-01T14:08:24Z","updated_at":"2018-08-01T14:08:24Z","author_association":"MEMBER","body":"By now it's pretty clear that such a functionality would be useful in many cases. On the other hand it's also pretty clear that a general solution is impossible or very hard to do requiring support in every single stage.\r\n\r\nThe problem I see with attempting a general solution is that apart from the few absolutely clear cases, many cases may introduce another configuration option for the stage because there might not be the one clear-cut way to attach the context. `grouped` would be one such example where taking the last context makes sense for Kafka but might be the wrong choice for other use cases.\r\n\r\nFor me that sounds like we shouldn't attempt a general solution because the trade-off between general benefit and overall complexity doesn't seem favorable for adding this functionality in the core (i.e. GraphStage).\r\n\r\nWhen we talked about it we thought about a `SubStream`-like approach, i.e. we could try to invent a subset API that works exactly for the cases that we know would be useful for Kafka. The `viaWithContext` API you suggest seems to be similar but wouldn't prevent building \"illegal\" flows but would fail at runtime when a component of the flow cannot carry the context.\r\n\r\nE.g. like this:\r\n\r\n```scala\r\nFlow[In]\r\n  .withContext\r\n     .map(...)\r\n     .filter(...)\r\n     .mapAsync(...)\r\n  .provideContext // (Out, In)\r\n```\r\n\r\nThe good thing would be that it could be implemented in a quite non-intrusive way on the side."}}
{"actor":{"display_login":"jrudolph"}}