Accumulate and commit offsets at poll time #851

seglo · 2019-07-08T20:47:02Z

Purpose

This PR proposes 2 major changes to the offset committing.

Decouple back-pressure when committing offsets (background in Allow user to commit offsets without handling response when using committable sources #845)
Send accumulated commit requests asynchronously during the akka.kafka.consumer.poll-interval (inspired by comments in Commit only once per poll-interval #849 and Smarter committer flow #850)

Changes

Replace ask pattern with a fire and forget commit request to the KafkaConsumerActor
Accumulate commit requests in KafkaConsumerActor
At poll time (akka.kafka.consumer.poll-interval) merge all commit requests and perform an asynchronous commit before fetching records.
Throw a akka.kafka.CommitTimeoutException from the commit callback when the round trip takes greater than akka.kafka.consumer.commit-timeout. This exception was previously thrown by the ask timeout handler in the KafkaAsyncConsumerCommitterRef.
Commit any outstanding commit requests during graceful shutdown
Deprecate Committable.commitScaladsl and Committable.commitJavadsl in favour of Committable.commit because we no longer need to return a Future or CompletableFuture.
Remove parallelism from CommitterSettings

More Proposed Changes

Should we deprecate CommitterSettings altogether? We can change all Committer flows to groupedWithin using the akka.kafka.consumer.poll-interval. I can't think of a good reason to commit more frequently than this interval since commits will not be sent immediately to Kafka.
Several tests are no longer relevant if we adopt this new approach to committing. I've commented them out for now.

seglo · 2019-07-08T20:49:23Z

I created this PR as a draft since it's incomplete. I need some feedback about the direction it's going before proceeding.

ennru · 2019-08-12T15:06:16Z

I started digging into your PR now. Would you please rebase it on "master" to get the RebalanceListener changes in.

ennru

Started digging into this. Some comments.

ennru · 2019-08-12T15:14:56Z

core/src/main/scala/akka/kafka/internal/KafkaConsumerActor.scala

+   */
+  private def maybeCommit(): Unit =
+    if (!rebalanceInProgress && commitStash.nonEmpty) {
+      val combinedStash = commitStash.flatMap(_.offsets).toMap


Is there any reason why commitStash isn't kept as Map[TopicPartition, OffsetAndMetadata]?

Not really. The commitStash is just maintaining a list of all the Commit messages that it receives and processes them into the map at commit time. Is there a reason to maintain the stash as a Map between commits?

It is a little more compact in memory to use one map, instead of keeping all maps in the Commit messages and we need to flatten those anyway.

Ok, I'll do that.

ennru · 2019-08-12T15:17:12Z

core/src/main/scala/akka/kafka/internal/KafkaConsumerActor.scala

@@ -338,6 +327,7 @@ import scala.util.control.NonFatal
      commitRefreshing.committed(offsets)

    case Stop =>
+      maybeCommit()


This is great. But it would possibly need to be a commitSync?

That's a good question. If commits are in progress then the KafkaConsumerActor will start using the stopping behaviour, but that behavior will only do up to 1 more poll (and possibly handle some or all remaining commit callbacks) and then stop the actor. I think I assumed the polling would continue until all commit callbacks have returned, but since that's not the case I agree that doing a commitSync would make more sense.

Do you know what the purpose of allowing 1 final poll when in stopping?

It is just that: collect replies to any outstanding commits.
And that solution is actually better now that I play around with it, as it doesn't block.

Yes, I misread part of the condition that checks if there are still comits in progress. So it will poll until all callbacks have returned.

if (stopInProgress && commitsInProgress == 0) {

So to be clear, you're fine with this implementaiton as-is?

ennru · 2019-08-12T15:23:28Z

core/src/main/scala/akka/kafka/scaladsl/Consumer.scala

@@ -222,10 +222,13 @@ object Consumer {
   * Convenience for "at-most once delivery" semantics. The offset of each message is committed to Kafka
   * before being emitted downstream.
   */
+  // TODO: this should probably be deprecated since it can no longer be guaranteed
+  //   we should guide the user to using auto commit interval setting of the Consumer to do this instead
  def atMostOnceSource[K, V](settings: ConsumerSettings[K, V],


This could be replaced with a plain source with auto-committing enabled on Kafka-level via enable.auto.commit=true.

Do you want to do that in this PR?

ennru · 2019-08-12T15:33:25Z

core/src/main/scala/akka/kafka/internal/KafkaConsumerActor.scala

+          if (exception == null) {
+            self ! Committed(offsets.asScala.toMap)
+          } else {
+            log.error("Kafka commit failed", exception)


What failures do we expect here? I guess the stream should fail, or retry committing?

The current Alpakka Kafka behaviour is to return a failure as the reply to the committer.

if (exception != null) sendReply(Status.Failure(exception))

IIUC it is up to the user to handle the Failure of the Future that was returned by the commit. In Committer.batchFlow it would just return the Failure and kill the stream. We can change this to fail the stream to remain consistent.

WRT to retrying commits:

In cases where we can retry sending the commit we'll need to be careful we don't end up committing out of order. I recall discussing this with someone (maybe the Alpakka team?) It's discussed in Kafka: The Definitive Guide and summarized in this SO post:

https://stackoverflow.com/questions/53240589/kafka-commitasync-retries-with-commit-order

ennru · 2019-08-12T15:37:12Z

core/src/main/scala/akka/kafka/internal/MessageBuilder.scala

-    if (offsets.isEmpty)
-      Future.successful(Done)
-    else {
+  override def commitScaladsl(): Future[Done] = Future.successful {


This should instead return a singleton instance of Future[Done].

I'm not sure what you mean. Something like this?

override def commitScaladsl(): Future[Done] = { commit() Future.successful(Done) }

Or do you mean assign Future.successful(Done) to a static member and just return that all the time?

ennru · 2019-08-12T15:38:56Z

core/src/main/scala/akka/kafka/scaladsl/Committer.scala

+        b
+      }
+
+  private def batch(settings: CommitterSettings): Flow[Committable, CommittableOffsetBatch, NotUsed] =


Any reason for this separate method?

It was a remnant from an earlier work I did. It's not necessary any more I'll remove it.

seglo · 2019-08-13T15:46:51Z

@ennru Thanks for reviewing. I rebased master and added a few small changes based on your feedback.

ennru

I'm playing around with just implementing #849 and I believe we gain many of the benefits which this solution has. That smaller scope would allow us to release it as a patch or minor version.

ennru · 2019-08-14T06:46:18Z

core/src/main/scala/akka/kafka/internal/KafkaConsumerActor.scala

@@ -338,6 +327,7 @@ import scala.util.control.NonFatal
      commitRefreshing.committed(offsets)

    case Stop =>
+      maybeCommit()


It is just that: collect replies to any outstanding commits.
And that solution is actually better now that I play around with it, as it doesn't block.

ennru · 2019-08-14T06:48:48Z

core/src/main/scala/akka/kafka/internal/KafkaConsumerActor.scala

+   */
+  private def maybeCommit(): Unit =
+    if (!rebalanceInProgress && commitStash.nonEmpty) {
+      val combinedStash = commitStash.flatMap(_.offsets).toMap


It is a little more compact in memory to use one map, instead of keeping all maps in the Commit messages and we need to flatten those anyway.

ennru · 2019-08-30T13:35:58Z

With #862 merged and #874 implementing this a bit less aggressive, we can close this. Thank you for suggesting and exploring it.

2m mentioned this pull request Aug 6, 2019

Alpakka sprint plan 2019-07-05 akka/akka-meta#108

Closed

13 tasks

ennru mentioned this pull request Aug 6, 2019

Alpakka team sprint plan 2019-08-05 akka/akka-meta#111

Closed

11 tasks

ennru reviewed Aug 12, 2019

View reviewed changes

seglo added 2 commits August 13, 2019 10:50

Accumulate and commit offsets at poll time

4c2deb6

PR feedback

422f3d2

seglo force-pushed the seglo/poll-committer branch from 999901f to 422f3d2 Compare August 13, 2019 15:44

ennru reviewed Aug 14, 2019

View reviewed changes

ennru mentioned this pull request Aug 14, 2019

Aggregate offsets and commit before poll #862

Merged

ennru mentioned this pull request Aug 26, 2019

Smarter committer flow #850

Closed

ennru closed this Aug 30, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Accumulate and commit offsets at poll time #851

Accumulate and commit offsets at poll time #851

seglo commented Jul 8, 2019

seglo commented Jul 8, 2019

ennru commented Aug 12, 2019

ennru left a comment

ennru Aug 12, 2019

seglo Aug 13, 2019

ennru Aug 14, 2019

seglo Aug 14, 2019

ennru Aug 12, 2019

seglo Aug 13, 2019 •

edited

Loading

ennru Aug 14, 2019

seglo Aug 14, 2019

ennru Aug 12, 2019

seglo Aug 13, 2019

seglo Aug 13, 2019

ennru Aug 12, 2019

seglo Aug 13, 2019

ennru Aug 12, 2019

seglo Aug 13, 2019

ennru Aug 12, 2019

seglo Aug 13, 2019

seglo commented Aug 13, 2019

ennru left a comment

ennru Aug 14, 2019

ennru Aug 14, 2019

ennru commented Aug 30, 2019

Accumulate and commit offsets at poll time #851

Accumulate and commit offsets at poll time #851

Conversation

seglo commented Jul 8, 2019

Purpose

Changes

More Proposed Changes

seglo commented Jul 8, 2019

ennru commented Aug 12, 2019

ennru left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

seglo Aug 13, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

seglo commented Aug 13, 2019

ennru left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ennru commented Aug 30, 2019

seglo Aug 13, 2019 •

edited

Loading