Skip to content

[SPARK-27042][SS] Invalidate cached Kafka producer in case of task retry #26470

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 2 commits into from
Closed

[SPARK-27042][SS] Invalidate cached Kafka producer in case of task retry #26470

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
25446ba
[SPARK-27042][SS] Invalidate cached Kafka producer in case of task retry
gaborgsomogyi Nov 11, 2019
d384324
Comment fix
gaborgsomogyi Nov 13, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion ...nal/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/CachedKafkaProducer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ import scala.util.control.NonFatal

import org.apache.kafka.clients.producer.{Callback, KafkaProducer, ProducerRecord}

import org.apache.spark.SparkEnv
import org.apache.spark.{SparkEnv, TaskContext}
import org.apache.spark.internal.Logging
import org.apache.spark.kafka010.{KafkaConfigUpdater, KafkaRedactionUtil}
import org.apache.spark.sql.kafka010.InternalKafkaProducerPool._
Expand Down Expand Up @@ -93,6 +93,10 @@ private[kafka010] object CachedKafkaProducer extends Logging {
.setAuthenticationConfigIfNeeded()
.build()
val key = toCacheKey(updatedKafkaParams)
if (TaskContext.get != null && TaskContext.get.attemptNumber >= 1) {
Copy link
Contributor

@vanzin vanzin Nov 13, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I know this is just matching what the consumer code does, but I'm a little unsure that it's correct (in both cases).

Can't you re-use the consumer in different jobs? If so, you can have:

  • job 1 task 1 attempt 0 runs with producer 1 and fails
  • job 1 task 1 attempt 1 runs with producer 2 and succeeds
  • job 2 task 1 attempt 0 runs with producer 1 and re-uses the same instance that should have been closed?

So I'm wondering if instead the task itself shouldn't be invalidating the producer when it detects an error.

Copy link
Contributor

@HeartSaVioR HeartSaVioR Nov 14, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Strictly saying, we should discard the instance (not instances) when the instance throws some error, but then we may need to wrap the code to catch exception and discard the instance wherever we use the instance. A bit verbose, but we still have to return the instance to the pool even it succeeds, so maybe not impossible.

I'm not aware whether Kafka consumer or Kafka producer is self-healing - if they provide such functionality, we wouldn't need to even discard them at any case.

Copy link
Contributor Author

@gaborgsomogyi gaborgsomogyi Nov 14, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To answer the question1: Yes, job 2 can use job 1 created producers in the same JVM. Such case another task retry is needed to recover in the following way:

  • job 2 task 1 attempt 0 runs with producer 1 and fails
  • job 2 task 1 attempt 1 runs with producer 3 (this would invalidate producer 1 and 2 if not used)

If I consider the use-case this would mean 2 jobs must write the same TopicPartition at the same time. Until now I haven't seen use-cases where the exact same topic would be used by 2 different job (of course this doesn't mean it's not done).

To answer the question2: Kafka consumers and producers doesn't provide any kind of self healing.

I think I have to take some time to check the other way because I've similar fear what @HeartSaVioR mentioned (horror complicated code with catch blocks all the places where producer touched). The general fear what I have with the invalidation from exception side is the following: if the code is leaking and doesn't catch one single exception then the task may not heal itself.

Copy link
Contributor

@vanzin vanzin Nov 14, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

2 jobs must write the same TopicPartition at the same time

Why at the same time? All you need to hit my case is two tasks on the first job using different producers, and can't that happen e.g. if you have executors with more than one core?

It seems like Spark would eventually self-heal at the cost of some failed tasks, but can't we do better? As Jungtaek said we already have to return the producer to the pool. It's just a matter of, instead of returning it, invalidating it if an error is detected.

Copy link
Member

@zsxwing zsxwing Nov 14, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IIUC, this is trying to close a shared producer because of a random failure (including user code errors)? IMO, we should not do this for producer. It can be shared with multiple Spark jobs as long as they are using the same Kafka parameters. Kafka actually suggests to share the producer in their doc:

The producer is thread safe and sharing a single producer instance across threads will generally be faster than having multiple instances.

Hence I would assume it can self-heal. Kafka consumer is a different story. It's not thread-safe and cannot be shared with multiple tasks at the same time. That's why we can close the old one since we are pretty sure it's not used by another task in the same JVM.

Copy link
Contributor

@HeartSaVioR HeartSaVioR Nov 15, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IIUC, this is trying to close a shared producer because of a random failure (including user code errors)? IMO, we should not do this for producer. It can be shared with multiple Spark jobs as long as they are using the same Kafka parameters.

We changed the producer pool to follow the same approach as Kafka consumer - #25853 to resolve the long-standing "producer close on idle" issue. #19096 proves closing idle producer while producer can be shard with multiple tasks is very complicated - we should consider adding "in-use" and then should deal with thread-safety. One task one producer is pretty much simpler, and we didn't observe performance issue there.

Copy link
Contributor Author

@gaborgsomogyi gaborgsomogyi Nov 15, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why at the same time?

I thought job 1 and job 2 are different queries. Single query + multiple cores could end-up in the same situation.

Hence I would assume it can self-heal.

I would assume it can't heal if programming failure would be in the producer code itself. On the other had I see your general concern related this. What is your point considering #25853?

Copy link
Member

@zsxwing zsxwing Nov 15, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would assume it can't heal if programming failure would be in the producer code itself.

That's true. Softwares have bugs. But I have never heard any Kafka producer issues reported when using Spark. The task failure most likely is a user error, transient network error, or something unrelated. And a task failure like this should be isolated and should not impact other tasks in the same executor. This seems overkill.

Copy link
Member

@zsxwing zsxwing Nov 15, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What is your point considering #25853?

Thanks for pointing me to this PR. I posted my concerns there.

Copy link
Contributor Author

@gaborgsomogyi gaborgsomogyi Jan 8, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since #25853 rolled back this PR just doesn't make sense, closing...

logDebug(s"Invalidating key $key")
producerPool.invalidateKey(key)
}
producerPool.borrowObject(key, updatedKafkaParams)
}

Expand Down
1 change: 1 addition & 0 deletions external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaDataConsumer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -627,6 +627,7 @@ private[kafka010] object KafkaDataConsumer extends Logging {
kafkaParams: ju.Map[String, Object]): KafkaDataConsumer = {
if (TaskContext.get != null && TaskContext.get.attemptNumber >= 1) {
val cacheKey = new CacheKey(topicPartition, kafkaParams)
logDebug(s"Invalidating key $cacheKey")

// If this is reattempt at running the task, then invalidate cached consumer if any.
consumerPool.invalidateKey(cacheKey)
Expand Down
21 changes: 21 additions & 0 deletions ...afka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/CachedKafkaProducerSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,27 @@ class CachedKafkaProducerSuite extends SharedSparkSession with PrivateMethodTest
assert(producerPool.size(toCacheKey(kafkaParams2)) === 1)
}

test("acquire should return a new instance with Task retry") {
try {
val kafkaParams = getKafkaParams()

val context1 = new TaskContextImpl(0, 0, 0, 0, 0, null, null, null)
TaskContext.setTaskContext(context1)
val producer1 = CachedKafkaProducer.acquire(kafkaParams)
CachedKafkaProducer.release(producer1)

val context2 = new TaskContextImpl(0, 0, 0, 0, 1, null, null, null)
TaskContext.setTaskContext(context2)
val producer2 = CachedKafkaProducer.acquire(kafkaParams)
CachedKafkaProducer.release(producer2)

assert(producer1 !== producer2)
assert(producerPool.size(toCacheKey(kafkaParams)) === 1)
} finally {
TaskContext.unset()
}
}

test("Concurrent use of CachedKafkaProducer") {
val data = (1 to 1000).map(_.toString)
testUtils.createTopic(topic, 1)
Expand Down