branch-4.1: [fix](job-manager) cancelTaskById should not be blocked by unrelated streaming jobs #62940 (#63262)

Cherry-picked from #62940

---------

Co-authored-by: wudi <wudi@selectdb.com>

Author: github-actions[bot]

fe/fe-core/src/main/java/org/apache/doris/job/manager/JobManager.java

@@ -424,10 +424,10 @@ public void replayDeleteJob(T replayJob) throws JobException {
      */
     public void cancelTaskById(String jobName, Long taskId) throws JobException {
         for (T job : jobMap.values()) {
-            if (job.getJobConfig().getExecuteType().equals(JobExecuteType.STREAMING)) {
-                throw new JobException("streaming job not support cancel task by id");
-            }
             if (job.getJobName().equals(jobName)) {
+                if (job.getJobConfig().getExecuteType().equals(JobExecuteType.STREAMING)) {
+                    throw new JobException("streaming job not support cancel task by id");
+                }
                 job.cancelTaskById(taskId);
                 job.logUpdateOperation();
                 return;

fe/fe-core/src/test/java/org/apache/doris/job/manager/JobManagerTest.java

@@ -19,16 +19,23 @@
 
 import org.apache.doris.analysis.UserIdentity;
 import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.job.base.AbstractJob;
+import org.apache.doris.job.base.JobExecuteType;
+import org.apache.doris.job.base.JobExecutionConfiguration;
+import org.apache.doris.job.exception.JobException;
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.utframe.TestWithFeService;
 
 import com.google.common.collect.Sets;
 import mockit.Expectations;
 import org.junit.Assert;
 import org.junit.Test;
+import org.mockito.Mockito;
 
 import java.io.IOException;
 import java.util.HashSet;
+import java.util.Map;
 
 public class JobManagerTest {
     @Test
@@ -62,4 +69,37 @@ public void testJobAuth() throws IOException, AnalysisException {
             Assert.assertTrue(e.getMessage().contains("table1"));
         }
     }
+
+    private static AbstractJob mockJob(long id, String name, JobExecuteType type) {
+        AbstractJob job = Mockito.mock(AbstractJob.class);
+        Mockito.when(job.getJobId()).thenReturn(id);
+        Mockito.when(job.getJobName()).thenReturn(name);
+        JobExecutionConfiguration cfg = new JobExecutionConfiguration();
+        cfg.setExecuteType(type);
+        Mockito.when(job.getJobConfig()).thenReturn(cfg);
+        return job;
+    }
+
+    @Test
+    @SuppressWarnings({"unchecked", "rawtypes"})
+    public void testCancelTaskByIdNotBlockedByOtherStreamingJob() throws JobException {
+        JobManager manager = new JobManager();
+        AbstractJob streamingJob = mockJob(1L, "streaming_job", JobExecuteType.STREAMING);
+        AbstractJob batchJob = mockJob(2L, "batch_job", JobExecuteType.RECURRING);
+        Map<Long, AbstractJob> jobMap = (Map<Long, AbstractJob>) Deencapsulation.getField(manager, "jobMap");
+        jobMap.put(1L, streamingJob);
+        jobMap.put(2L, batchJob);
+
+        // Cancelling the batch job must not be blocked by the unrelated streaming job in jobMap.
+        manager.cancelTaskById("batch_job", 100L);
+        Mockito.verify(batchJob).cancelTaskById(100L);
+
+        // Cancelling the streaming job itself still rejected.
+        try {
+            manager.cancelTaskById("streaming_job", 100L);
+            Assert.fail("expected JobException for streaming job");
+        } catch (JobException e) {
+            Assert.assertTrue(e.getMessage().contains("streaming job not support"));
+        }
+    }
 }