Pass ProducerSettings to stage #952

seglo · 2019-10-24T20:19:53Z

Purpose

Pass the ProducerSettings object directly into producer stages so that the stage has full access to the producer factory helper methods. It also saves on the number of parameters that need to be passed to these stages.

I also took the opportunity to add more producer factory options to ProducerSettings so that it's not necessary to maintain overloads to pass external or shared KafkaProducer instances to Producer flows and sinks. I deprecated methods that take these overloads.

References

Preliminary work for asynchronous producer creation in Add partitioned transactional source phase 1 #930

Changes

Pass ProducerSettings to stage.
Move producer factories to settings
Deprecate flow/sink overloads that take producers

ennru

This makes a lot of sense together with the changes for Akka Discovery.
Especially that we can reduce the number of Producer factory methods.

This requires an update of

Sharing producer instance

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

core/src/main/resources/reference.conf

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

ennru · 2019-10-26T15:43:22Z

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

@@ -86,6 +86,8 @@ object Transactional {
      transactionalId: String
  ): Flow[Envelope[K, V, ConsumerMessage.PartitionOffset], Results[K, V, ConsumerMessage.PartitionOffset], NotUsed] = {
    require(transactionalId != null && transactionalId.length > 0, "You must define a Transactional id.")
+    require(settings.producerFactory.isDefined, "You cannot use a shared or external producer factory.")
+    require(settings.producerFactoryAsync.isDefined, "You cannot use a shared or external async producer factory.")


So this won't work with Akka Discovery? Isn't it OK as long as it creates a new instance? Guarding against withProducer is the important bit, right?

This should work with Akka Discovery. The internal producerFactorySync and producerFactoryAsync values are optional and only set when the user uses a withProducer method. They're not used internally to create a producer by default. We'll know for sure once the tests aren't blocked by other issues.

Ah, you're right. I mixed up the enrichAsync and producerFactoryAsync.
A feature we tried to unlock with the producer factory was the possibility to wrap the producer with the OpenTracing decorator.

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

seglo · 2019-10-28T18:59:36Z

Ready for a final review.

ennru · 2019-10-28T20:58:56Z

core/src/main/scala/akka/kafka/ProducerSettings.scala

@@ -349,7 +399,11 @@ class ProducerSettings[K, V] @InternalApi private[kafka] (
   * (without blocking for `enriched`).
   */
  def createKafkaProducerAsync()(implicit executionContext: ExecutionContext): Future[Producer[K, V]] =
-    enriched.map(producerFactory)
+    (producerFactoryAsync, producerFactorySync) match {
+      case (Some(asyncFactory), _) => asyncFactory(executionContext)


What is the use case for the asyncFactory?
Why doesn't it get the enriched settings passed?

I copied this from the Transactional.flow implementation that's been been removed in this PR.
The "async" producer factory was passed to TransactionalProducerStage: https://github.com/akka/alpakka-kafka/pull/952/files#diff-832423d99d7759061130b358267d97a3L101

I don't know if we need this. We no longer require it internally.

It seems irregular to me, remove it.

I [re-]discovered the documentation example used to point to this overload that takes a Future[KafkaProducer[K,V]] that was added with the discovery PR (https://github.com/akka/alpakka-kafka/pull/952/files#diff-832205ad71b0f8522d293275289adc35L60). It seems like the only use for this is if you were to create a KafkaProducer asynchronously externally from Alpakka Kafka, but I'm not sure why we need to support that.

I'll revert the documentation to showing how to share an external KafkaProducer synchronously.

I guess it's because ProducerSettings exposes createKafkaProducerAsync and createKafkaProducerCompletionStage. I'm using this internally for the transactions PR. Maybe it would be useful to share to the user if they want to get an asynchronous producer for some other purpose?

I'll hold off removing this for now.

I squashed the commits in this branch to make rebasing easier for the transactions PR.

Ah, thank you for digging. Yes, I added those to allow passing the enriched producer to the factory methods with a shared producer. With this solution for passing the producer factory, we don't need those.

ennru

We should remove the async factories, either pre-merge or after.
LGTM.

* Move producer factories to settings. * Deprecate flow/sink overloads that take producers * Block use of producer factories with transactional flows and sinks * Deprecate ProducerSettings.producerFactory * Invert withProducer checks on Transactional.flow * Make ProducerSpec compatible with async producer assignment

probot-autolabeler bot added core testkit tests labels Oct 24, 2019

ennru reviewed Oct 25, 2019

View reviewed changes

core/src/main/scala/akka/kafka/internal/DefaultProducerStage.scala Outdated Show resolved Hide resolved

seglo added this to the 2.0.0 milestone Oct 25, 2019

seglo force-pushed the seglo/pass-settings-to-producer branch from 7e72595 to 98203c2 Compare October 25, 2019 20:32

probot-autolabeler bot added the documentation label Oct 25, 2019

seglo force-pushed the seglo/pass-settings-to-producer branch from 98203c2 to 662f6b9 Compare October 25, 2019 20:56

ennru reviewed Oct 26, 2019

View reviewed changes

seglo force-pushed the seglo/pass-settings-to-producer branch from 35d78e7 to 666e84d Compare October 26, 2019 21:19

seglo requested a review from ennru October 28, 2019 18:59

ennru reviewed Oct 28, 2019

View reviewed changes

seglo force-pushed the seglo/pass-settings-to-producer branch from 134674a to 990480c Compare October 29, 2019 20:00

ennru approved these changes Oct 29, 2019

View reviewed changes

seglo force-pushed the seglo/pass-settings-to-producer branch from 990480c to afae626 Compare October 29, 2019 21:04

seglo merged commit f6fe343 into akka:master Oct 29, 2019

seglo deleted the seglo/pass-settings-to-producer branch October 29, 2019 23:38

ennru mentioned this pull request Nov 11, 2019

Producer: remove unreleased variants; simplify producer passing #967

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Pass ProducerSettings to stage #952

Pass ProducerSettings to stage #952

seglo commented Oct 24, 2019 •

edited

Loading

ennru left a comment •

edited by seglo

Loading

ennru Oct 26, 2019

seglo Oct 26, 2019

ennru Oct 27, 2019

seglo commented Oct 28, 2019

ennru Oct 28, 2019

seglo Oct 28, 2019

ennru Oct 29, 2019

seglo Oct 29, 2019

seglo Oct 29, 2019

seglo Oct 29, 2019

ennru Oct 29, 2019

seglo Oct 29, 2019

ennru left a comment

Pass ProducerSettings to stage #952

Pass ProducerSettings to stage #952

Conversation

seglo commented Oct 24, 2019 • edited Loading

Purpose

References

Changes

ennru left a comment • edited by seglo 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

seglo commented Oct 28, 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ennru left a comment

Choose a reason for hiding this comment

seglo commented Oct 24, 2019 •

edited

Loading

ennru left a comment •

edited by seglo

Loading