apache · taojintianxia · Nov 23, 2023 · Nov 23, 2023
diff --git a/...ardingsphere/data/pipeline/common/registrycenter/repository/PipelineGovernanceFacade.java b/...ardingsphere/data/pipeline/common/registrycenter/repository/PipelineGovernanceFacade.java
@@ -20,6 +20,9 @@
 import lombok.AccessLevel;
 import lombok.Getter;
 import org.apache.shardingsphere.data.pipeline.common.metadata.node.PipelineNodePath;
+import org.apache.shardingsphere.data.pipeline.common.registrycenter.repository.item.PipelineJobItemFacade;
+import org.apache.shardingsphere.data.pipeline.common.registrycenter.repository.job.PipelineJobFacade;
+import org.apache.shardingsphere.data.pipeline.common.registrycenter.repository.metadata.PipelineMetaDataFacade;
 import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
 import org.apache.shardingsphere.mode.repository.cluster.listener.DataChangedEventListener;
 
@@ -32,32 +35,17 @@ public final class PipelineGovernanceFacade {
     @Getter(AccessLevel.NONE)
     private final ClusterPersistRepository repository;
 
-    private final PipelineJobConfigurationGovernanceRepository jobConfigurationGovernanceRepository;
+    private final PipelineJobFacade jobFacade;
 
-    private final PipelineJobOffsetGovernanceRepository jobOffsetGovernanceRepository;
+    private final PipelineJobItemFacade jobItemFacade;
 
-    private final PipelineJobItemProcessGovernanceRepository jobItemProcessGovernanceRepository;
-
-    private final PipelineJobItemErrorMessageGovernanceRepository jobItemErrorMessageGovernanceRepository;
-
-    private final PipelineJobCheckGovernanceRepository jobCheckGovernanceRepository;
-
-    private final PipelineJobGovernanceRepository jobGovernanceRepository;
-
-    private final PipelineMetaDataDataSourceGovernanceRepository metaDataDataSourceGovernanceRepository;
-
-    private final PipelineMetaDataProcessConfigurationGovernanceRepository metaDataProcessConfigurationGovernanceRepository;
+    private final PipelineMetaDataFacade metaDataFacade;
 
     public PipelineGovernanceFacade(final ClusterPersistRepository repository) {
         this.repository = repository;
-        jobConfigurationGovernanceRepository = new PipelineJobConfigurationGovernanceRepository(repository);
-        jobOffsetGovernanceRepository = new PipelineJobOffsetGovernanceRepository(repository);
-        jobItemProcessGovernanceRepository = new PipelineJobItemProcessGovernanceRepository(repository);
-        jobItemErrorMessageGovernanceRepository = new PipelineJobItemErrorMessageGovernanceRepository(repository);
-        jobCheckGovernanceRepository = new PipelineJobCheckGovernanceRepository(repository);
-        jobGovernanceRepository = new PipelineJobGovernanceRepository(repository);
-        metaDataDataSourceGovernanceRepository = new PipelineMetaDataDataSourceGovernanceRepository(repository);
-        metaDataProcessConfigurationGovernanceRepository = new PipelineMetaDataProcessConfigurationGovernanceRepository(repository);
+        jobFacade = new PipelineJobFacade(repository);
+        jobItemFacade = new PipelineJobItemFacade(repository);
+        metaDataFacade = new PipelineMetaDataFacade(repository);
     }
 
     /**

diff --git a/...ItemErrorMessageGovernanceRepository.java → ...ItemErrorMessageGovernanceRepository.java b/...ItemErrorMessageGovernanceRepository.java → ...ItemErrorMessageGovernanceRepository.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.shardingsphere.data.pipeline.common.registrycenter.repository;
+package org.apache.shardingsphere.data.pipeline.common.registrycenter.repository.item;
 
 import lombok.RequiredArgsConstructor;
 import org.apache.shardingsphere.data.pipeline.common.metadata.node.PipelineMetaDataNode;

diff --git a/...dingsphere/data/pipeline/common/registrycenter/repository/item/PipelineJobItemFacade.java b/...dingsphere/data/pipeline/common/registrycenter/repository/item/PipelineJobItemFacade.java
@@ -0,0 +1,37 @@
+/*
+ * 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.common.registrycenter.repository.item;
+
+import lombok.Getter;
+import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
+
+/**
+ * Pipeline job item facade.
+ */
+@Getter
+public final class PipelineJobItemFacade {
+
+    private final PipelineJobItemProcessGovernanceRepository process;
+
+    private final PipelineJobItemErrorMessageGovernanceRepository errorMessage;
+
+    public PipelineJobItemFacade(final ClusterPersistRepository repository) {
+        process = new PipelineJobItemProcessGovernanceRepository(repository);
+        errorMessage = new PipelineJobItemErrorMessageGovernanceRepository(repository);
+    }
+}
diff --git a/...neJobItemProcessGovernanceRepository.java → ...neJobItemProcessGovernanceRepository.java b/...neJobItemProcessGovernanceRepository.java → ...neJobItemProcessGovernanceRepository.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.shardingsphere.data.pipeline.common.registrycenter.repository;
+package org.apache.shardingsphere.data.pipeline.common.registrycenter.repository.item;
 
 import com.google.common.base.Strings;
 import lombok.RequiredArgsConstructor;

diff --git a/...PipelineJobCheckGovernanceRepository.java → ...PipelineJobCheckGovernanceRepository.java b/...PipelineJobCheckGovernanceRepository.java → ...PipelineJobCheckGovernanceRepository.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.shardingsphere.data.pipeline.common.registrycenter.repository;
+package org.apache.shardingsphere.data.pipeline.common.registrycenter.repository.job;
 
 import com.google.common.base.Strings;
 import lombok.RequiredArgsConstructor;

diff --git a/...JobConfigurationGovernanceRepository.java → ...JobConfigurationGovernanceRepository.java b/...JobConfigurationGovernanceRepository.java → ...JobConfigurationGovernanceRepository.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.shardingsphere.data.pipeline.common.registrycenter.repository;
+package org.apache.shardingsphere.data.pipeline.common.registrycenter.repository.job;
 
 import lombok.RequiredArgsConstructor;
 import org.apache.shardingsphere.data.pipeline.common.metadata.node.PipelineMetaDataNode;

diff --git a/.../shardingsphere/data/pipeline/common/registrycenter/repository/job/PipelineJobFacade.java b/.../shardingsphere/data/pipeline/common/registrycenter/repository/job/PipelineJobFacade.java
@@ -0,0 +1,43 @@
+/*
+ * 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.common.registrycenter.repository.job;
+
+import lombok.Getter;
+import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
+
+/**
+ * Pipeline job facade.
+ */
+@Getter
+public final class PipelineJobFacade {
+
+    private final PipelineJobGovernanceRepository job;
+
+    private final PipelineJobConfigurationGovernanceRepository configuration;
+
+    private final PipelineJobOffsetGovernanceRepository offset;
+
+    private final PipelineJobCheckGovernanceRepository check;
+
+    public PipelineJobFacade(final ClusterPersistRepository repository) {
+        job = new PipelineJobGovernanceRepository(repository);
+        configuration = new PipelineJobConfigurationGovernanceRepository(repository);
+        offset = new PipelineJobOffsetGovernanceRepository(repository);
+        check = new PipelineJobCheckGovernanceRepository(repository);
+    }
+}
diff --git a/...tory/PipelineJobGovernanceRepository.java → .../job/PipelineJobGovernanceRepository.java b/...tory/PipelineJobGovernanceRepository.java → .../job/PipelineJobGovernanceRepository.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.shardingsphere.data.pipeline.common.registrycenter.repository;
+package org.apache.shardingsphere.data.pipeline.common.registrycenter.repository.job;
 
 import lombok.RequiredArgsConstructor;
 import org.apache.shardingsphere.data.pipeline.common.job.PipelineJob;

diff --git a/...ipelineJobOffsetGovernanceRepository.java → ...ipelineJobOffsetGovernanceRepository.java b/...ipelineJobOffsetGovernanceRepository.java → ...ipelineJobOffsetGovernanceRepository.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.shardingsphere.data.pipeline.common.registrycenter.repository;
+package org.apache.shardingsphere.data.pipeline.common.registrycenter.repository.job;
 
 import com.google.common.base.Strings;
 import lombok.RequiredArgsConstructor;

diff --git a/...taDataDataSourceGovernanceRepository.java → ...taDataDataSourceGovernanceRepository.java b/...taDataDataSourceGovernanceRepository.java → ...taDataDataSourceGovernanceRepository.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.shardingsphere.data.pipeline.common.registrycenter.repository;
+package org.apache.shardingsphere.data.pipeline.common.registrycenter.repository.metadata;
 
 import lombok.RequiredArgsConstructor;
 import org.apache.shardingsphere.data.pipeline.common.metadata.node.PipelineMetaDataNode;

diff --git a/...phere/data/pipeline/common/registrycenter/repository/metadata/PipelineMetaDataFacade.java b/...phere/data/pipeline/common/registrycenter/repository/metadata/PipelineMetaDataFacade.java
@@ -0,0 +1,37 @@
+/*
+ * 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.common.registrycenter.repository.metadata;
+
+import lombok.Getter;
+import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
+
+/**
+ * Pipeline job item facade.
+ */
+@Getter
+public final class PipelineMetaDataFacade {
+
+    private final PipelineMetaDataDataSourceGovernanceRepository dataSource;
+
+    private final PipelineMetaDataProcessConfigurationGovernanceRepository processConfiguration;
+
+    public PipelineMetaDataFacade(final ClusterPersistRepository repository) {
+        dataSource = new PipelineMetaDataDataSourceGovernanceRepository(repository);
+        processConfiguration = new PipelineMetaDataProcessConfigurationGovernanceRepository(repository);
+    }
+}
diff --git a/...essConfigurationGovernanceRepository.java → ...essConfigurationGovernanceRepository.java b/...essConfigurationGovernanceRepository.java → ...essConfigurationGovernanceRepository.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.shardingsphere.data.pipeline.common.registrycenter.repository;
+package org.apache.shardingsphere.data.pipeline.common.registrycenter.repository.metadata;
 
 import lombok.RequiredArgsConstructor;
 import org.apache.shardingsphere.data.pipeline.common.metadata.node.PipelineMetaDataNode;

diff --git a/...ache/shardingsphere/data/pipeline/core/job/service/PipelineJobIteErrorMessageManager.java b/...ache/shardingsphere/data/pipeline/core/job/service/PipelineJobIteErrorMessageManager.java
@@ -46,7 +46,7 @@ public PipelineJobIteErrorMessageManager(final String jobId, final int shardingI
      * @return map, key is sharding item, value is error message
      */
     public String getErrorMessage() {
-        return Optional.ofNullable(governanceFacade.getJobItemErrorMessageGovernanceRepository().load(jobId, shardingItem)).orElse("");
+        return Optional.ofNullable(governanceFacade.getJobItemFacade().getErrorMessage().load(jobId, shardingItem)).orElse("");
     }
 
     /**
@@ -55,7 +55,7 @@ public String getErrorMessage() {
      * @param error error
      */
     public void updateErrorMessage(final Object error) {
-        governanceFacade.getJobItemErrorMessageGovernanceRepository().update(jobId, shardingItem, null == error ? "" : buildErrorMessage(error));
+        governanceFacade.getJobItemFacade().getErrorMessage().update(jobId, shardingItem, null == error ? "" : buildErrorMessage(error));
     }
 
     private String buildErrorMessage(final Object error) {
@@ -66,6 +66,6 @@ private String buildErrorMessage(final Object error) {
      * Clean job item error message.
      */
     public void cleanErrorMessage() {
-        governanceFacade.getJobItemErrorMessageGovernanceRepository().update(jobId, shardingItem, "");
+        governanceFacade.getJobItemFacade().getErrorMessage().update(jobId, shardingItem, "");
     }
 }
diff --git a/...java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobItemManager.java b/...java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobItemManager.java
@@ -55,7 +55,7 @@ public void updateStatus(final String jobId, final int shardingItem, final JobSt
         }
         jobItemProgress.get().setStatus(status);
         PipelineAPIFactory.getPipelineGovernanceFacade(PipelineJobIdUtils.parseContextKey(jobId))
-                .getJobItemProcessGovernanceRepository().update(jobId, shardingItem, YamlEngine.marshal(swapper.swapToYamlConfiguration(jobItemProgress.get())));
+                .getJobItemFacade().getProcess().update(jobId, shardingItem, YamlEngine.marshal(swapper.swapToYamlConfiguration(jobItemProgress.get())));
     }
 
     /**
@@ -66,7 +66,7 @@ public void updateStatus(final String jobId, final int shardingItem, final JobSt
      * @return job item progress
      */
     public Optional<T> getProgress(final String jobId, final int shardingItem) {
-        return PipelineAPIFactory.getPipelineGovernanceFacade(PipelineJobIdUtils.parseContextKey(jobId)).getJobItemProcessGovernanceRepository().load(jobId, shardingItem)
+        return PipelineAPIFactory.getPipelineGovernanceFacade(PipelineJobIdUtils.parseContextKey(jobId)).getJobItemFacade().getProcess().load(jobId, shardingItem)
                 .map(optional -> swapper.swapToObject(YamlEngine.unmarshal(optional, swapper.getYamlProgressClass(), true)));
     }
 
@@ -77,7 +77,7 @@ public Optional<T> getProgress(final String jobId, final int shardingItem) {
      */
     public void persistProgress(final PipelineJobItemContext jobItemContext) {
         PipelineAPIFactory.getPipelineGovernanceFacade(PipelineJobIdUtils.parseContextKey(jobItemContext.getJobId()))
-                .getJobItemProcessGovernanceRepository().persist(jobItemContext.getJobId(), jobItemContext.getShardingItem(), convertProgressYamlContent(jobItemContext));
+                .getJobItemFacade().getProcess().persist(jobItemContext.getJobId(), jobItemContext.getShardingItem(), convertProgressYamlContent(jobItemContext));
     }
 
     /**
@@ -87,7 +87,7 @@ public void persistProgress(final PipelineJobItemContext jobItemContext) {
      */
     public void updateProgress(final PipelineJobItemContext jobItemContext) {
         PipelineAPIFactory.getPipelineGovernanceFacade(PipelineJobIdUtils.parseContextKey(jobItemContext.getJobId()))
-                .getJobItemProcessGovernanceRepository().update(jobItemContext.getJobId(), jobItemContext.getShardingItem(), convertProgressYamlContent(jobItemContext));
+                .getJobItemFacade().getProcess().update(jobItemContext.getJobId(), jobItemContext.getShardingItem(), convertProgressYamlContent(jobItemContext));
     }
 
     @SuppressWarnings("unchecked")

diff --git a/...ain/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManager.java b/...ain/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManager.java
@@ -75,12 +75,12 @@ public Optional<String> start(final PipelineJobConfiguration jobConfig) {
         String jobId = jobConfig.getJobId();
         ShardingSpherePreconditions.checkState(0 != jobConfig.getJobShardingCount(), () -> new PipelineJobCreationWithInvalidShardingCountException(jobId));
         PipelineGovernanceFacade governanceFacade = PipelineAPIFactory.getPipelineGovernanceFacade(PipelineJobIdUtils.parseContextKey(jobId));
-        if (governanceFacade.getJobConfigurationGovernanceRepository().isExisted(jobId)) {
+        if (governanceFacade.getJobFacade().getConfiguration().isExisted(jobId)) {
             log.warn("jobId already exists in registry center, ignore, job id is `{}`", jobId);
             return Optional.of(jobId);
         }
-        governanceFacade.getJobGovernanceRepository().create(jobId, jobAPI.getJobClass());
-        governanceFacade.getJobConfigurationGovernanceRepository().persist(jobId, jobConfig.convertToJobConfigurationPOJO());
+        governanceFacade.getJobFacade().getJob().create(jobId, jobAPI.getJobClass());
+        governanceFacade.getJobFacade().getConfiguration().persist(jobId, jobConfig.convertToJobConfigurationPOJO());
         return Optional.of(jobId);
     }
 
@@ -119,7 +119,7 @@ private void startCurrentDisabledJob(final String jobId) {
     }
 
     private void startNextDisabledJob(final String jobId, final String toBeStartDisabledNextJobType) {
-        PipelineAPIFactory.getPipelineGovernanceFacade(PipelineJobIdUtils.parseContextKey(jobId)).getJobCheckGovernanceRepository().getLatestCheckJobId(jobId).ifPresent(optional -> {
+        PipelineAPIFactory.getPipelineGovernanceFacade(PipelineJobIdUtils.parseContextKey(jobId)).getJobFacade().getCheck().getLatestCheckJobId(jobId).ifPresent(optional -> {
             try {
                 new PipelineJobManager(TypedSPILoader.getService(PipelineJobAPI.class, toBeStartDisabledNextJobType)).startDisabledJob(optional);
                 // CHECKSTYLE:OFF
@@ -141,7 +141,7 @@ public void stop(final String jobId) {
     }
 
     private void stopPreviousJob(final String jobId, final String toBeStoppedPreviousJobType) {
-        PipelineAPIFactory.getPipelineGovernanceFacade(PipelineJobIdUtils.parseContextKey(jobId)).getJobCheckGovernanceRepository().getLatestCheckJobId(jobId).ifPresent(optional -> {
+        PipelineAPIFactory.getPipelineGovernanceFacade(PipelineJobIdUtils.parseContextKey(jobId)).getJobFacade().getCheck().getLatestCheckJobId(jobId).ifPresent(optional -> {
             try {
                 new PipelineJobManager(TypedSPILoader.getService(PipelineJobAPI.class, toBeStoppedPreviousJobType)).stop(optional);
                 // CHECKSTYLE:OFF
@@ -176,7 +176,7 @@ private void stopCurrentJob(final String jobId) {
     public void drop(final String jobId) {
         PipelineContextKey contextKey = PipelineJobIdUtils.parseContextKey(jobId);
         PipelineAPIFactory.getJobOperateAPI(contextKey).remove(String.valueOf(jobId), null);
-        PipelineAPIFactory.getPipelineGovernanceFacade(contextKey).getJobGovernanceRepository().delete(jobId);
+        PipelineAPIFactory.getPipelineGovernanceFacade(contextKey).getJobFacade().getJob().delete(jobId);
     }
 
     /**

diff --git a/...g/apache/shardingsphere/data/pipeline/core/metadata/PipelineDataSourcePersistService.java b/...g/apache/shardingsphere/data/pipeline/core/metadata/PipelineDataSourcePersistService.java
@@ -39,7 +39,7 @@ public final class PipelineDataSourcePersistService implements PipelineMetaDataP
     @Override
     @SuppressWarnings("unchecked")
     public Map<String, DataSourcePoolProperties> load(final PipelineContextKey contextKey, final String jobType) {
-        String dataSourcesProps = PipelineAPIFactory.getPipelineGovernanceFacade(contextKey).getMetaDataDataSourceGovernanceRepository().load(jobType);
+        String dataSourcesProps = PipelineAPIFactory.getPipelineGovernanceFacade(contextKey).getMetaDataFacade().getDataSource().load(jobType);
         if (Strings.isNullOrEmpty(dataSourcesProps)) {
             return Collections.emptyMap();
         }
@@ -55,6 +55,6 @@ public void persist(final PipelineContextKey contextKey, final String jobType, f
         for (Entry<String, DataSourcePoolProperties> entry : propsMap.entrySet()) {
             dataSourceMap.put(entry.getKey(), swapper.swapToMap(entry.getValue()));
         }
-        PipelineAPIFactory.getPipelineGovernanceFacade(contextKey).getMetaDataDataSourceGovernanceRepository().persist(jobType, YamlEngine.marshal(dataSourceMap));
+        PipelineAPIFactory.getPipelineGovernanceFacade(contextKey).getMetaDataFacade().getDataSource().persist(jobType, YamlEngine.marshal(dataSourceMap));
     }
 }