[Data] Remove default task-level timeout and clamp end_offset to watermark in Kafka datasource

[owenowenisme]

Description

• Default timeout_ms to None (no task-level timeout) for read_kafka. Each consumer.consume() call
  is already capped at 10s, which is sufficient.
• Clamp end_offset to the high watermark during offset resolution so reads never target offsets
  beyond available data.
• Raise ValueError if start_offset exceeds the clamped end_offset.
• Simplify the read loop by removing per-consume timeout shrinking logic.

Related issues

Link related issues: "Fixes #1234", "Closes #1234", or "Related to #1234".

Additional information

Optional: Add implementation details, API changes, usage examples, screenshots, etc.

[gemini-code-assist]

Code Review

This pull request introduces several improvements to the Kafka datasource. It removes the default task-level timeout, simplifying the timeout logic by relying on a fixed 10-second timeout for each consume call. A key change is clamping the end_offset to the high watermark, which robustly prevents read tasks from hanging when trying to read beyond available data. The logic for handling invalid offset ranges is also improved, now raising a ValueError if the start_offset exceeds the clamped end_offset. The code is cleaner, more robust, and the accompanying test changes correctly validate the new behavior. Overall, these are excellent changes.

_{Note: Security Review did not run due to the size of the PR.}

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

[slfan1989]

Generally agree, except for one small point.

[slfan1989]

Is it expected that the timeout changes from 10s to None?

Consider keeping timeout_ms=10000 as default for backward compatibility？

[bveeramani]

@owenowenisme did the primary consumer of this API request we change the default?

@slfan1989 this API is labeled as stability="alpha", so we can technically break the defaults while still honoring our backwards-compatibility obligations, though it's not ideal

[slfan1989]

Thank you very much for the explanation, LGTM.