Describe the bug
When running a pulsar consumer in a spark job in local (not local-cluster) mode, the client cache is shared between executor and driver threads. If a producer disconnects for longer than the client timeout duration and then reconnects the executor's next call to getOrCreate() will evict the old cached client from the cache while the driver still holds references to the client. This causes a RejectedExecutionException, which is caught and logged. The executor thread then hangs indefinitely.
To Reproduce
If requested, I can provide a script that reproduces the bug using a TestContainer for the pulsar instance, but it's all boilerplate. You just create read a stream from a producer, disconnect the producer, and reconnect the producer after the timeout duration has elapsed. If you are running your spark session in local mode, an exception is logged but not raised, and the executor hangs. In local-cluster mode it works as expected.
Steps to reproduce the behavior:
- Setup a pulsar instance, a pulsar producer, and a spark session with
.master(local[2]) and .config("spark.pulsar.client.cache.timeout", "5s").
- Read/write from the stream
- Kill the producer for longer than the cache timeout setting, then recreate the producer.
- The consumer will log the RejectedExecutionException and hang.
Expected behavior
I expect the consumer to reconnect to the producer successfully, as it does when run in local-cluster mode instead of local mode. At a minimum, it should raise the exception instead of logging and hanging so that spark script can handle the exception or the pod exit and be recreated.
Possible Corrective Action
|
protected var client: PulsarClientImpl = |
change
var to
def so stale references aren't reused.
Context
tested with spark 3.4, 4.0, and 4.1. Same behavior in all of them.
Describe the bug
When running a pulsar consumer in a spark job in local (not local-cluster) mode, the client cache is shared between executor and driver threads. If a producer disconnects for longer than the client timeout duration and then reconnects the executor's next call to getOrCreate() will evict the old cached client from the cache while the driver still holds references to the client. This causes a RejectedExecutionException, which is caught and logged. The executor thread then hangs indefinitely.
To Reproduce
If requested, I can provide a script that reproduces the bug using a TestContainer for the pulsar instance, but it's all boilerplate. You just create read a stream from a producer, disconnect the producer, and reconnect the producer after the timeout duration has elapsed. If you are running your spark session in local mode, an exception is logged but not raised, and the executor hangs. In local-cluster mode it works as expected.
Steps to reproduce the behavior:
.master(local[2])and.config("spark.pulsar.client.cache.timeout", "5s").Expected behavior
I expect the consumer to reconnect to the producer successfully, as it does when run in local-cluster mode instead of local mode. At a minimum, it should raise the exception instead of logging and hanging so that spark script can handle the exception or the pod exit and be recreated.
Possible Corrective Action
pulsar-spark/src/main/scala/org/apache/spark/sql/pulsar/PulsarHelper.scala
Line 63 in f9a497f
change
vartodefso stale references aren't reused.Context
tested with spark 3.4, 4.0, and 4.1. Same behavior in all of them.