Replace AirflowException with ValueError in ProduceToTopicOperator

[FrankYang0529]

Based on CLAUDE.md, the community is actively reducing AirflowException usage. Replace raise AirflowException(...) with raise ValueError(...) in ProduceToTopicOperator.

• related: Move template-field validation/transformation out of operator __init__ (exemption-list burn-down) #70296

---

Was generative AI tooling used to co-author this PR?

• Yes - Claude Code

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[shahar1]

Apparently #70333 was opened earlier and addresses the problem more precisely.
One thing in mind - here you tried to address two issues in one PR (validate_operators_init_exemption.txt + known_airflow_execptions) - I prefer to handle each separtely.

Feel free to tackle the other operators (there are still plenty), or to turn this PR into addressing the known_airflow_exceptions part once the other is merged.

[FrankYang0529]

I will turn this one into addressing the known_airflow_exceptions part after #70333 is merged.

[shahar1]

We might need to take an intermediate step of creating an exception class that derives from both AirflowException and ValueError, because it will break the retry mechanism that is based on detecting the exception type.
It had been discussed in the past but was put on hold, I'll try to revive this discussion.

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.

[jason810496]

I wonder should we introduced compatible layer for the exception (e.g. subclass of both AirflowException and ValueError) just in case users leverage the callback hook and catch against the AirflowException.

If the scenario I mentioned is valid, this is kind of a breaking change for the users.

[xBis7]

What is the goal of changing the exception class? Is it to take a step closer to cleaning up the AirflowException type or is it to make sure that when we get an error, it's easier to debug it?

because it will break the retry mechanism that is based on detecting the exception type.

I've looked at the code and I don't think that there is an issue there. I don't see any check for AirflowException.

But there is a good chance that users are wrapping the operator's execute() call with a try - except AirflowException. As far as users are concerned, this is a breaking change, just like @jason810496 said.

If we can determine the goal for this change, then we can figure out what direction to follow. For example, if this change is going to happen across all operators then we might want to create a new class type extending AirflowException as already suggested.

[FrankYang0529]

Thanks for the review. This PR started as part of #70296, which moves template field validation out of __init__. #70333 fixed that part for this operator first, so I turned this PR into another change I found along the way: replacing raise AirflowException(...) with raise ValueError(...). The change follows the contributing guide, and AGENTS.md says "When you touch code that already raises AirflowException, prefer narrowing it to a more specific exception rather than leaving or duplicating it."

Several merged PRs have already replaced AirflowException with a built-in exception in code that runs during a task. None of them added a class that extends AirflowException.

• Validate DatabricksSQLStatementsSensor exclusivity at __init__ #70831
• Use ValueError when no ArangoDB collection operation is specified #70539
• Use ValueError for unsupported recursive glob in ADLS upload #70540
• Fix AirbyteJobSensor marking a cancelled job as success in deferrable mode #69786
• Add deferrable mode to QuickSight operator and sensor #70218
• Replace AirflowExceptions with native exceptions in Snowflake provider #66279
• Use built-in exceptions for Postgres hook input validation #70537
• Use built-in exceptions for JDBC sqlalchemy connection extras #70535
• In GCP Datafusion hook handle start_pipeline() when response is None / Invalid #60688
• Add Support for GitHub App Installation Authentication in GithubHook #54812
• Handle "Too Many Request" for Azure Instance Metadata Service #55462
• feat: add async discord notifier #56911

A user who wraps any of these in try/except AirflowException runs into the same change as with this PR. Should these also get a new class that extends AirflowException? If so, it seems better to decide that once for all providers rather than in this PR alone. I'm happy to add the class here if we go that way.

[jason810496]

A user who wraps any of these in try/except AirflowException runs into the same change as with this PR.

My concern is the user facing exception handler like on_failure_callback since 2.x or the retry policy introduced in 3.x.

Should these also get a new class that extends AirflowException? If so, it seems better to decide that once for all providers rather than in this PR alone. I'm happy to add the class here if we go that way.

The best approach I can think of is introducing compatible exception in core and wait for the new core release, then add the compatible import of that new exception in the providers.

Or another direction is just keep the exception as-is and don't fix, since the swapping the exception for code style best practices might introduce bigger user impact than we expect.