Fix KafkaBaseHook.test_connection missing oauth_cb for managed Kafka

[TimurRakhmatullin86]

Problem

KafkaBaseHook.test_connection() builds its AdminClient from the raw connection
configuration and (since the recent callback-resolution change) resolves dotted-path
callbacks — but it never injects the managed OAuth token callback that get_conn adds
for Google Managed Service for Apache Kafka and Amazon MSK IAM connections. As a result,
"Test Connection" in the UI fails for those managed connections even though the hook
itself authenticates and connects correctly — the test path never gets the oauth_cb
token callback.

def test_connection(self) -> tuple[bool, str]:
    try:
        config = self.get_connection(self.kafka_config_id).extra_dejson
        self._resolve_callbacks(config)          # no managed oauth_cb injection
        t = AdminClient(config).list_topics(timeout=10)

Fix

The configuration-building logic shared with get_conn (bootstrap validation, dotted-path
callback resolution, and the managed OAuth token injection for Google Managed Kafka /
Amazon MSK IAM) is extracted into a _build_config() helper that both get_conn and
test_connection use. The UI test now exercises exactly the same configuration as a real
connection, so managed connections test successfully. get_conn behaviour is unchanged (a
pure extraction).

Tests

• New test_test_connection_injects_managed_kafka_oauth asserts the managed oauth_cb is
  present in the config passed to AdminClient from test_connection.
• The existing test_connection unit tests now use a concrete non-managed broker address
  (localhost:9092) instead of a MagicMock, since test_connection now builds the full
  runtime config and a mock host would spuriously match the managed-Kafka path.
• Full providers/apache/kafka/tests/unit/apache/kafka/hooks/test_base.py suite passes
  (23/23); ruff check and ruff format --check are clean.

Rebased onto main after the dotted-path callback-resolution change landed; the diff was
reworked to route both get_conn and test_connection through the shared _build_config
helper (the earlier revision predated that refactor).

[potiuk]

Thanks — and the fix is the right shape rather than the quick one.

The bug is that test_connection() reimplemented config construction: extra_dejson plus _resolve_callbacks(), but not the managed-OAuth injection that get_conn() does. So a Google Managed Kafka or MSK IAM cluster would authenticate fine in a real task and then fail "Test Connection" in the UI — the most confusing possible failure mode, because the thing meant to diagnose connectivity was the only thing broken.

The quick fix would have been to copy the injection into test_connection. Extracting _build_config() and having both paths call it is better: it removes the possibility of the two drifting again, which is exactly how this bug arose. The docstring says so explicitly too.

The test changes are more meaningful than they look. Swapping {"bootstrap.servers": MagicMock()} for a real "localhost:9092" isn't cosmetic — _build_config now inspects the bootstrap string to decide whether managed OAuth applies, and a MagicMock can't be pattern-matched. And test_test_connection_injects_managed_kafka_oauth uses a realistic ...managedkafka.my-project.cloud.goog:9092 host and asserts oauth_cb reaches the AdminClient config, which is precisely the regression.

One question, not blocking: that new test patches airflow.providers.google.cloud.hooks.managed_kafka.ManagedKafkaHook — a google-provider symbol from a kafka-provider test. It's fine in CI where both are installed, but worth confirming the kafka suite isn't expected to run standalone without google present, otherwise the patch target won't import.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting