Kafka SourceProvider implementation and Slick integration example #80

seglo · 2020-04-28T21:09:24Z

Kafka SourceProvider that allows for consumer group membership as well as seeking to externally provided offsets.
MergeableOffset - An offset that translates into multiple rows in the offset store. Projections persist 1 or more rows with saveOffset. Projections read all rows with readOffset and merge them together into a MergeableOffset that can be used to store a Kafka partition to offset map.
Integration test with Slick exactlyOnce API
Kafka projections could be implemented using Kafka transactions. We can explore this at a later time. See Exactly once Kafka projections #94 for details.

akka-projection-core/src/main/scala/akka/projection/internal/OffsetStore.scala

akka-projection-kafka/src/main/scala/akka/kafka/scaladsl/Processor.scala

akka-projection-kafka/src/main/scala/akka/projection/kafka/KafkaSourceProvider.scala

akka-projection-kafka/src/main/scala/akka/projection/kafka/internal/KafkaOffsetStore.scala

akka-projection-kafka/src/main/scala/akka/projection/kafka/internal/KafkaProjectionImpl.scala

seglo · 2020-04-28T23:05:02Z

project/Dependencies.scala

@@ -15,6 +15,7 @@ object Dependencies {
  object Versions {
    val akka = "2.6.4"
    val alpakka = "2.0.0-RC2"
+    val alpakkaKafka = "2.0.2+21-0427b570"


To use the unreleased SendProducer API

akka-projection-kafka/src/main/scala/akka/projection/kafka/MessageEnvelope.scala

akka-projection-kafka/src/main/scala/akka/projection/kafka/KafkaSourceProvider.scala

akka-projection-kafka/src/main/scala/akka/projection/kafka/internal/KafkaOffsetStore.scala

akka-projection-kafka/src/main/scala/akka/projection/kafka/internal/KafkaProjectionImpl.scala

octonato

I said most of it in the meeting, but sending my comments anyway to get it registered.

akka-projection-core/src/main/scala/akka/projection/internal/OffsetStore.scala

akka-projection-kafka/src/main/scala/akka/projection/kafka/KafkaSourceProvider.scala

akka-projection-kafka/src/main/scala/akka/projection/kafka/internal/KafkaProjectionImpl.scala

patriknw

looking promising

akka-projection-kafka/src/main/scala/akka/projection/kafka/KafkaSourceProvider.scala

akka-projection-slick/src/main/scala/akka/projection/slick/internal/SlickOffsetStore.scala

patriknw · 2020-04-30T11:49:03Z

examples/src/test/scala/docs/KafkaToSlickExample.scala

+           }
+         }
+        }
+        """)


config should be for the ActorSystem, and here accessed via context.system.settings.config

Do you mean move this to an application.conf and load it from the ActorSystem? How would we provide configuration for other examples in this project if it's in a file? Should we create a sub project per example?

It can still be in code, but just defined above when the ActorSystem is created.

Ok, got it.

TODO

I think you meant that I should extend ScalaTestWithActorTestKit like the other integration tests, but I couldn't do this since I'm using Alpakka Kafka's testkit. I had to recreate some of the boilerplate, but the config is applied in the same way as SlickSpec. WDYT?

akka-projection-kafka/src/main/scala/akka/projection/kafka/KafkaSourceProvider.scala

octonato

I left a few comments here and there.

I have concerns about the Mergeable offsets and how it is increasing the complexity. I would like to avoid that one use case affects the APIs in a non-obvious way for the users.

If we can keep it as an implementation detail that will be great.

akka-projection-core/src/main/scala/akka/projection/internal/OffsetSerialization.scala

akka-projection-kafka/src/main/scala/akka/projection/kafka/KafkaSourceProvider.scala

akka-projection-slick/src/main/scala/akka/projection/slick/internal/SlickOffsetStore.scala

patriknw

looking good

akka-projection-core/src/main/scala/akka/projection/internal/OffsetSerialization.scala

akka-projection-kafka/src/main/scala/akka/projection/kafka/KafkaSourceProvider.scala

akka-projection-slick/src/main/scala/akka/projection/slick/internal/SlickOffsetStore.scala

patriknw · 2020-05-04T09:32:51Z

akka-projection-slick/src/main/scala/akka/projection/slick/internal/SlickOffsetStore.scala

  }

  class OffsetStoreTable(tag: Tag) extends Table[OffsetRow](tag, "AKKA_PROJECTION_OFFSET_STORE") {

-    def projectionId = column[String]("PROJECTION_ID", O.Length(255, varying = false), O.PrimaryKey)
+    def projectionName = column[String]("PROJECTION_NAME", O.Length(255, varying = false))


That can be useful for backoffice admin ui also. We should add it to the Cassandra table also even though we don't support the MergeableOffset there.

👍

Though I think we should support MergeableOffset in the CassandraOffsetStore too.

examples/src/test/scala/docs/KafkaToSlickExample.scala

patriknw · 2020-05-04T09:40:08Z

akka-projection-slick/src/main/scala/akka/projection/slick/internal/SlickOffsetStore.scala

-      offsetTable.filter(_.projectionId === projectionId.id).result.headOption.map { maybeRow =>
-        maybeRow.map(row => fromStorageRepresentation[Offset](row.offsetStr, row.manifest))
+      offsetTable.filter(_.projectionName === projectionId.name).result.map { maybeRow =>
+        maybeRow.map(row => SingleOffset(projectionId, row.manifest, row.offsetStr, row.mergeable))
      }


I think this is a good first implementation. Later we could optimize for the single-offset case by first reading one row for the projectionId.id, and if that is mergable == false then it's know to be single-offset, otherwise (including when not found) we can read all rows for the projectionId.name.

Yes. I think the worst case would be loading all offsets for a Kafka consumer group with 1000's of partitions, but even then the readOffset use case only happens at startup or a rebalance, so it's going to be a negligible performance impact in the majority of use cases.

seglo · 2020-05-06T17:52:04Z

@patriknw @renatocaval I think this is ready for another [final] review.

octonato

Thanks @seglo. It's looking really good.

I'm concerned with the idea of subscribing to more than one topic from inside one projection.

A part from that, I left here and there a few comments.

There is one issue with the ActorSystem in KafkaSpecBase that we should address.

akka-projection-core/src/main/scala/akka/projection/internal/MergeableOffset.scala

akka-projection-core/src/test/scala/akka/projection/internal/OffsetSerializationSpec.scala

akka-projection-core/src/main/scala/akka/projection/scaladsl/SourceProvider.scala

akka-projection-kafka/src/main/scala/akka/projection/kafka/KafkaSourceProvider.scala

octonato · 2020-05-06T20:35:15Z

akka-projection-kafka/src/main/scala/akka/projection/kafka/KafkaSourceProvider.scala

+  override def source(readOffsets: ReadOffsetsHandler): Future[Source[ConsumerRecord[K, V], _]] = {
+    // get the total number of partitions to configure the `breadth` parameter, or we could just use a really large
+    // number.  i don't think using a large number would present a problem.
+    val numPartitionsF = Future.sequence(topics.map(client.getPartitionsFor)).map(_.map(_.length).sum)


If we keep one topic per projection that issue disappears

No, it just removes one layer of nesting. We'll still query the topic for the number of partitions, the above is just summing the number of partitions of all topics passed in.

Yes, I got it, but I understood that the concern was the Future.sequence. Of course, this only a real concern if the set of topics is very large.

I see. I think it's probably fine for now. I've never seen any client subscriptions for too many topics.

akka-projection-kafka/src/test/scala/akka/projection/kafka/KafkaSpecBase.scala

examples/src/test/scala/akka/projection/examples/KafkaToSlickIntegrationSpec.scala

octonato

Thanks @seglo. It's looking really good.

I'm concerned with the idea of subscribing to more than one topic from inside one projection.

A part from that, I left here and there a few comments.

There is one issue with the ActorSystem in KafkaSpecBase that we should address.

patriknw

looking good, a few comments...

...ction-cassandra/src/main/scala/akka/projection/cassandra/internal/CassandraOffsetStore.scala

patriknw · 2020-05-07T10:44:14Z

akka-projection-slick/src/main/scala/akka/projection/slick/internal/SlickProjectionImpl.scala

    }

+    val futSource = sourceProvider.source(readOffsets)
+


One funny thing is possible for the AtLeastOnce. One projection instance is consuming from topic-partition and saving the offsets with AtLeastOnce. Then that topic-partition is re-assigned to another projection instance. Some elements from the first projection might still be in flight and not saved yet, but when they are saved the offset could be less than than what the new projection instance has saved. Old is overwriting new.

Probably not a big problem since the new projection will soon overwrite again. It would still be at-least-once if 3rd projection instance takes over. We could track this in a separate issue.

Related thought, but for exactlyOnce. The old projection instance is processing offset 17, but before it has been committed to the database there is a hand-over of that topic-partition to another projection instance, reading offset 16, and processing 17. The old instance saves has successfully processed 17 and saves offset 17. The new instance will also process 17. It's not exactly-once.

Any way we can control the hand-over for such scenario?

Related thought, but for exactlyOnce. The old projection instance is processing offset 17, but before it has been committed to the database there is a hand-over of that topic-partition to another projection instance, reading offset 16, and processing 17. The old instance saves has successfully processed 17 and saves offset 17. The new instance will also process 17. It's not exactly-once.

Any way we can control the hand-over for such scenario?

That's a good point. If we were using Alpakka Kafka transactions then we could rely on logic within Kafka transactions and extra bookkeeping done in Alpakka Kafka to ensure inflight messages from a revoked partition aren't committed, or all committed before the partition is revoked.

The Source we're using right now doesn't have any of these guarantees and is really only suitable to guarantee at-least-once. We could implement an Alpakka Kafka PartitionAssignmentHandler that lets us perform blocking operations during a rebalance. Such as waiting for the partitioned Source for a revoked partition complete, before letting the rebalance process continue (or cancelling it pre-emptively). This will require some not-so-fun concurrency locking code to implement though. But since we control the streams implementation it might be an option in projections.

These weird partition rebalancing problems come up often with Alpakka Kafka and result in a lot of the project's complexity. I'll create an issue to consider solutions for this case.

akka-projection-kafka/src/main/scala/akka/projection/kafka/KafkaSourceProvider.scala

akka-projection-core/src/main/scala/akka/projection/scaladsl/SourceProvider.scala

akka-projection-kafka/src/test/resources/logback-test.xml

akka-projection-kafka/src/test/resources/application.conf

akka-projection-kafka/src/test/scala/akka/projection/kafka/KafkaSpecBase.scala

patriknw · 2020-05-07T11:18:48Z

examples/src/test/scala/akka/projection/examples/KafkaToSlickIntegrationSpec.scala

+ * Copyright (C) 2020 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.projection.examples


Good if this is placed in a non-akka package, docs.projection.exampes or such. Then we verify that we don't use any internal APIs here.

Ah, my example uses SlickOffsetStore to assert offset state. I was thinking this example could serve also as an integration test, but maybe we need both. Should we create an integration project that can remain in the akka namespace and then clean up the example to not reference internal APIs?

Keep it as is. We'll see when we write the doc examples.

May I suggest the Alpakka standard of docs.scaladsl and docs.javadsl.

patriknw · 2020-05-07T11:26:11Z

examples/src/test/scala/akka/projection/examples/KafkaToSlickIntegrationSpec.scala

+          case Some(userEventCount) => userEventCount.copy(count = userEventCount.count + 1)
+          case _                    => UserEventCount(eventType, 1)
+        }
+        _ <- userEventCountTable.insertOrUpdate(count)


so the findByEventType and insertOrUpdate are atomic in same transaction?
Is the count read by findByEventType guaranteed to not be updated by another EventTypeCountRepository instance inbetween the read and the insertOrUpdate?
Is all of it translated to sql running in the DB, or is it some DB lock, or how does that work?

Good point. I don't know how that works. I doubt read and writes would be automatically handled in one transaction. I was following the same pattern used in SlickProjectionImplSpec, but this breaks down when more than 1 projection is updating some projection source with a different partition/shard/slice. /cc @renatocaval

I'll need to rethink this example.

What I would use for CRUD, like this, is optimistic locking. A version number column that is read in findByEventType and then used as condition in the update to be the same as what was read. The update also increments it by 1. If no row was updated it means that someone else updated inbetween. Then we could fail, and even use the retry mechanism from #95

Would be pretty nice to include that pattern in this example.

An alternative would be to have one counter per projectionId, so that it's a single writer.

added issue #107 for this

Reads and writes will be run in same transaction, but the issue here is not at Slick level, but the configured transaction isolation.

To be more precise with what a mean, this is a general transaction issue and users need to configure the DB they way they want/need it and code accordingly to it.

To have full protection here we will need either some optimistic locking or use a strict transaction isolation level (eg: Serializable). I don't think it's worth doing that in our tests unless we want to use it as a documentation sample.

For the purpose of this example I was thinking of just querying to see if a count for an event type exists (instead of retrieving the count value itself) and then do either of the following:

If it exists, issue an update statement and increment the count in place, instead of first reading it and incrementing it in code. i.e. UPDATE EVENTS_TYPE_COUNT SET COUNT = COUNT + 1 WHERE EVENT_TYPE = ?

If it does not exist, insert the the record. If the insert fails because another projection created it already then an update will happen on the retry. An alternative would just be to seed the records with the projection table DDL, but as @patriknw mentioned it would be a nice way to showcase retry logic.

akka-projection-core/src/main/scala/akka/projection/javadsl/SourceProvider.scala

...projection-kafka/src/main/scala/akka/projection/kafka/internal/KafkaSourceProviderImpl.scala

ennru · 2020-05-08T16:38:52Z

examples/src/test/scala/akka/projection/examples/KafkaToSlickIntegrationSpec.scala

+ * Copyright (C) 2020 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.projection.examples


May I suggest the Alpakka standard of docs.scaladsl and docs.javadsl.

ennru · 2020-05-08T16:44:40Z

project/Common.scala

@@ -62,6 +63,8 @@ object Common extends AutoPlugin {
        "-skip-packages",
        "akka.pattern" // for some reason Scaladoc creates this
      ),
+    // FIXME enable "-Xfatal-warnings", but then we need the silencer


In Scala 2.13.2 @nowarn is built-in.

created issue #108

octonato

LGTM, ready for merge.

(did I last push to remove .DS_Store files that were committed by accident)

patriknw · 2020-05-11T09:34:46Z

Merged to avoid further merge conflicts. Some follow up in separate PRs.

seglo force-pushed the seglo/kafka-impl branch from aafd6fb to 62d2f7b Compare April 28, 2020 21:28

WIP

c940210

seglo force-pushed the seglo/kafka-impl branch from 62d2f7b to c940210 Compare April 28, 2020 22:51

seglo commented Apr 28, 2020

View reviewed changes

akka-projection-core/src/main/scala/akka/projection/internal/OffsetStore.scala Outdated Show resolved Hide resolved

seglo commented Apr 28, 2020

View reviewed changes

akka-projection-kafka/src/main/scala/akka/kafka/scaladsl/Processor.scala Outdated Show resolved Hide resolved

seglo commented Apr 28, 2020

View reviewed changes

akka-projection-kafka/src/main/scala/akka/projection/kafka/KafkaSourceProvider.scala Outdated Show resolved Hide resolved

seglo commented Apr 28, 2020

View reviewed changes

akka-projection-kafka/src/main/scala/akka/projection/kafka/internal/KafkaOffsetStore.scala Outdated Show resolved Hide resolved

seglo commented Apr 28, 2020

View reviewed changes

akka-projection-kafka/src/main/scala/akka/projection/kafka/internal/KafkaProjectionImpl.scala Outdated Show resolved Hide resolved

seglo commented Apr 28, 2020

View reviewed changes

patriknw reviewed Apr 29, 2020

View reviewed changes

seglo and others added 2 commits April 29, 2020 11:25

WIP akka-projection-kafka

00aa567

example of Kafka offsets stored in Cassandra

616ebd1

patriknw mentioned this pull request Apr 29, 2020

example of Kafka offsets stored in Cassandra #81

Closed

octonato reviewed Apr 29, 2020

View reviewed changes

seglo added 3 commits April 29, 2020 18:04

MergeableOffsets

cafdab1

Merge branch 'seglo/kafka-impl-2' into seglo/kafka-impl

d777606

Update example

636f28c

seglo force-pushed the seglo/kafka-impl branch from 269ecdd to 3cd6acf Compare April 30, 2020 00:47

cleanup

16ba475

seglo force-pushed the seglo/kafka-impl branch from 3cd6acf to 16ba475 Compare April 30, 2020 00:48

patriknw reviewed Apr 30, 2020

View reviewed changes

octonato reviewed Apr 30, 2020

View reviewed changes

seglo added 4 commits April 30, 2020 16:40

fix test regressions

4011dab

MergeableOffset serialization tests

e78c5e1

cleanup

a5cf833

Recursive offset type for offsets that make up MergeableOffset

a2c2442

patriknw reviewed May 4, 2020

View reviewed changes

seglo force-pushed the seglo/kafka-impl branch from db04fb9 to c48ebab Compare May 4, 2020 19:46

Integration spec

7a937ad

seglo force-pushed the seglo/kafka-impl branch from c48ebab to 7a937ad Compare May 4, 2020 19:52

KafkaSourceProviderSpec

2ef8ed8

seglo changed the title ~~WIP- Kafka projection implementation~~ Kafka SourceProvider implementation and Slick integration example May 6, 2020

seglo marked this pull request as ready for review May 6, 2020 16:28

seglo added 2 commits May 6, 2020 12:44

Merge branch 'master' into seglo/kafka-impl

6510b9e

awaitProduce for longer timeout

3339aad

seglo requested review from patriknw and octonato May 6, 2020 17:51

octonato reviewed May 6, 2020

View reviewed changes

patriknw reviewed May 7, 2020

View reviewed changes

seglo mentioned this pull request May 7, 2020

Support MergeableOffset in CassandraOffsetStore to support KafkaSourceProvider #97

Open

seglo added 2 commits May 7, 2020 15:30

Remove nesting of MergeableOffsets.Offset

69b956c

Update javadsl.SourceProvider and impls

48dc024

seglo mentioned this pull request May 7, 2020

Ensure that in-flight messages from KafkaSourceProvider are not processed for revoked partitions #98

Closed

review feedback

056832c

ennru reviewed May 8, 2020

View reviewed changes

seglo and others added 6 commits May 8, 2020 14:21

Merge branch 'master' into seglo/kafka-impl

7c53a55

Merge branch 'master' into seglo/kafka-impl

d7f03ee

One actorsystem per kafkaspec

088682e

No support for MergeableOffset in CassandraOffsetStore

8d3d68e

envelope->record

136012e

headers

2691b83

patriknw mentioned this pull request May 11, 2020

failed: KafkaToSlickIntegrationSpec #109

Closed

pending test

13c2940

octonato approved these changes May 11, 2020

View reviewed changes

Merge branch 'master' into seglo/kafka-impl

0b10150

patriknw force-pushed the seglo/kafka-impl branch from b54886c to 0b10150 Compare May 11, 2020 08:52

patriknw merged commit 37bc017 into akka:master May 11, 2020

Kafka SourceProvider implementation and Slick integration example #80

Kafka SourceProvider implementation and Slick integration example #80

Conversation

seglo commented Apr 28, 2020 • edited Loading

Choose a reason for hiding this comment

octonato left a comment

Choose a reason for hiding this comment

patriknw 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

octonato left a comment

Choose a reason for hiding this comment

patriknw 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

seglo commented May 6, 2020

octonato 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

octonato left a comment

Choose a reason for hiding this comment

patriknw 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

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

octonato left a comment

Choose a reason for hiding this comment

patriknw commented May 11, 2020

seglo commented Apr 28, 2020 •

edited

Loading