Support Workload Identity Federation (keySource=WIF_JSON) for keyless auth from AWS workloads

## Problem

There is currently no way to authenticate the connector to GCP without a long-lived service-account key when running on AWS, in our case Kafka Connect on ECS Fargate. `GcpClientBuilder.KeySource` supports `FILE`, `JSON`, and `APPLICATION_DEFAULT`: the first two require a static SA key, and `APPLICATION_DEFAULT` only looks like it covers the gap. `GoogleCredentials.fromStream` does handle `external_account`, so `GOOGLE_APPLICATION_CREDENTIALS` can point at a WIF config, but the credential it builds takes AWS credentials from `credential_source`, meaning environment variables or EC2 IMDS, and IMDS does not serve task-role credentials: it is unreachable from a Fargate task, and on the EC2 launch type it returns the host's instance profile rather than the task role. ADC is also resolved from process-wide state, so it cannot give each connector its own external_account config.

Workload Identity Federation is Google's supported answer to this (AWS role → STS token exchange → SA impersonation, no stored key). The connector can't express a WIF credential today because `keyfile` is always parsed as a service-account key rather than an `external_account` config.

## Why google-auth can't do this on ECS/Fargate

Even with WIF wired up, the standard google-auth path fails: its `AwsCredentials` never reads the ECS container-credentials endpoint (`169.254.170.2$AWS_CONTAINER_CREDENTIALS_RELATIVE_URI`), where task-role credentials actually live. The connector starts, then fails when it first tries to obtain a token.

This is a long-standing, acknowledged gap upstream:

- [`google-cloud-java#12605`](https://github.com/googleapis/google-cloud-java/issues/12605) — "Workload Identity Federation should support ECS Fargate container credentials", open since 2022-08-01. Earlier reports of the same gap: [#714](https://github.com/googleapis/google-auth-library-java/issues/714), [#794](https://github.com/googleapis/google-auth-library-java/issues/794).
- [`google-auth-library-java#1374`](https://github.com/googleapis/google-auth-library-java/pull/1374) — a PR adding native ECS detection, closed unmerged in April 2024 with explicit guidance: *"we support custom credential suppliers. This is the approach we recommend for now"*, with native support deferred to "a larger overhaul".

The usual workaround, injecting `AWS_ACCESS_KEY_ID`/`AWS_SECRET_ACCESS_KEY`/`AWS_SESSION_TOKEN` at container start (which #1374 was written to avoid), isn't viable: those credentials expire within hours with no refresh, and the env vars shadow the AWS default credential chain for the whole worker JVM, breaking any other plugin or config provider that relies on the task role. So the fix has to be scoped to the connector's own credential construction, which is exactly what a supplier does. That API arrived in [#1336](https://github.com/googleapis/google-auth-library-java/pull/1336) and shipped in 1.23.0, the version this project already gets via `libraries-bom 26.33.0`.

## Proposal

Add a `WIF_JSON` key source in which `keyfile` holds an `external_account` credential configuration:

1. **`KeySource.WIF_JSON`**. The config validator and `getKeySource()` are both derived from `values()`/`valueOf`, so the enum addition wires itself up.
2. **A `WIF_JSON` branch in `credentials()`** that dispatches on the keyfile's `subject_token_type`. AWS (`urn:ietf:params:aws:token-type:aws4_request`, sourced from `ExternalAccountCredentials.SubjectTokenTypes.AWS4` rather than hard-coded) takes the AWS path; anything else fails fast with a clear "AWS only" message rather than guessing.
3. **A small `AwsSecurityCredentialsSupplier`** that reads the ECS container-credentials endpoint and is handed to `AwsCredentials.newBuilder()`. Scoped to the connector's own credential construction, so it sets no process-wide state and doesn't disturb other plugins.

Two design choices worth stating up front:

- **No AWS SDK dependency.** The supplier fetches on demand and never caches. The ECS agent keeps the endpoint fresh, and google-auth calls it only when it needs a new subject token, so the refresh handling that is the SDK's main value isn't needed. That's a single class of under 200 lines and zero new dependencies, versus `awssdk:auth` plus an HTTP client and its own Jackson.
- **`credential_source` is omitted** (and stripped if present). The branch validates through `GCPValidator.validateCredentialJson` as the `JSON`/`FILE` branches do, and its URI allowlist rejects AWS IMDS `credential_source` URIs by default, while a keyfile without one passes untouched, and the supplier provides the AWS credentials anyway. The alternatives are worse: allowlisting the IMDS URI via `io.aiven.commons.envcheck.uri` in every deployment, or bypassing the validator and losing its SSRF protection on `token_url` and the impersonation URL.

No dependency changes are needed: the project already uses `libraries-bom 26.33.0` (google-auth 1.23.0, which has native `external_account` and supplier support), and gson is already on the compile classpath transitively.

## Scope

Included: inline `external_account` JSON via `keyfile`, AWS provider, ECS task credentials (Fargate *and* EC2 launch type; the axis is the presence of `AWS_CONTAINER_CREDENTIALS_RELATIVE_URI`, not the launch type).

Deliberately not included, each a clean follow-up:

- `WIF_FILE` (external_account config read from a path), trivial to mirror from the existing `FILE` branch.
- Non-AWS providers (Azure, OIDC file/URL). These present a plain JWT that google-auth fetches natively via `fromStream` and need `credential_source` left intact, so the dispatcher is structured to let such a branch slot in without inheriting the AWS-specific stripping.
- EC2 instance IMDS (a JVM on an EC2 host using the instance profile rather than an ECS task). This needs `credential_source` kept and no supplier, the opposite of this design, and fails fast with a clear message today.
- `AWS_CONTAINER_CREDENTIALS_FULL_URI` + auth token (EKS Pod Identity).

## Known limitation: `useCredentialsProjectId`

Flagging this rather than burying it, since it interacts with a feature you maintain (#124): **`useCredentialsProjectId=true` does not work with `WIF_JSON`**. The flag works today because `ServiceOptions.Builder.setCredentials` copies a project id off the credential, but only when it is a `ServiceAccountCredentials`, which is what `FILE` and `JSON` produce. The WIF branch builds an `AwsCredentials`, a different branch of the hierarchy, so that check never fires, and there would be nothing to read anyway: an `external_account` config has no `project_id` field. `projectId` falls through to `getDefaultProjectId()`, whose fallbacks (`GOOGLE_CLOUD_PROJECT`, gcloud config, GCE metadata) are all absent on an AWS worker, so client construction fails at startup with "A project ID is required for this service but could not be determined from the builder or the environment."

There is no good workaround. `quota_project_id` is a billing-attribution header, not the client's project id, and the project-like values in a WIF keyfile (the pool project in `audience`, the service account's project in `service_account_impersonation_url`) are not necessarily where jobs should run. Setting `GOOGLE_CLOUD_PROJECT` would satisfy the chain, but then the project comes from the environment rather than the credential, which is both not what the flag promises and the JVM-wide coupling this change avoids. The same would apply to a future `WIF_FILE`: it is the credential class that matters, not how the config arrives.

Suggestion: reject the combination during config validation with a message pointing at `project`, rather than let it surface as an opaque startup failure. Happy to add that here if you agree.

## Open question: behaviour at scale

Everything below is a single connector, so I cannot answer this from measurement. Flagging it as a design question rather than a bug.

The container credentials endpoint is served per task, so every WIF connector in a worker hits the same endpoint in the same container: load concentrates per worker, not per cluster. Each credential instance costs two calls per refresh, and on the REST path refreshes on the token clock, roughly hourly. That is negligible as an average (our 110-connector worker would see about 0.2 calls a second), but a deployment or rebalance makes every connector acquire within seconds of the others and stay aligned afterwards, turning it into a few hundred requests inside a couple of seconds, once an hour. Storage Write API connectors do not join that burst, since they re-acquire on stream lifecycle rather than the clock.

Measured: 6 requests within 15ms at startup and 3.4s at refresh, 11 consecutive cycles overnight, no throttling. Not measured: many WIF connectors in one worker after a synchronised restart. I also know of no published rate limit for this endpoint, only for the task metadata endpoint. It is worth raising because of the consequence rather than the likelihood: `getCredentials` throws on any non-200 with no retry, so one throttled response fails a task, and a failed Connect task stays failed until restarted by hand.

Two mitigations, neither implemented yet, differing mainly in what they cost the design:

- **Bounded retry with jitter** (attempt cap, time budget, retry only 429/5xx/timeouts). All its state is stack-local to one `getCredentials` call, so the class stays stateless, thread-safe without synchronization and trivially serializable. The time budget is bounded by two things: google-auth's 3m45s refresh margin, and `max.poll.interval.ms`, since a blocking refresh stalls the task. Around 20s leaves headroom on both, allowing for two fetches per acquisition. It de-synchronises retries but not the refresh schedule itself, so it is damage control for a burst rather than a cure.
- **A cache**, in one of three sizes. The supplier currently re-reads the endpoint every time it is asked, which is wasteful twice over: the two fetches of an acquisition arrive milliseconds apart and ask for the same thing, and the AWS credentials it returns stay valid for about six hours while the GCP token refreshes hourly. So a one-second TTL would serve the second fetch from memory and halve the calls; caching until the `Expiration` the endpoint already returns would make it one fetch per six-hour rotation instead of two an hour, roughly twelvefold fewer; and one cache shared across the worker JVM would make that a single fetch per rotation for the whole worker, whatever the connector count, which is the only variant that really answers the density question. The cost is that any of them adds state outliving the call, needs the cached field `transient`, falsifies the "stateless, no caching" properties claimed above, and in the `Expiration` form arguably re-implements a slice of the AWS SDK.

I lean towards the retry in the initial PR and would rather leave the caching question to you. Do you know of a rate limit here, or of anyone running WIF at this density? Happy to test 20 to 30 WIF connectors in one worker before the PR if that is more useful than after.

## Status

I have this implemented and working on a fork, with unit tests for the supplier's parse and failure paths and for the `WIF_JSON` build path (19 tests). The branch is rebased on current `main` (as of `d099f780`), and `mvn -P ci clean package` and `mvn -P ci test` both pass on JDK 17 (471 tests, including checkstyle, RAT and the new spotless google-java-format gate).

It has also been exercised against real AWS→GCP auth on Kafka Connect on ECS Fargate, writing to BigQuery with no service-account key present anywhere. Rather than just assert that it works, the specifics that a reviewer might reasonably doubt:

- **Both write paths.** ~244k records across a `useStorageWriteApi=true` run and a `useStorageWriteApi=false` run, 3 tasks over 3 partitions, ~17h total.
- **Token refresh works, is called only when needed, and keeps working.** On the REST path the first refresh landed 56m14s after acquisition, google-auth's 1h token minus its 3m45s refresh margin, i.e. within a second of the predicted time. It then ran unattended for ~10h: **11 consecutive refresh cycles at 56.2 min with no drift, 3 STS exchanges per cycle (one per task), zero errors and zero failed fetches**, writes continuing across every boundary. Between acquisitions ~12.6k records were written with zero calls to the endpoint, so it is genuinely not hit per request.
- **The ~6h ECS task-credential rotation is a non-event**, which is the payoff of the supplier holding no state: the ECS agent rotates the task role's credentials underneath, and because the supplier re-reads the endpoint on every call it simply picks up whatever is current. The refresh cycles either side of that boundary succeeded like any other.
- **Re-acquisition follows the client's connection lifecycle, not the token clock.** On the REST path (`insertAll`) every request consults the credential, so the token refreshes on schedule. On the Storage Write API it is consulted once per gRPC `AppendRows` stream, and the `JsonStreamWriter` cached per table in `StorageWriteApiDefaultStream` holds those streams open: with `useStorageWriteApi=true` the connector ran 5h49m past token expiry, writing continuously, without a single re-acquisition and without failing. Worth knowing before testing, since a steady Storage Write API workload never exercises refresh at all.
- **Each acquisition costs two endpoint fetches but only one STS exchange.** Because this config sets `service_account_impersonation_url`, the credential mints a subject token, then delegates to `ImpersonatedCredentials`, whose source mints its own; the first is discarded. Upstream behaviour rather than anything this branch introduces, but it doubles the endpoint calls and is easy to misread when counting them.
- **Concurrency.** 3 tasks acquire independently on the REST path: 6 endpoint calls within 15ms at startup and 6 within 3.4s at the synchronised refresh, across 3 threads sharing the supplier's static `HttpClient`/`Gson`. No throttling from the ECS agent and no failures. The supplier has no retry (it throws on any non-200), so this was the case I most wanted to see under contention.
- **Data integrity.** 244,022 records with contiguous ids: count, id span and distinct-id count all equal, so zero gaps and zero duplicates across 11 credential rotations.
- The keyfile used carried a `credential_source` block, so the strip-then-validate path is the one exercised end to end.

The work is split into five self-contained commits, each with its own message, so it should already fit the "each commit should be self-contained with an informative message" requirement in CONTRIBUTING without needing a squash.

Happy to open the PR if this is something you'd like in the connector.

Start at GcpClientBuilder: inspect KeySource, credentials(), getKeySource(), and the GCPValidator.validateCredentialJson path described in the issue. Review the implemented AwsSecurityCredentialsSupplier and its supplier parse/failure and WIF_JSON build tests, then run mvn -P ci test; done means the scoped ECS task-credential flow and validation behavior remain covered without regressions.

Escrito por el modelo de indexación a partir del texto del issue.