This is an automated email from the ASF dual-hosted git repository.
sandynz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git
The following commit(s) were added to refs/heads/master by this push:
new 82202bb7647 pipeline: skip missing job config when listing jobs
(#38834)
82202bb7647 is described below
commit 82202bb7647169fb6561796f458372e0f7235821
Author: Hongsheng Zhong <[email protected]>
AuthorDate: Tue Jun 9 13:34:12 2026 +0800
pipeline: skip missing job config when listing jobs (#38834)
---
.../core/job/service/PipelineJobManager.java | 16 +++-
.../core/job/service/PipelineJobManagerTest.java | 96 ++++++++++++++++++++++
2 files changed, 109 insertions(+), 3 deletions(-)
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManager.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManager.java
index 0979392bcf0..2c07c275bdb 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManager.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManager.java
@@ -21,6 +21,7 @@ import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.data.pipeline.core.context.PipelineContextKey;
import
org.apache.shardingsphere.data.pipeline.core.exception.job.PipelineJobCreationWithInvalidShardingCountException;
+import
org.apache.shardingsphere.data.pipeline.core.exception.job.PipelineJobNotFoundException;
import org.apache.shardingsphere.data.pipeline.core.job.JobStatus;
import org.apache.shardingsphere.data.pipeline.core.job.api.PipelineAPIFactory;
import
org.apache.shardingsphere.data.pipeline.core.job.config.PipelineJobConfiguration;
@@ -174,18 +175,27 @@ public final class PipelineJobManager {
* @param contextKey context key
* @return jobs info
*/
- @SuppressWarnings("unchecked")
public List<PipelineJobInfo> getJobInfos(final PipelineContextKey
contextKey) {
try {
return
PipelineAPIFactory.getJobStatisticsAPI(contextKey).getAllJobsBriefInfo().stream().filter(this::isValidJob)
- .map(each -> new PipelineJobInfo(new
PipelineJobMetaData(PipelineJobIdUtils.getElasticJobConfigurationPOJO(each.getJobName())),
- jobType.getJobTarget(new
PipelineJobConfigurationManager(jobType.getOption()).getJobConfiguration(each.getJobName()))))
+
.map(this::createJobInfo).filter(Optional::isPresent).map(Optional::get)
.collect(Collectors.toList());
} catch (final UnsupportedOperationException ex) {
return Collections.emptyList();
}
}
+ @SuppressWarnings("unchecked")
+ private Optional<PipelineJobInfo> createJobInfo(final JobBriefInfo
jobInfo) {
+ try {
+ JobConfigurationPOJO jobConfigPOJO =
PipelineJobIdUtils.getElasticJobConfigurationPOJO(jobInfo.getJobName());
+ PipelineJobConfiguration jobConfig =
jobType.getOption().getYamlJobConfigurationSwapper().swapToObject(jobConfigPOJO.getJobParameter());
+ return Optional.of(new PipelineJobInfo(new
PipelineJobMetaData(jobConfigPOJO), jobType.getJobTarget(jobConfig)));
+ } catch (final PipelineJobNotFoundException ignored) {
+ return Optional.empty();
+ }
+ }
+
private boolean isValidJob(final JobBriefInfo jobInfo) {
return !jobInfo.getJobName().startsWith("_") &&
jobType.getOption().isTransmissionJob() &&
jobType.getType().equals(PipelineJobIdUtils.parseJobType(jobInfo.getJobName()).getType());
}
diff --git
a/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManagerTest.java
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManagerTest.java
new file mode 100644
index 00000000000..f5aedce9051
--- /dev/null
+++
b/kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManagerTest.java
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.data.pipeline.core.job.service;
+
+import org.apache.shardingsphere.data.pipeline.core.context.PipelineContextKey;
+import
org.apache.shardingsphere.data.pipeline.core.exception.job.PipelineJobNotFoundException;
+import org.apache.shardingsphere.data.pipeline.core.job.api.PipelineAPIFactory;
+import
org.apache.shardingsphere.data.pipeline.core.job.config.PipelineJobConfiguration;
+import
org.apache.shardingsphere.data.pipeline.core.job.config.yaml.swapper.YamlPipelineJobConfigurationSwapper;
+import org.apache.shardingsphere.data.pipeline.core.job.id.PipelineJobIdUtils;
+import org.apache.shardingsphere.data.pipeline.core.job.type.PipelineJobOption;
+import org.apache.shardingsphere.data.pipeline.core.job.type.PipelineJobType;
+import org.apache.shardingsphere.data.pipeline.core.pojo.PipelineJobInfo;
+import org.apache.shardingsphere.data.pipeline.core.pojo.PipelineJobTarget;
+import org.apache.shardingsphere.elasticjob.infra.pojo.JobConfigurationPOJO;
+import
org.apache.shardingsphere.elasticjob.lite.lifecycle.api.JobStatisticsAPI;
+import org.apache.shardingsphere.elasticjob.lite.lifecycle.domain.JobBriefInfo;
+import org.apache.shardingsphere.infra.instance.metadata.InstanceType;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
+
+class PipelineJobManagerTest {
+
+ @Test
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ void assertGetJobInfosSkipsMissingJobConfiguration() {
+ PipelineContextKey contextKey = new
PipelineContextKey(InstanceType.PROXY);
+ JobStatisticsAPI jobStatisticsAPI = mock(JobStatisticsAPI.class);
+
when(jobStatisticsAPI.getAllJobsBriefInfo()).thenReturn(Arrays.asList(createJobBriefInfo("j0101p0000_active"),
createJobBriefInfo("j0101p0000_missing")));
+ PipelineJobType jobType = mock(PipelineJobType.class);
+ PipelineJobType parsedJobType = mock(PipelineJobType.class);
+ PipelineJobOption jobOption = mock(PipelineJobOption.class);
+ when(jobType.getType()).thenReturn("FOO");
+ when(parsedJobType.getType()).thenReturn("FOO");
+ when(jobType.getOption()).thenReturn(jobOption);
+ when(jobOption.isTransmissionJob()).thenReturn(true);
+ YamlPipelineJobConfigurationSwapper swapper =
mock(YamlPipelineJobConfigurationSwapper.class);
+ when(jobOption.getYamlJobConfigurationSwapper()).thenReturn(swapper);
+ PipelineJobConfiguration jobConfig =
mock(PipelineJobConfiguration.class);
+ doReturn(jobConfig).when(swapper).swapToObject("active_param");
+ when(jobType.getJobTarget(jobConfig)).thenReturn(new
PipelineJobTarget("foo_db", "foo_tbl"));
+ JobConfigurationPOJO activeJobConfig = createJobConfiguration();
+ try (
+ MockedStatic<PipelineAPIFactory> apiFactory =
mockStatic(PipelineAPIFactory.class);
+ MockedStatic<PipelineJobIdUtils> jobIdUtils =
mockStatic(PipelineJobIdUtils.class)) {
+ apiFactory.when(() ->
PipelineAPIFactory.getJobStatisticsAPI(contextKey)).thenReturn(jobStatisticsAPI);
+ jobIdUtils.when(() ->
PipelineJobIdUtils.parseJobType("j0101p0000_active")).thenReturn(parsedJobType);
+ jobIdUtils.when(() ->
PipelineJobIdUtils.parseJobType("j0101p0000_missing")).thenReturn(parsedJobType);
+ jobIdUtils.when(() ->
PipelineJobIdUtils.getElasticJobConfigurationPOJO("j0101p0000_active")).thenReturn(activeJobConfig);
+ jobIdUtils.when(() ->
PipelineJobIdUtils.getElasticJobConfigurationPOJO("j0101p0000_missing")).thenThrow(new
PipelineJobNotFoundException("j0101p0000_missing"));
+ List<PipelineJobInfo> actual = new
PipelineJobManager(jobType).getJobInfos(contextKey);
+ assertThat(actual.size(), is(1));
+ assertThat(actual.get(0).getMetaData().getJobId(),
is("j0101p0000_active"));
+ assertThat(actual.get(0).getTarget().getTableName(),
is("foo_tbl"));
+ }
+ }
+
+ private JobBriefInfo createJobBriefInfo(final String jobName) {
+ JobBriefInfo result = new JobBriefInfo();
+ result.setJobName(jobName);
+ return result;
+ }
+
+ private JobConfigurationPOJO createJobConfiguration() {
+ JobConfigurationPOJO result = new JobConfigurationPOJO();
+ result.setJobName("j0101p0000_active");
+ result.setJobParameter("active_param");
+ result.setShardingTotalCount(1);
+ return result;
+ }
+}
