"With context" support (for use with FlowWithContext) #780
Conversation
|
The implementation of the Java |
core/src/main/scala/akka/kafka/internal/TransactionalSource.scala
Outdated
Show resolved
Hide resolved
|
@RayRoestenburg Would be great if you'd have a look at this. |
|
@ennru sorry I got my mail filters setup wrong somehow, will have a look. |
| * Batches offsets from context and commits them to Kafka. | ||
| */ | ||
| @ApiMayChange | ||
| def sinkWithOffsetContext[E](settings: CommitterSettings): Sink[(E, Committable), Future[Done]] = |
There was a problem hiding this comment.
We still need to fix the name here. sinkWithCommitableContext? Same for the flow above?
There was a problem hiding this comment.
But it is the same as the flow?! withOffsetContext
There was a problem hiding this comment.
Yes. I think that one should also be renamed to flowWithCommittableContext as that is what the context type is.
There was a problem hiding this comment.
This is a hard one. The API is actually lying, it can't handle "any" Committable as everything but offsets will fail here
There was a problem hiding this comment.
I changed those new withOffsetContext methods to require CommittableOffset other Committables won't do anyway.
| /** | ||
| * Batches offsets and commits them to Kafka, emits `CommittableOffsetBatch` for every committed batch. | ||
| */ | ||
| def batchFlow[C <: Committable](settings: CommitterSettings): Flow[C, CommittableOffsetBatch, NotUsed] = |
There was a problem hiding this comment.
This was lacking in the Java DSL, I'm tempted to require CommittableOffset here as well, but that would differ from the Scala DSL.
There was a problem hiding this comment.
Yes, I think it is good to keep them the same.
docs/src/main/paradox/consumer.md
Outdated
| | Factory method | Stream element type | Emits | | ||
| |-------------------------|--------------------------------|--------------| | ||
| | `sink` | `Committable` | N/A | | ||
| | `sinkWithOffsetContext` | Any (`Committable` in context) | N/A | |
There was a problem hiding this comment.
| | `sinkWithOffsetContext` | Any (`Committable` in context) | N/A | | |
| | `sinkWithOffsetContext` | Any (`CommittableOffset` in context) | N/A | |
docs/src/main/paradox/consumer.md
Outdated
| | `sinkWithOffsetContext` | Any (`Committable` in context) | N/A | | ||
| | `flow` | `Committable` | `Done` | | ||
| | `batchFlow` | `Committable` | `CommittableOffsetBatch` | | ||
| | `flowWithOffsetContext` | Any (`Committable` in context) | `NotUsed` (`CommittableOffsetBatch` in context) | |
There was a problem hiding this comment.
| | `flowWithOffsetContext` | Any (`Committable` in context) | `NotUsed` (`CommittableOffsetBatch` in context) | | |
| | `flowWithOffsetContext` | Any (`CommittableOffset` in context) | `NotUsed` (`CommittableOffsetBatch` in context) | |
Purpose
This adds
Producer.withContextto support Akka's newFlowWithContextout of the box.It works the same as
flexiFlowwith the difference that thePassThroughis unavailable to the user as it is used to carry the context along internally.Adds
Consumer.sourceWithOffsetContextwhich createsSourceWithContext[ConsumerRecord[K, V], CommittableOffset, Control].References
Source.asSourceWithContextFlow.asFlowWithContextChanges
Envelopeto set the pass-through valueProducer
flowWithContextmethodsConsumer
sourceWithOffsetContextTransactional
sourceWithOffsetContextsinkWithOffsetContextflowWithOffsetContextCommitter
flowWithOffsetContextsinkWithOffsetContext