Deprecate committableSink

[ennru]

committableSink uses per element committing which performs much worse than Committer.sink. The recommendation should be to use flexiFlow with Committer.sink instead.

[patricknoir]

Hi ennru, I also prefer the Committer.sink but right now I'm facing an issue as this is not supporting implementation of CommittableOffset. Digging through the code there is a pattern matching cast in the MessageBulder class (line 221 of v1.0.5) which expect CommittableOffset to be always the internal implementation. Any plan to fix this?

[ennru]

What other implementation of CommittableOffset are you expecting it to handle? Committable and CommittableOffset are not supposed to be extended outside of Alpakka Kafka.

[patricknoir]

Hi Ennru, apologies I have just seen the annotation regarding NoInheritance. To describe my use case, I have created an Envelope[P] message which extended your CommittableOffset to be able to detect when commitScaladsl() call is performed to do some other operations (i.e. : complete a span etc...). Find below code snippet (Open to suggestions)
`

trait Envelope[+P] extends Committable { self =>
val headers: Map[String, String]
val traceId:String
val payload: Either[Throwable, P]
val span: Span
...
override def commitScaladsl(): Future[Done] = {
val future = initCommittable.commitScaladsl()
future.onComplete {
case Failure(exception) =>
Kamon.runWithSpan(this.span) {
span.fail("producer.error", exception)
span.finish()
}
case Success(_) =>
Kamon.runWithSpan(span) {
span.finish()
}
}(ExecutionContext.global) //Use the global execution context as this is not a critical IO operation.
future
}
override def commitJavadsl(): CompletionStage[Done] = initCommittable.commitJavadsl()
override def batchSize: Long = initCommittable.batchSize

`

[ennru]

You may try to use the pass-through instead of extending Committable. As you found out, the implementation requires certain implementations to be used.

[ennru]

Fixed with #932