[Data] Fixed max_retries for hash shuffle (#57575)

Cherry-pick of #57572

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