[Data] Rebased get_eligible_operators onto has_pending_bundles

[alexeykudinkin]

Why are these changes needed?

Cleaning up get_eligible_ops onto has_pending_bundles method.

Related issue number

Checks

• I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
• I've run pre-commit jobs to lint the changes in this PR. (pre-commit setup)
• I've included any doc changes needed for https://docs.ray.io/en/master/.
  • I've added any new APIs to the API Reference. For example, if I added a
    method in Tune, I've added it in doc/source/tune/api/ under the
    corresponding .rst file.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested :(

[gemini-code-assist]

Code Review

This pull request refactors the codebase to use the new has_pending_bundles method, which simplifies checks for pending data in operator input queues. The changes are mostly straightforward and improve code clarity by abstracting away the implementation details of checking for pending bundles. I've found one area for improvement in the implementation of has_pending_bundles itself for better performance, which I've detailed in a specific comment.

[gemini-code-assist]

The current implementation of has_pending_bundles calculates the sum of lengths of all input queues by calling _pending_dispatch_input_bundles_count(), which can be inefficient as it doesn't short-circuit. A more efficient approach would be to use any() to check if any input queue is non-empty, which was the behavior of the method it replaced. Additionally, the new method is missing a docstring.

Suggested change

	def has_pending_bundles(self) -> bool:
	return self._pending_dispatch_input_bundles_count() > 0
	def has_pending_bundles(self) -> bool:
	"""Check if the operator has any pending bundles in its input queues."""
	return any(q for q in self.input_queues)