c-broggy opened a new issue, #39789:
URL: https://github.com/apache/beam/issues/39789
### What happened?
# Title
KafkaIO OAUTHBEARER auth against GCP Managed Service for Apache Kafka broken
starting 2.69.0 (works on 2.68.0)
# Body
## What happened
Using Python `ReadFromKafka` with `sasl.mechanism=OAUTHBEARER` and
`sasl.login.callback.handler.class=com.google.cloud.hosted.kafka.auth.GcpLoginCallbackHandler`
to connect to **GCP Managed Service for Apache Kafka** from a Dataflow
streaming job
(`DataflowRunner`, `--experiments=use_runner_v2`), the Kafka consumer fails
100% of
connection attempts with:
```
org.apache.kafka.common.errors.TimeoutException: Timeout expired while
fetching topic metadata
at
org.apache.beam.sdk.io.kafka.KafkaIO$Read$GenerateKafkaSourceDescriptor...
```
with worker logs showing, for every connection attempt:
```
[Consumer clientId=..., groupId=...] Connection to node -1 (<broker>:9092)
terminated
during authentication. This may happen due to any of the following reasons:
(1) Firewall blocking Kafka TLS traffic (eg it may only allow HTTPS traffic),
(2) Transient network issue.
```
This is **not** a network/firewall issue — same
VPC/subnetwork/service-account/broker
worked correctly on Beam 2.68.0 and earlier.
## Bisection result
Tested directly against a live GCP Managed Kafka cluster (Dataflow,
`DataflowRunner`,
identical pipeline code, identical JDK build (`17.0.18+8-Debian-1deb12u1`,
controlled via
a pinned custom SDK container image so the JDK is not a variable), identical
`kafka-clients` (3.9.0) and identical `GcpLoginCallbackHandler.class`
bytecode
(SHA256-verified) across every version tested:
| Beam version | Result |
|---|---|
| 2.67.0 | ✅ Works |
| 2.68.0 | ✅ Works (confirmed live data flowing) |
| **2.69.0** | ❌ **Fails** — 100% auth termination |
| 2.71.0 | ❌ Fails |
| 2.75.0 | ❌ Fails |
The break happens exactly between the **2.68.0 → 2.69.0** release.
## Dependency diff, 2.68.0 → 2.69.0
Diffing every `META-INF/maven/*/pom.properties` entry bundled in
`beam-sdks-java-io-expansion-service-{2.68.0,2.69.0}.jar`, after excluding
unrelated file-format IO connectors (ORC/Parquet/JTS, bundled in the same
shadowJar but irrelevant to Kafka), the changes touching the HTTP/auth path
are:
```
org.apache.httpcomponents.client5:httpclient5 5.4.3 -> 5.5
io.confluent:kafka-schema-registry-client 7.6.1 -> 7.5.5 (downgrade)
io.confluent:common-utils 7.6.1 -> 7.5.5 (downgrade)
net.bytebuddy:byte-buddy 1.14.12 -> 1.17.7
```
`google-auth-library-oauth2-http` and `google-auth-library-credentials` are
**unchanged** across this specific boundary (both `1.37.1` / `1.30.0`), so
this
looks like a transport-layer (`httpclient5`) or Confluent
schema-registry-client
regression rather than a `google-auth-library` version bump, despite
`google-auth-library-oauth2-http` moving significantly (1.36.0 → 1.47.0)
across
the wider 2.67.0 → 2.75.0 range.
## What's been ruled out
- `kafka-clients` version: identical (3.9.0) at every version tested
- `GcpLoginCallbackHandler.class`: byte-identical (SHA256-verified) at every
version tested
- `managed-kafka-auth-login-handler` (Google's own artifact): identical,
`1.0.5`
- JDK patch version: identical (`17.0.18+8-Debian-1deb12u1`), controlled via
a pinned
custom SDK container image built via multi-stage `COPY` from a known-good
image,
ruling out JDK/TLS-default drift as a factor
- Network/firewall/subnetwork/service account/IAM: identical config across
all tests;
ruled out because the *same* infra with only the Beam SDK version changed
reproduces
the failure deterministically
## Environment
- Runner: `DataflowRunner`, `--experiments=use_runner_v2`,
`--dataflow_service_options=streaming_mode_at_least_once`
- SDK: Python 3.12, `apache-beam[gcp]`
- Kafka: GCP Managed Service for Apache Kafka, `SASL_SSL` + `OAUTHBEARER`
- Auth: `com.google.cloud.hosted.kafka.auth.GcpLoginCallbackHandler`
(Application
Default Credentials via the pipeline's attached service account)
## Reproduction
Minimal `consumer_config` used:
```python
{
"bootstrap.servers": "<managed-kafka-bootstrap>:9092",
"security.protocol": "SASL_SSL",
"sasl.mechanism": "OAUTHBEARER",
"sasl.login.callback.handler.class":
"com.google.cloud.hosted.kafka.auth.GcpLoginCallbackHandler",
"sasl.jaas.config":
"org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;",
}
```
Passed to `ReadFromKafka(consumer_config=consumer_config, topics=[...])` in a
streaming pipeline submitted with `--sdk_container_image` pinned to a custom
image
per Beam version tested (JDK controlled/identical across all).
Happy to share the full bisection harness (build + throwaway-consumer-group
test
script) if useful for reproducing this quickly across future release
candidates.
### Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
### Issue Components
- [ ] Component: Python SDK
- [ ] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [ ] Component: IO connector
- [ ] Component: Beam YAML
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Infrastructure
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Prism Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [x] Component: Google Cloud Dataflow Runner
--
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]
