[Java] Add headers provider

[danilonajkov-db]

What changes are proposed in this pull request?

Adds custom HeadersProvider authentication to the Java SDK, matching the Rust stream-builder API.

• Adds the HeadersProvider interface with getHeaders() and optional invalidate().
• Adds StreamBuilder.headersProvider() as an alternative to .oauth().
• Supports JSON, Protocol Buffer, and Arrow streams.
• Retains the provider for stream recovery and explicit stream recreation.
• Keeps the deprecated legacy create*Stream APIs unchanged.
• Ensures JNI callbacks run on blocking threads and reclaim local references after every invocation.
• Validates custom gRPC metadata keys fallibly instead of allowing invalid names to panic.
• Documents that header names should come from a small, fixed set.

Authentication Behavior

.oauth() and .headersProvider() are mutually exclusive. Calling either replaces the previously configured authentication method.

When the server rejects credentials, the SDK calls HeadersProvider.invalidate() before retrying so providers can clear cached tokens.

Testing

• cargo fmt --all --check
• cargo check -p zerobus-jni
• cargo test -p databricks-zerobus-ingest-sdk invalid_custom_header_name_returns_error
• Java 8-compatible compilation of the SDK and provider harness
• Staging E2E:
  • Provider fetched an OAuth token inside getHeaders()
  • Created a JSON stream through StreamBuilder.headersProvider()
  • Ingested a record
  • Flushed successfully

[teodordelibasic-db]

Mostly edge cases and races flagged by LLM.

[teodordelibasic-db]

A callback that exceeds recoveryTimeoutMs keeps running after the stream attempt times out. Tokio cannot cancel a spawn_blocking task once it has started, so dropping the future in rust/sdk/src/stream/grpc/connection.rs:80 leaves the Java call holding its worker and cloned GlobalRef. The next retry can start another callback, and repeated stream attempts can accumulate blocked workers and retained provider objects.

Could we serialize calls per provider and hold the guard inside the blocking closure, so cancellation cannot release it while the Java call is still running?

pub struct JavaHeadersProvider {
    provider_ref: GlobalRef,
    callback_lock: Arc<tokio::sync::Mutex<()>>,
}

let guard = Arc::clone(&self.callback_lock).lock_owned().await;
let provider_ref = self.provider_ref.clone();
tokio::task::spawn_blocking(move || {
    let _guard = guard;
    get_headers_blocking(provider_ref)
})
.await

This would not make a synchronous Java call cancellable, but it would keep one stuck provider adapter from consuming a new worker on every retry. A process-wide bounded executor would give a stronger global bound.