KafkaProducer.produce fails fast

[rodrigo-molina]

Description

Introduce KafkaProducer.failFastProduce producer configuration:

fs2-kafka/modules/core/src/main/scala/fs2/kafka/ProducerSettings.scala

Lines 218 to 228 in aed7ab3

	/**
	* Controls whether [[fs2.kafka.KafkaProducer.produce]] fails immediately if any
	* [[org.apache.kafka.clients.producer.KafkaProducer.send]] callback resolves with error.
	*
	* When set to `true`, the `produce` method will fail fast, returning an error as soon as any
	* record in the [[ProducerRecords]] fails to be sent.
	*
	* The default value is `false`, meaning the `produce` method will not fail fast and will
	* continue processing other records even if some callbacks fail.
	*/
	def failFastProduce: Boolean

Proposed solution

When failFastProduce is set to true, each fs2.kafka.producer.produce call of ProducerRecords creates a scala.concurrent.Promise[Throwable]. This Promise resolution is raced against calling org.apache.kafka.clients.producer.KafkaProducer.send per record. Any failing callback will complete the promise with the failure, canceling the rest of the pending org.apache.kafka.clients.producer.KafkaProducer.send and in-flight callbacks.

fs2-kafka/modules/core/src/main/scala/fs2/kafka/KafkaProducer.scala

Lines 186 to 198 in 2559d96

	if (failFastProduce)
	Async[F]
	.delay(Promise[Throwable]())
	.flatMap { produceRecordError =>
	Async[F]
	.race(
	Async[F]
	.fromFutureCancelable(Async[F].delay(produceRecordError.future, Async[F].unit)),
	produceRecords(produceRecordError.some)
	)
	.rethrow
	}
	else produceRecords(None)

Explanation

fs2.kafka.KafkaProducer.produce returns F[F[...]], where the first effect invokes the underlying org.apache.kafka.clients.producer.KafkaProducer.send and attaches promises to the corresponding callbacks. The second effect is waiting for the callbacks’ resolution. In use cases where all the records in the batch need to be confirmed, any callback failure will cause the final effect resolution to fail, achieving the same outcome sooner.

In non-transient errors such as invalid username, password, or non-existent topics, org.apache.kafka.clients.producer.KafkaProducer.send is blocked for max.block.ms, default to 60 seconds. For example, a fs2.kafka.producer attempting to produce N ProducerRecords will take at least N * 60 seconds to fail. In other words, faulty Kafka producers will take considerable time to fail by repeatedly calling brokers for metadata synchronization.

See:

• max.block.ms default
• KafkaProducer.doSend waitOnMetadata

[aartigao]

Hey Rodrigo! Thanks for your contribution 🙌🏽

I have some doubts about this approach, though. What if, as a user, I don't want it to fail fast? (because I don't care if some records in the batch fail?). This is valid scenario too. As an example, at $WORK we push constantly the locations of couriers coming from mobile app, and we don't want a single record to ruin the whole batch (because in a matter of seconds, a new location will be sent anyways).

But I see your point here, and it's a valid scenario too. Thing is, to me, this is a behavioral change that could potentially affect current users of the lib. And we can't introduce this kind of changes straight away. I'd be more than happy to add it as a feature under a configuration setting in ProducerSettings (a setting disabled by default to be compatible with all users). Something like failFastBatch or failFastSend or other naming suggestions.

Let me know what you think about it 😉

[rodrigo-molina]

Hey Alan, thanks for your detailed response.

As you mentioned, I agree with the importance of not disrupting the current behavior, especially since the library supports fire-and-forget scenarios.

I'll work on making the producer fail fast feature configurable. What do you think about the name failFastProduce? It aligns with KafkaProducer.produce.

[aartigao]

Sounds good!

[rodrigo-molina]

I've updated the code and the PR description accordingly.

Please let me know your thoughts 😄

[erikvanoosten]

Another relevant question: why does the kafka client not fail immediately for non-transient errors such as invalid username, password, and non-existing topics?

[rodrigo-molina]

Another relevant question: why does the kafka client not fail immediately for non-transient errors such as invalid username, password, and non-existing topics?

Thanks @erikvanoosten for chipping in.

In my opinion, non-transient errors should fail fast since they will continue to fail (and block the producer with repeated metadata request calls) until the error is resolved externally.

While this discussion is relevant to the scope of this PR, it may be interpreted as an extra feature. On the one hand, there's the idea of making the producer fail fast on any failure within the batch; on the other hand, there's the possibility of failing fast only on specific, predefined errors (assumed to be non-recoverable).

A proposal to address both is allowing users to configure which specific exceptions trigger the fail-fast behavior. This way, users can define what qualifies as a non-transient error for their specific use case and/or decide which errors to fail fast on. To simplify this for users, we could even provide a set of default error types known to be non-transient in Kafka. To avoid changing the producer's current semantics, we can disable the feature by default (no exception is failed fast by default). Things I like about this proposal:

1. The fail fast feature becomes highly configurable for users
2. Some defaults provide recommendations for basic use cases (like fail fast on non-transient errors) and prevent users from deep diving into some Kafka exceptions
3. Current KafkaProducer.produce semantics are not changed

Challenges to the proposal:

1. For "things I like second point", distinguishing some errors can be tricky and non-consistent over Kafka client versions. For instance, a topic not found is a TimeoutException with the message "Topic $nonExistentTopic not present in metadata after $max.block.ms ms.". If we don't want to maintain this as part of the library, we can avoid providing the defaults; the proposal still gives users enough autonomy to implement "fail fast on non-transient errors" in their context.

Finally, as an implementation detail, fail-fast errors can be provided as a partial function in the ProducerSettings. Something like:

ProducerSettings {
...
   /** 
     * Controls whether [[fs2.kafka.KafkaProducer.produce]] fails immediately if any 
     * [[org.apache.kafka.clients.producer.KafkaProducer.send]] callback resolves with error. 
     * 
     * When set to `true`, the `produce` method will fail fast, returning an error as soon as any 
     * record in the [[ProducerRecords]] fails to be sent. 
     * 
     * The default value is `false`, meaning the `produce` method will not fail fast and will 
     * continue processing other records even if some callbacks fail. 
     */ 
   def failFastProduce: Boolean 


   /** 
     * Controls which exceptions makes [[fs2.kafka.KafkaProducer.produce]] fails immediately.
     * 
     * This setting is only applied when [[failFastProduce]] is set to true.
     * 
     * The default value is to fail fast on all exceptions.
     */ 
   def failFastProduceOn: PartialFunction[Throwable, Boolean]

...
}

Please let me know your thoughts.

[erikvanoosten]

While this discussion is relevant to the scope of this PR, it may be interpreted as an extra feature...

Yeah I agree that this PR is a nice and performant solution (even though it is a kludge, as in, an inelegant solution). We can't wait for the kafka maintainers to fix this properly (if they will at all).

[aartigao]

Instead of defining here the default value, wouldn't it be more convenient to set it in the create method where the other default values are set?

[rodrigo-molina]

It's addressed in commit: 5472f25

[aartigao]

A Promise is created but never listened (if I read correctly). Is that deliberate?

[erikvanoosten]

The author of this PR is on holiday so I'll try to answer for him: it seems that the race that is present in modules/core/src/main/scala/fs2/kafka/KafkaProducer.scala, was not added here. It is my gues that this was an oversight.

[rodrigo-molina]

yes, the race is missing.
I'll add it.

[rodrigo-molina]

good catch. It's added in commit 6442060

I also added a test for it:

fs2-kafka/modules/core/src/test/scala/fs2/kafka/TransactionalKafkaProducerSpec.scala

Lines 479 to 489 in 6442060

	val result = intercept[RuntimeException] {
	TransactionalKafkaProducer
	.stream(settings)
	.evalMap(_.produce(Chunk.singleton(committable)))
	.compile
	.lastOrError
	.unsafeRunSync()
	}
	result shouldBe expectedErrorOnSecondRecord
	assert(transactionAborted, "The transaction should be aborted")

An interesting discovery is that KafkaProducer.produce and TransactionalKafkaProducer.produce behave differently. Specifically, KafkaProducer.produce defers the responsibility of waiting for callback resolutions to the user, as the F[F[_]] result. In contrast, TransactionalKafkaProducer.produce waits for the callbacks.

fs2-kafka/modules/core/src/main/scala/fs2/kafka/KafkaProducer.scala

Lines 45 to 47 in 64a751e

	def produce(
	records: ProducerRecords[K, V]
	): F[F[ProducerResult[K, V]]]

fs2-kafka/modules/core/src/main/scala/fs2/kafka/TransactionalKafkaProducer.scala

Lines 41 to 43 in 64a751e

	def produce(
	records: TransactionalProducerRecords[F, K, V]
	): F[ProducerResult[K, V]]

fs2-kafka/modules/core/src/main/scala/fs2/kafka/TransactionalKafkaProducer.scala

Lines 144 to 148 in 64a751e

	val produce = records
	.traverse(
	KafkaProducer.produceRecord(keySerializer, valueSerializer, producer, blocking)
	)
	.flatMap(_.sequence)

fs2-kafka/modules/core/src/main/scala/fs2/kafka/KafkaProducer.scala

Lines 178 to 180 in 64a751e

	records
	.traverse(produceRecord(keySerializer, valueSerializer, producer, blocking))
	.map(_.sequence)

note the flatMap and map difference.

[rodrigo-molina]

I couldn't help myself here. Please let me know if you prefer to keep this change in a different PR.