zifeif2 opened a new pull request, #59285: URL: https://github.com/apache/spark/pull/59285
### What changes were proposed in this pull request? This PR reacquires a Kafka consumer after the out-of-range recovery path releases the old consumer back to the pool. It also adds a regression test that closes the returned consumer immediately to reproduce pool cleanup racing with recovery. ### Why are the changes needed? When `KafkaDataConsumer` handles an `OffsetOutOfRangeException`, it releases its borrowed consumer to discard cached state. With `failOnDataLoss=false`, recovery then used the released local reference to query the available offset range and retry the fetch. The pool can close that idle consumer or lend it to another task, so using the stale reference can fail with an `IllegalStateException` or access a consumer no longer exclusively owned by the task. Reacquiring through the pool restores exclusive ownership before recovery continues. ### Does this PR introduce _any_ user-facing change? Yes. Kafka reads with `failOnDataLoss=false` no longer fail intermittently when pool cleanup coincides with recovery from an out-of-range offset. There is no API change. ### How was this patch tested? Added a regression test that forces pool cleanup when the invalid consumer is returned. The targeted `KafkaDataConsumerSuite` test passed in an equivalent downstream Spark build: ``` KafkaDataConsumerSuite -- -z "reacquire consumer before recovering from an out-of-range offset" ``` `git diff --check` passed. The Apache SBT invocation was also attempted, but this host could not resolve Maven Central while loading the project, before compilation or test execution. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Codex (GPT-6). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
