Refactor to make consumer stages to use base classes #45

IgorFedchenko · 2019-08-24T13:22:29Z

This work is a part of issue #36 .

Alpakka project uses base classes to implement common logic between stages. There are lots of abstractions, and even more would be required to be implemented in c# because scala has mixin feature and allows developers to do kind of inline inheritance (like specifying base class and overrides in same place as a part of inheritance syntax) - and in c# we need to implement separate child class each time and use it instead.

This is a first PR in the series of PRs in issue #36 , and the goal of this one is to:

Use KafkaSourceStage base class for all consumer stages definition
Use SingleSourceStageLogic base class of some of consumer stage logic classes. So far we have PlainSourceStage and CommittableSourceStage, and both of them inherit from this class.

While common logic of message consuming and handling is moved to SingleSourceStageLogic class, the difference is in downstream messages - PlainSourceStage pushes plain consumed records, while CommittableSourceStage pushes CommittableMessage's, which allows committing offsets when desired. To support this difference, there is a IMessageBuilder interface and different implementations, so far only PlainMessageBuilder and CommittableSourceMessageBuilder are implemented.

To simplify the review, I will make review and put some comments on changes right now.

…e logic

src/Akka.Streams.Kafka.Tests/KafkaFixture.cs

src/Akka.Streams.Kafka/Dsl/KafkaConsumer.cs

IgorFedchenko · 2019-08-24T13:33:27Z

src/Akka.Streams.Kafka/Messages/CommittableMessage.cs

        {
-            return _task();
+            return Task.FromResult(_committer.Commit());


Alpakka's Commit method returns Future[Done], and committed partition offsets can be accessed from CommittableMessage directly, so no need to return them from Commit method. Also, in different stages there are different implementations of IInternalCommitter commit method, so more general Task response is better here.

Also, the Func<List<TopicPartitionOffset>> task field is replaced with IInternalCommitter committer field to allow using additional info for commitment as alpakka does (and we will use this flexibility in KafkaInternalCommitter class).

I like this change - easier to support future versions of Kafka client where they might return a real Task upon calling commit. Make sure you add some XML-DOC to these classes though.

src/Akka.Streams.Kafka/Stages/Consumers/Abstract/SingleSourceStageLogic.cs

IgorFedchenko · 2019-08-24T13:40:34Z

src/Akka.Streams.Kafka/Stages/Consumers/Abstract/SingleSourceStageLogic.cs

                    {
-                        Push(stage.Out, message);
+                        Push(shape.Outlet, messageBuilder.CreateMessage(message, _consumer));


Using generic IMessageBuilder to build downstream message.
The tricky part here is passing _consumer instance to builder. Alpakka does not do that. Instead, they use GraphStage's materializer property to get consumer actor reference and use it in message builder like this:

override protected def logic(shape: SourceShape[CommittableMessage[K, V]]): GraphStageLogic with Control = new SingleSourceLogic[K, V, CommittableMessage[K, V]](shape, settings, subscription) with CommittableMessageBuilder[K, V] { override def metadataFromRecord(record: ConsumerRecord[K, V]): String = _metadataFromRecord(record) override def groupId: String = settings.properties(ConsumerConfig.GROUP_ID_CONFIG) lazy val committer: KafkaAsyncConsumerCommitterRef = { val ec = materializer.executionContext new KafkaAsyncConsumerCommitterRef(consumerActor, settings.commitTimeout)(ec) } }

We do not use any consumer actor right now, neither Akka.Streams gives materializer property - or maybe I did not find it - so to allow builder to execute commit action, I am passing consumer instance directly.

@IgorFedchenko the materializer might be passed in as a Scala implicit - which makes the code somewhat hard to read. I'll take a look and see how they're doing this.

Or maybe you might have some other ideas

IgorFedchenko

All right, I left some notes on file changes, now it is ready for review.

Aaronontheweb

Have some minor changes that need to be implemented

src/Akka.Streams.Kafka.Tests/KafkaFixture.cs

Aaronontheweb · 2019-08-26T17:21:43Z

src/Akka.Streams.Kafka/Messages/CommittableMessage.cs

        {
-            return _task();
+            return Task.FromResult(_committer.Commit());


I like this change - easier to support future versions of Kafka client where they might return a real Task upon calling commit. Make sure you add some XML-DOC to these classes though.

src/Akka.Streams.Kafka/Stages/Consumers/Abstract/KafkaSourceStage.cs

Aaronontheweb · 2019-08-26T17:24:32Z

src/Akka.Streams.Kafka/Stages/Consumers/Committers.cs

+        /// <summary>
+        /// Commit all offsets (of different topics) belonging to the same stage
+        /// </summary>
+        List<TopicPartitionOffset> Commit();


Does this list need to be mutable?

I do not think so - it is kafka's driver returning mutable collection, but indeed, there is no point in such mutations. Will update this

Aaronontheweb · 2019-08-26T17:24:57Z

src/Akka.Streams.Kafka/Stages/Consumers/Committers.cs

+
+namespace Akka.Streams.Kafka.Stages.Consumers
+{
+    public interface IInternalCommitter 


Need an XML-DOC that describes this as an internal API.

Also, should this interface be internal or does it need to be exposed to outside of the library?

No, you are right, this is internal API - will fix this

Aaronontheweb · 2019-08-26T17:25:23Z

src/Akka.Streams.Kafka/Stages/Consumers/Concrete/CommittableSourceStage.cs

+
+namespace Akka.Streams.Kafka.Stages.Consumers
+{
+    internal class CommittableSourceStage<K, V> : KafkaSourceStage<K, V, CommittableMessage<K, V>>


Needs some XML-DOC

Aaronontheweb · 2019-08-26T17:25:34Z

src/Akka.Streams.Kafka/Stages/Consumers/Concrete/PlainSourceStage.cs

+
+namespace Akka.Streams.Kafka.Stages.Consumers
+{
+    internal class PlainSourceStage<K, V> : KafkaSourceStage<K, V, ConsumeResult<K, V>>


IgorFedchenko

Added xml comments, and changed public to internal in few places, as per your changes request.

src/Akka.Streams.Kafka.Tests/KafkaFixture.cs

IgorFedchenko · 2019-08-26T17:52:15Z

src/Akka.Streams.Kafka/Stages/Consumers/Committers.cs

+
+namespace Akka.Streams.Kafka.Stages.Consumers
+{
+    public interface IInternalCommitter 


No, you are right, this is internal API - will fix this

IgorFedchenko · 2019-08-26T17:53:46Z

src/Akka.Streams.Kafka/Stages/Consumers/Committers.cs

+        /// <summary>
+        /// Commit all offsets (of different topics) belonging to the same stage
+        /// </summary>
+        List<TopicPartitionOffset> Commit();


I do not think so - it is kafka's driver returning mutable collection, but indeed, there is no point in such mutations. Will update this

IgorFedchenko · 2019-08-26T18:38:49Z

src/Akka.Streams.Kafka/Stages/Consumers/Concrete/CommittableSourceStage.cs

+
+namespace Akka.Streams.Kafka.Stages.Consumers
+{
+    internal class CommittableSourceStage<K, V> : KafkaSourceStage<K, V, CommittableMessage<K, V>>


IgorFedchenko · 2019-08-26T18:39:03Z

src/Akka.Streams.Kafka/Stages/Consumers/Concrete/PlainSourceStage.cs

+
+namespace Akka.Streams.Kafka.Stages.Consumers
+{
+    internal class PlainSourceStage<K, V> : KafkaSourceStage<K, V, ConsumeResult<K, V>>


IgorFedchenko · 2019-08-26T18:42:07Z

src/Akka.Streams.Kafka/Stages/Consumers/Committers.cs

+        /// <summary>
+        /// Commit all offsets (of different topics) belonging to the same stage
+        /// </summary>
+        Task Commit();


@Aaronontheweb Alpakka returnes Task here instead of topic partitions list as I did before - and this makes sense, because the public ICommittable interface returns general Task without typed value.

I think that's fine

IgorFedchenko · 2019-08-26T18:44:25Z

Sorry, I see some conflicts with dev branch - will resolve now

…se_classes

IgorFedchenko · 2019-08-26T18:46:39Z

Ok, now ready for review.

Aaronontheweb

LGTM

IgorFedchenko added 2 commits August 24, 2019 13:39

Use KafkaSourceStage base class for all consumer stages

0dd9664

Use SingleSourceStageLogic base with message builders for source stag…

ba64bad

…e logic

IgorFedchenko commented Aug 24, 2019

View reviewed changes

src/Akka.Streams.Kafka.Tests/KafkaFixture.cs Show resolved Hide resolved

IgorFedchenko commented Aug 24, 2019

View reviewed changes

src/Akka.Streams.Kafka/Dsl/KafkaConsumer.cs Show resolved Hide resolved

IgorFedchenko commented Aug 24, 2019

View reviewed changes

src/Akka.Streams.Kafka/Stages/Consumers/Abstract/SingleSourceStageLogic.cs Show resolved Hide resolved

IgorFedchenko commented Aug 24, 2019

View reviewed changes

IgorFedchenko marked this pull request as ready for review August 24, 2019 13:41

Merge branch 'dev' into refactor_to_use_base_classes

cd4a1ad

Aaronontheweb requested changes Aug 26, 2019

View reviewed changes

IgorFedchenko added 4 commits August 26, 2019 21:14

Added xml docs to committers and made them internal API

9023a99

Added xml comments to public messages + only interfaces should be public

fb055bd

Updated IInternalCommitter to return plain task

c957232

Added comments to consumer stages

61d0444

IgorFedchenko commented Aug 26, 2019

View reviewed changes

Merge remote-tracking branch 'akkadotnet/dev' into refactor_to_use_ba…

172da07

…se_classes

Aaronontheweb approved these changes Aug 27, 2019

View reviewed changes

Aaronontheweb merged commit 45a95ed into akkadotnet:dev Aug 27, 2019

IgorFedchenko deleted the refactor_to_use_base_classes branch September 4, 2019 10:39

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Refactor to make consumer stages to use base classes #45

Refactor to make consumer stages to use base classes #45

IgorFedchenko commented Aug 24, 2019

IgorFedchenko Aug 24, 2019

Aaronontheweb Aug 26, 2019

IgorFedchenko Aug 24, 2019

Aaronontheweb Aug 27, 2019

Aaronontheweb Aug 27, 2019

IgorFedchenko left a comment

Aaronontheweb left a comment

Aaronontheweb Aug 26, 2019

Aaronontheweb Aug 26, 2019

IgorFedchenko Aug 26, 2019

Aaronontheweb Aug 26, 2019

IgorFedchenko Aug 26, 2019

Aaronontheweb Aug 26, 2019

IgorFedchenko Aug 26, 2019

Aaronontheweb Aug 26, 2019

IgorFedchenko Aug 26, 2019

IgorFedchenko left a comment

IgorFedchenko Aug 26, 2019

IgorFedchenko Aug 26, 2019

IgorFedchenko Aug 26, 2019

IgorFedchenko Aug 26, 2019

IgorFedchenko Aug 26, 2019

Aaronontheweb Aug 27, 2019

IgorFedchenko commented Aug 26, 2019

IgorFedchenko commented Aug 26, 2019

Aaronontheweb left a comment

Refactor to make consumer stages to use base classes #45

Refactor to make consumer stages to use base classes #45

Conversation

IgorFedchenko commented Aug 24, 2019

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

IgorFedchenko left a comment

Choose a reason for hiding this comment

Aaronontheweb 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

IgorFedchenko 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

IgorFedchenko commented Aug 26, 2019

IgorFedchenko commented Aug 26, 2019

Aaronontheweb left a comment

Choose a reason for hiding this comment