[WIP] mapConcat context mapping strategies

[seglo]

Motivation

Akka Streams SourceWithContext and FlowWithContext allows the user to propagate a context object per element in the stream. This generally includes metadata about the element such as source offset data or correlation IDs. For stages with 1:1 element mapping it’s easy to propagate contexts on behalf of the user to the transformed element. For stages that have 1:0 or 1:many transformations there is no one right way to propagate the context.

A common use case for contexts is to include positional/offset data from the source element’s system (i.e. a queue). In the case of Alpakka Kafka, there are sources that can propagate partition offset information in the context for each element, which can then be used to commit that an element has been fully processed. The default implementation of mapConcat today will propagate the same input context to all output elements, but with this strategy we don’t know if an element represents the last element from a previous 1:many transformation, so we can’t say the source offset has been fully processed until we wait for the next offset. If we wait for the next offset then commits will always be at least one element behind.

This document proposes a design that lets the user choose a strategy in situations where they’re using contexts and flatMap-like stages such as mapConcat and flatMapConcat.

Context Mapping Strategies

Same

Map the input context of the input element to all output elements on behalf of the user. This is the current behaviour of mapConcat and flatMapConcat and can remain the default behaviour when a strategy is not provided.

First

Allow the user to transform the context of the first element only given the input element and input context, and output element. All other elements will receive the input element’s context.

Last

Allow the user to transform the context of the last element only given the input element, input context, output element, and index of output element. All other elements will receive the input element’s context.

The Last strategy can also be used to implement an “Only” strategy, when only one output element exists.

Iterate

Allow the user to transform the context of each output element. For each output element the user can transform the context given the input element, input context, output element, output element index, and whether there is a next (hasNext) element.

The Iterate strategy is a low-level strategy that lets you map to an output context that can satisfy any 1:many use case.

The Iterate strategy can be used to recreate the Same, First, and/or Last strategy. It can also be used to implement an “Only” strategy, when only one output element exists.

Iterate or Empty

The iterate strategy, including a way to transform an “empty” use case when there are no output elements. This allows the user to push an output element and output context that represents a 1:0 mapping use case.

PR

This PR is an experimental branch demonstrating the above context mapping strategies in mapConcat and its variants.

[akka-ci]

Test FAILed.

[seglo]

This is mostly a copy of StatefulMapConcat. Maybe they could be merged, if it makes sense.

[jrudolph]

Couldn't it just be using statefulMapConcat?

[seglo]

It could, but I didn't update statefulMapConcat to use the new custom stage. A new stage is required to execute the UDFs at the right time.

[jrudolph]

Looking good so far, @seglo! Some comments:

• For a strategy to be included, it would be nice to havea basic example or explanation about why it is important and general enough to be used. The contextMapping in the first example seems somewhat useful but hard to say how generally useful it is.
• You can already do whatever you want by temporarily dropping out of the WithContext API. So, whatever API we come up it needs to be better (for whatever metric) than what you can do with just via. E.g. in the example from the test

        val contextMapping =
        ContextMapStrategy.iterate(fn = (_: String, inCtx: OffsetContext, _: String, index: Int, hasNext: Boolean) => {
          if (index == 0 && hasNext) OffsetContext(inCtx.offset, First)
          else if (index == 0 && !hasNext) OffsetContext(inCtx.offset, Only)
          else if (!hasNext) OffsetContext(inCtx.offset, Last)
          else OffsetContext(inCtx.offset, Within)
        })
		// ...
        .mapConcat(
          { str =>
            List(1, 2, 3).map(i => s"$str-$i")
          },
          contextMapping
        )

can for the concrete case also be written as

.via(Flow[(String, OffsetContext)].mapConcat {
          case (in, OffsetContext(ctx, _)) =>
            Vector(
              s"$in-1" -> OffsetContext(ctx, First),
              s"$in-2" -> OffsetContext(ctx, Within),
              s"$in-3" -> OffsetContext(ctx, Last))
        })

or more generally as

        .via(Flow[(String, OffsetContext)].mapConcat {
          case (in, OffsetContext(ctx, _)) =>
            val result = List(1, 2, 3).map(i => s"$in-$i")

            if (result.isEmpty) Vector.empty
            else if (result.size == 1) result.map(_ -> OffsetContext(ctx, Only))
            else {
              (result.head -> OffsetContext(ctx, First)) +:
              result.drop(1).dropRight(1).map(_ -> OffsetContext(ctx, Within)) :+
              (result.last -> OffsetContext(ctx, Last))
            }
        })

I guess the two main benefits of the new approach would be:

1. separation of concerns
2. being able to reuse the context mapping strategies

It's nice that the API seems to cover most cases you could come up with but does it really need that flexibility? (I really don't know, maybe it is needed). One benefit would probably also be that the suggested API could work almost the same API-wise for flatMapConcat for which it would be much harder to change a given non-context usage into a WithContext usage manually (e.g. just doing something on the last element is pretty hard to do as you need to look one element into the future).

[jrudolph]

API-wise, usages might look nicer if all the parameters would be put into a wrapper object.

[jrudolph]

Is the index really needed? Otherwise, it could be isFirst: Boolean and isLast: Boolean if that's what the index would be used for usually.

[seglo]

Thanks for the feedback @jrudolph

For a strategy to be included, it would be nice to havea basic example or explanation about why it is important and general enough to be used. The contextMapping in the first example seems somewhat useful but hard to say how generally useful it is.

I could have elaborated on this further in the Motivation section. I first started thinking about this as a result of a request from Cloudflow for Alpakka Kafka. To enable at-least-once delivery guarantees in situations where users use mapConcat they have to know when all the output elements have been produced so that the consumed offset they were transformed from can be committed back to Kafka. Currently they always commit an offset when they encounter the next offset, which means they are always one offset behind when committing.

If an API exists for context mapping then we could provide a default strategy in Alpakka Kafka (or the user could implement their own) that can be re-used to indicate that the context of the last output element is the last element, then a commit could be performed.

You're right this could be handled without contexts at all at an app level, but this would be a useful way to separate this concern from the rest of the user's application code. There are use cases where this specific use case breaks down, such as when elements arrive out of order, but I think that's an edge case that's the responsibility of the user to deal with.

[ennru]

This looks like a good direction.

As @jrudolph points out, flatMapConcat would be important to try out as it doesn't offer these workarounds. The hasNext knowledge is not as easily available for it.

It might make sense to come up with other predefined strategies. Eg. Option based which map the context to Some for the first or last element and None for all others (that strategy would request for a Scala/Java distinction).

[ennru]

Do [[ links work for type parameters? Ctx is missing one closing bracket.

[seglo]

I'm not sure. I'll check.

[seglo]

As @jrudolph points out, flatMapConcat would be important to try out as it doesn't offer these workarounds. The hasNext knowledge is not as easily available for it.

Yes, that's a good point. I focused on mapConcat first because it was an easier use case, but could still demonstrate the proposed API. I'll implement flatMapConcat.

[akka-ci]

Test FAILed.

[akka-ci]

Test FAILed.

[seglo]

This will take on a new direction. I'll use some aspects of this work in a new proposal.