Large future issues in Pulsar and Kafka crates

[ChrisWrenDev]

Description:
We recently enabled the clippy lints #![warn(clippy::large_futures, clippy::large_stack_frames)] in risingwave_connector. This revealed warnings in both the Pulsar and Kafka source enumerator implementations.

A temporary workaround (Box::pin in the root async method) was applied in RisingWave, but the underlying issue likely comes from these upstream crates.

Next Steps:

• Open issues in the pulsar and kafka repos to raise awareness.
• Contribute PRs upstream to fix the struct allocations.
• After their patched versions are released, bump the dependencies in RisingWave.
• Remove the temporary Box::pin workaround once upstream fixes are in place.

[ChrisWrenDev]

Issue raised in pulsar-rs: streamnative/pulsar-rs#347

[ChrisWrenDev]

Pull request has been raised in pulsar-rs: streamnative/pulsar-rs#348

[ChrisWrenDev]

@wenym1 I've been digging into the clippy::large_futures warning in KafkaSplitEnumerator. The size appears to come from the async compute path in SHARED_KAFKA_ADMIN.try_get_with_by_ref(...) rather than anything inherent to Kafka or Moka. That closure captures several large values (e.g., KafkaProperties, ClientConfig, broker string), which likely inflates the future and triggers the lint.

I tried -Z print-type-sizes to identify the exact contributors, but the output wasn’t actionable for this code path. So the “large capture” diagnosis is a best-current assumption—supported by the observation that the lint disappears when the heavy work is moved out of the cache’s async closure.

A behavior-preserving fix I found is to switch the cached value from Arc<KafkaAdmin> to Arc<OnceCell<Arc<KafkaAdmin>>> and perform admin creation in OnceCell::get_or_try_init(...). The cache closure then only creates/returns an empty OnceCell (small future), while initialization happens outside the compute closure.

This approach keeps the existing time-to-idle semantics and removes the need for Box::pin. If initialization fails, we simply retain an uninitialized cell in the cache until it idles out (the same five-minute TTI), then retry. It also prevents concurrent races that could otherwise create multiple admins for the same key.

I realise this may not be the ideas solution as it adds one level of indirection (Moka + OnceCell) and a slightly more complex cached type, but I haven’t found a simpler change that preserves behavior and clears clippy::large_futures without boxing.

If this direction sounds acceptable, I can open a PR with the OnceCell approach; otherwise I'm open to solution suggestions or we can stick with the existing Box::pin.