[ML] Support batch cancel (internal only) by tags within a workspace (Azure#28712)

* [ML] Support batch cancel (internal only) by tags within a workspace

* Fix tag params

* Fix job.list() ignoring tags

Author: TonyJ1

sdk/ml/azure-ai-ml/azure/ai/ml/operations/_job_operations.py

@@ -256,6 +256,7 @@ def list(
             scheduled=schedule_defined,
             schedule_id=scheduled_job_name,
             **self._kwargs,
+            **kwargs,
         )
 
     def _handle_rest_errors(self, job_object):
@@ -312,7 +313,7 @@ def show_services(self, name: str, node_index: int = 0) -> Dict[str, ServiceInst
 
     @distributed_trace
     # @monitor_with_activity(logger, "Job.Cancel", ActivityType.PUBLICAPI)
-    def begin_cancel(self, name: str) -> LROPoller[None]:
+    def begin_cancel(self, name: str, **kwargs) -> LROPoller[None]:
         """Cancel job resource.
 
         :param str name: Name of the job.
@@ -322,12 +323,30 @@ def begin_cancel(self, name: str) -> LROPoller[None]:
         :rtype: ~azure.core.polling.LROPoller[None]
         :raise: ResourceNotFoundError if can't find a job matching provided name.
         """
-        return self._operation_2022_12_preview.begin_cancel(
-            id=name,
-            resource_group_name=self._operation_scope.resource_group_name,
-            workspace_name=self._workspace_name,
-            **self._kwargs,
-        )
+        tag = kwargs.get("tag", None) if kwargs else None
+
+        if not tag:
+            return self._operation_2022_12_preview.begin_cancel(
+                id=name,
+                resource_group_name=self._operation_scope.resource_group_name,
+                workspace_name=self._workspace_name,
+                **self._kwargs,
+                **kwargs,
+            )
+
+        # Note: Below batch cancel is experimental and for private usage
+        results = []
+        jobs = self.list(tag=tag)
+        # TODO: Do we need to show error message when no jobs is returned for the given tag?
+        for job in jobs:
+            result = self._operation_2022_12_preview.begin_cancel(
+                id=job.name,
+                resource_group_name=self._operation_scope.resource_group_name,
+                workspace_name=self._workspace_name,
+                **self._kwargs,
+            )
+            results.append(result)
+        return results
 
     def _try_get_compute_arm_id(self, compute: Union[Compute, str]):
         # TODO: Remove in PuP with native import job/component type support in MFE/Designer