Committing producer sink: producer and committer in a sink stage #963

ennru · 2019-11-05T13:34:43Z

Purpose

Introduce a sink stage which produces to Kafka, batches commit offsets and commits.

References

References #932

Changes

New CommittingProducerSinkStage
Unit tests running without a Kafka instance
Change some places to inject the producer via settings instead

Background Context

Kafka to Kafka flows with a committable source, a producer and a committer are a quite common setup which became even simpler with the Producer.committableSinks added in #932. This new stage replaces that implementation with this all-in-one stage.

ennru · 2019-11-18T13:18:06Z

The same test fails on Scala 2.13 for JDK 8 and 11 on Travis as the messages appear in a different order. Locally it works nicely. I need to dig a bit.

seglo

The CommittingProducerSinkStageLogic is very clean, I like it. The only concern I have is the duplication of code between it and DefaultProducerStageLogic. Also, I think we try running the benchmarks on this to see how much gain it gives us over the current impl.

seglo · 2019-11-21T16:04:51Z

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

+private final class CommittingProducerSinkStageLogic[K, V, IN <: Envelope[K, V, Committable]](
+    stage: CommittingProducerSinkStage[K, V, IN],
+    inheritedAttributes: Attributes
+) extends TimerGraphStageLogic(stage.shape)


Ideally, we would consolidate CommittingProducerSinkStageLogic with DefaultProducerStageLogic to DRY up all the async producer and producing logic. Is that the eventual goal?

I isolated the async producer creation. For producing I don't see a good case.

tests/src/test/scala/akka/kafka/internal/CommittingProducerSinkSpec.scala

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

seglo · 2019-11-21T16:17:26Z

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

+  private val collectOffsetCb: AsyncCallback[Committable] = getAsyncCallback[Committable] { offset =>
+    collectOffset(1, offset)
+  }
+
+  private val collectOffsetMultiCb: AsyncCallback[(Int, Committable)] = getAsyncCallback[(Int, Committable)] {
+    case (count, offset) =>
+      collectOffset(count, offset)
+  }


It seems these could be combined into a single async callback with the functionality of collectOffsetMultiCb, but with the name collectOffsetCb

They could, I want to avoid the tuple creation for the most used case of just a single offset.

seglo · 2019-11-21T16:21:16Z

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

+    else if (isClosed(stage.in) && awaitingProduceResult == 0L) commit("upstream closed")
+  }
+
+  private def commit(triggeredBy: String): Unit = {


I like the triggeredBy feature for traceability of a commit. Should we use types instead? It would be less error-prone to use with testing, though it doesn't look like there are tests that assert the triggeredBy of a commit.

seglo · 2019-11-21T16:34:50Z

tests/src/test/scala/akka/kafka/internal/CommittingProducerSinkSpec.scala

+    control.drainAndShutdown().futureValue shouldBe Done
+  }
+
+  it should "produce, and commit when batch size is reached" in assertAllStagesStopped {


How can we tell that the commit was triggered by the interval or reaching the batch?

The interval is 10 s so we can be sure the batch size of 2 is hit before that.

seglo · 2019-11-21T16:35:54Z

tests/src/test/scala/akka/kafka/internal/CommittingProducerSinkSpec.scala

+
+    val elements = immutable.Seq(
+      consumer.message(partition, "value 1"),
+      consumer.message(partition, "value 2")


Consider sending more messages than the batch size to assert batching works as expected.

seglo · 2019-11-21T16:39:43Z

tests/src/test/scala/akka/kafka/internal/CommittingProducerSinkSpec.scala

+    val producer = new MockProducer[String, String](true, new StringSerializer, new StringSerializer)
+    val producerSettings = ProducerSettings(system, new StringSerializer, new StringSerializer)
+      .withProducer(producer)
+    val commitInterval = 5.seconds


This test asserts that the completion performed the commit because the commit interval will never happen in time? I suggest adding a comment here and the other tests that rely on this trick if that's the case (it's not obvious to me).

Added comments.

seglo · 2019-11-21T18:49:05Z

tests/src/test/scala/akka/kafka/internal/CommittingProducerSinkSpec.scala

+    commitMsg.offsetAndMetadata.offset() shouldBe (consumer.startOffset + 1)
+    consumer.actor.reply(Done)
+
+    // TODO how should not getting a produce callback be handled?


I'm not sure. I didn't see an obvious way in MockProducer. This test seems to only assert that the 2nd commit never is acknowledged (instead a CommitTimeoutException occurs), correct?

Yes. I figured asserting produce callback would require a timer for each message, the missing commit callback would fail it eventually anyway.

tests/src/test/scala/akka/kafka/internal/CommittingProducerSinkSpec.scala

ennru · 2019-11-22T10:00:58Z

I do not expect a real difference in the benchmarks. The Kafka overhead will hide most of the benefits, I'm afraid.
This implementation is much lower in allocation and synchronisation costs, though.

ennru · 2019-11-22T12:57:11Z

Various unrelated test cases failed after my latest update:

…tage

ennru · 2019-11-27T13:06:53Z

The benchmarks show a quite relevant improvement.

seglo

LGTM. Is there any reason to add/update docs content for this?

ennru · 2019-11-27T18:29:24Z

No, this is an internal concern and the committableSink factories are new in 2.0.0.

probot-autolabeler bot added core testkit tests labels Nov 5, 2019

ennru force-pushed the committing-producer-sink branch from 50aa796 to 05c7509 Compare November 12, 2019 14:55

ennru self-assigned this Nov 18, 2019

ennru mentioned this pull request Nov 18, 2019

Alpakka team plan 2019-11-18 akka/akka-meta#124

Closed

10 tasks

ennru force-pushed the committing-producer-sink branch from b5d348f to 76c8a96 Compare November 19, 2019 10:16

seglo self-requested a review November 21, 2019 15:53

seglo reviewed Nov 21, 2019

View reviewed changes

ennru force-pushed the committing-producer-sink branch from f050c56 to 81b3f90 Compare November 22, 2019 09:59

ennru added 8 commits November 26, 2019 10:53

Committable producer sink: combine producer and committer in a sink s…

1d55084

…tage

More tests and hardening

2950c64

Remove import

c9a45b1

Signal rebalance events spec: allow different order

853270d

Compare only the assigned partitions

d28f45d

ProducerSpec: less timing dependent

5b56c23

Small changes to see if it helps fixing akka#895

b11914d

Review feedback; DRY deferred producer creation

1777758

ennru force-pushed the committing-producer-sink branch from 81b3f90 to 1777758 Compare November 26, 2019 10:05

ennru added 2 commits November 26, 2019 12:44

Adapt rebalance listener expectation to Kafka 2.4.0

a19b25d

Benchmarks to compare the special sink with the composed solution

26f153a

probot-autolabeler bot added the benchmarks label Nov 26, 2019

seglo self-requested a review November 27, 2019 14:55

seglo approved these changes Nov 27, 2019

View reviewed changes

ennru added this to the 2.0.0 milestone Nov 27, 2019

ennru merged commit 7967fb0 into akka:master Nov 27, 2019

ennru deleted the committing-producer-sink branch November 27, 2019 18:30

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Committing producer sink: producer and committer in a sink stage #963

Committing producer sink: producer and committer in a sink stage #963

ennru commented Nov 5, 2019

ennru commented Nov 18, 2019

seglo left a comment

seglo Nov 21, 2019

ennru Nov 22, 2019

seglo Nov 21, 2019

ennru Nov 22, 2019

seglo Nov 21, 2019

seglo Nov 21, 2019

ennru Nov 22, 2019

seglo Nov 21, 2019

seglo Nov 21, 2019

ennru Nov 22, 2019

seglo Nov 21, 2019

ennru Nov 22, 2019

ennru commented Nov 22, 2019

ennru commented Nov 22, 2019

ennru commented Nov 27, 2019

seglo left a comment

ennru commented Nov 27, 2019

Committing producer sink: producer and committer in a sink stage #963

Committing producer sink: producer and committer in a sink stage #963

Conversation

ennru commented Nov 5, 2019

Purpose

References

Changes

Background Context

ennru commented Nov 18, 2019

seglo 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

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

ennru commented Nov 22, 2019

ennru commented Nov 22, 2019

ennru commented Nov 27, 2019

seglo left a comment

Choose a reason for hiding this comment

ennru commented Nov 27, 2019