[Data] Fixed retry policy for hash-shuffle tasks (#57572)

<!-- Thank you for your contribution! Please review
https://github.com/ray-project/ray/blob/master/CONTRIBUTING.rst before
opening a pull request. -->

<!-- Please add a reviewer to the assignee section when you create a PR.
If you don't have the access to it, we will shortly find a reviewer and
assign them to your PR. -->

## Why are these changes needed?

Currently, Hash shuffle relies on default settings of `max_retries` for
tasks which is incorrect. Instead, we explicitly configure it to be
retrying indefinitely.

## Related issue number

<!-- For example: "Closes #1234" -->

## 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](https://docs.ray.io/en/latest/ray-contribute/getting-involved.html#lint-and-formatting))
- [ ] 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 :(

---------

Signed-off-by: Alexey Kudinkin <ak@anyscale.com>

Author: alexeykudinkin

python/ray/data/_internal/execution/operators/hash_shuffle.py

@@ -657,6 +657,8 @@ def _do_add_input_inner(self, input_bundle: RefBundle, input_index: int):
             ] = _shuffle_block.options(
                 **shuffle_task_resource_bundle,
                 num_returns=1,
+                # Make sure tasks are retried indefinitely
+                max_retries=-1,
             ).remote(
                 block_ref,
                 input_index,
@@ -896,7 +898,7 @@ def _on_aggregation_done(partition_id: int, exc: Optional[Exception]):
 
             # Request finalization of the partition
             block_gen = aggregator.finalize.options(
-                **finalize_task_resource_bundle
+                **finalize_task_resource_bundle,
             ).remote(partition_id)
 
             self._finalizing_tasks[partition_id] = DataOpTask(
@@ -1484,7 +1486,10 @@ def start_health_monitoring(self):
         self._pending_aggregators_refs = None
 
 
-@ray.remote
+@ray.remote(
+    # Make sure tasks are retried indefinitely
+    max_task_retries=-1
+)
 class HashShuffleAggregator:
     """Actor handling of the assigned partitions during hash-shuffle operation