enhancement: migrate remaining UnlimitedChannel consumers to context-aware get APIs

Problem

In this PR we introduced UnlimitedChannel.GetWithContext, but most long-running UnlimitedChannel consumers still use Get() / GetMultipleNoGroup() and therefore cannot observe ctx.Done() directly while blocked on an empty queue.

Today some places work around this by closing the channel from another goroutine when the context is canceled. That works, but it makes shutdown correctness depend on out-of-band close ordering instead of on the blocking call itself.

Examples in the current tree:

• downstreamadapter/sink/kafka/sink.go
  • calculateKeyPartitions
  • nonBatchEncodeRun
  • batch still blocks on GetMultipleNoGroup
• downstreamadapter/sink/pulsar/sink.go
  • calculateKeyPartitions
  • nonBatchEncodeRun
  • batch still blocks on GetMultipleNoGroup
• downstreamadapter/sink/redo/sink.go
  • sendMessages
• downstreamadapter/sink/mysql/causality/conflict_detector.go
  • Run
• downstreamadapter/dispatcher/block_event_executor.go
  • worker loop on executor.ready.Get()

This leaves us with inconsistent cancellation semantics:

• cloudstorage now has a direct context-aware queue wait path;
• other components still rely on channel closure to unblock workers;
• future refactors can easily miss the required close-on-cancel wiring.

What would you like to be added

Audit and migrate the remaining long-lived UnlimitedChannel consumers to context-aware APIs.

Concretely:

1. Use GetWithContext(ctx) for Get() call sites whose goroutine lifetime is governed by a context.
2. Add context-aware multi-get APIs if needed, for example a GetMultipleNoGroupWithContext(ctx, ...), so batch consumers do not have to keep relying on channel closure for shutdown.
3. Keep plain Get() only in places where the intended ownership model is explicitly "channel close ends the consumer".
4. Update comments/tests so the expected shutdown semantics are explicit.

Describe alternatives you've considered

1. Keep the current pattern and continue closing channels on ctx.Done().

   • This is workable, but easy to forget and spreads cancellation semantics across multiple goroutines.

2. Migrate only cloudstorage and leave the rest unchanged.

   • This keeps behavior inconsistent across sinks and executors, even though the same primitive is shared.

3. Replace all Get() call sites immediately in the current PR.

   • This would broaden the PR scope significantly. Tracking the remaining work in a follow-up issue is cleaner.

Teachability, Documentation, Adoption, Migration Strategy

This is an internal enhancement / cleanup issue, not a user-facing feature.

The migration should be mostly mechanical, but there are two points that need care:

• batch consumers currently depend on GetMultipleNoGroup() and will need a context-aware variant before they can be fully migrated;
• some loops intentionally tie lifetime to channel ownership rather than to context cancellation, so those should be left as Get() and documented as such.

The goal is to make shutdown behavior easier to reason about:

• if a goroutine is context-scoped, its queue wait should also be context-scoped;
• channel close should remain a data-structure lifecycle signal, not the only way to interrupt a blocked consumer.

Start by reading UnlimitedChannel.GetWithContext and auditing the listed consumers in the Kafka, Pulsar, Redo, MySQL causality, and block event executor files. Identify context-scoped waits and batch calls, then update them and their comments/tests so context cancellation is explicit while intentional channel-close ownership remains documented.