Refactor PipelineContextKey (#28996)

kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/context/PipelineContextKey.java

@@ -17,7 +17,6 @@
 
 package org.apache.shardingsphere.data.pipeline.common.context;
 
-import lombok.AccessLevel;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 import org.apache.shardingsphere.infra.instance.metadata.InstanceType;
@@ -27,42 +26,16 @@
 /**
  * Pipeline context key.
  */
-@RequiredArgsConstructor(access = AccessLevel.PRIVATE)
+@RequiredArgsConstructor
 @Getter
 public final class PipelineContextKey {
 
     private final String databaseName;
 
     private final InstanceType instanceType;
 
-    /**
-     * Build context key.
-     *
-     * @param databaseName database name
-     * @param instanceType instance type
-     * @return context key
-     */
-    public static PipelineContextKey build(final String databaseName, final InstanceType instanceType) {
-        return new PipelineContextKey(databaseName, instanceType);
-    }
-
-    /**
-     * Build context key for proxy.
-     *
-     * @return context key
-     */
-    public static PipelineContextKey buildForProxy() {
-        return new PipelineContextKey("", InstanceType.PROXY);
-    }
-
-    /**
-     * Build context key for proxy.
-     *
-     * @param databaseName database name
-     * @return context key
-     */
-    public static PipelineContextKey buildForProxy(final String databaseName) {
-        return new PipelineContextKey(databaseName, InstanceType.PROXY);
+    public PipelineContextKey(final InstanceType instanceType) {
+        this("", instanceType);
     }
 
     @Override
@@ -78,7 +51,7 @@ public boolean equals(final Object o) {
     }
 
     private String filterDatabaseName(final PipelineContextKey contextKey) {
-        return contextKey.getInstanceType() == InstanceType.PROXY ? "" : contextKey.getDatabaseName();
+        return InstanceType.PROXY == contextKey.getInstanceType() ? "" : contextKey.getDatabaseName();
     }
 
     @Override

kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/context/PipelineContextManager.java

@@ -19,6 +19,7 @@
 
 import lombok.AccessLevel;
 import lombok.NoArgsConstructor;
+import org.apache.shardingsphere.infra.instance.metadata.InstanceType;
 
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -47,7 +48,7 @@ public static PipelineContext getContext(final PipelineContextKey key) {
      * @return context
      */
     public static PipelineContext getProxyContext() {
-        return CONTEXT_MAP.get(PipelineContextKey.buildForProxy());
+        return CONTEXT_MAP.get(new PipelineContextKey(InstanceType.PROXY));
     }
 
     /**

kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/listener/PipelineContextManagerLifecycleListener.java

@@ -47,14 +47,14 @@ public void onInitialized(final String databaseName, final ContextManager contex
         if (DefaultDatabase.LOGIC_NAME.equals(databaseName)) {
             return;
         }
-        PipelineContextKey contextKey = PipelineContextKey.build(databaseName, contextManager.getInstanceContext().getInstance().getMetaData().getType());
+        PipelineContextKey contextKey = new PipelineContextKey(databaseName, contextManager.getInstanceContext().getInstance().getMetaData().getType());
         PipelineContextManager.putContext(contextKey, new PipelineContext(modeConfig, contextManager));
         PipelineMetaDataNodeWatcher.getInstance(contextKey);
         ElasticJobServiceLoader.registerTypedService(ElasticJobListener.class);
     }
 
     @Override
     public void onDestroyed(final String databaseName, final InstanceType instanceType) {
-        PipelineContextManager.removeContext(PipelineContextKey.build(databaseName, instanceType));
+        PipelineContextManager.removeContext(new PipelineContextKey(databaseName, instanceType));
     }
 }

kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/PipelineJobIdUtils.java

@@ -84,6 +84,6 @@ public static PipelineContextKey parseContextKey(final String jobId) {
         char instanceType = jobId.charAt(5);
         short databaseNameLength = Shorts.fromByteArray(Hex.decodeHex(jobId.substring(6, 10)));
         String databaseName = new String(Hex.decodeHex(jobId.substring(10, 10 + databaseNameLength)), StandardCharsets.UTF_8);
-        return PipelineContextKey.build(databaseName, InstanceTypeUtils.decode(instanceType));
+        return new PipelineContextKey(databaseName, InstanceTypeUtils.decode(instanceType));
     }
 }

kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/common/context/PipelineContextKeyTest.java

@@ -28,16 +28,16 @@ class PipelineContextKeyTest {
 
     @Test
     void assertHashCodeEqualsForProxyMode() {
-        PipelineContextKey contextKey1 = PipelineContextKey.build(null, InstanceType.PROXY);
-        PipelineContextKey contextKey2 = PipelineContextKey.build("sharding_db", InstanceType.PROXY);
+        PipelineContextKey contextKey1 = new PipelineContextKey(null, InstanceType.PROXY);
+        PipelineContextKey contextKey2 = new PipelineContextKey("sharding_db", InstanceType.PROXY);
         assertThat(contextKey1.hashCode(), is(contextKey2.hashCode()));
         assertThat(contextKey1, is(contextKey2));
     }
 
     @Test
     void assertHashCodeEqualsForJdbcMode() {
-        PipelineContextKey contextKey1 = PipelineContextKey.build("logic_db", InstanceType.JDBC);
-        PipelineContextKey contextKey2 = PipelineContextKey.build("sharding_db", InstanceType.JDBC);
+        PipelineContextKey contextKey1 = new PipelineContextKey("logic_db", InstanceType.JDBC);
+        PipelineContextKey contextKey2 = new PipelineContextKey("sharding_db", InstanceType.JDBC);
         assertThat(contextKey1.hashCode(), not(contextKey2.hashCode()));
         assertThat(contextKey1, not(contextKey2));
     }

kernel/data-pipeline/distsql/handler/src/main/java/org/apache/shardingsphere/cdc/distsql/handler/query/ShowStreamingListExecutor.java

@@ -22,6 +22,7 @@
 import org.apache.shardingsphere.data.pipeline.common.context.PipelineContextKey;
 import org.apache.shardingsphere.data.pipeline.common.pojo.TableBasedPipelineJobInfo;
 import org.apache.shardingsphere.distsql.handler.ral.query.QueryableRALExecutor;
+import org.apache.shardingsphere.infra.instance.metadata.InstanceType;
 import org.apache.shardingsphere.infra.merge.result.impl.local.LocalDataQueryResultRow;
 
 import java.util.Arrays;
@@ -38,7 +39,7 @@ public final class ShowStreamingListExecutor implements QueryableRALExecutor<Sho
 
     @Override
     public Collection<LocalDataQueryResultRow> getRows(final ShowStreamingListStatement sqlStatement) {
-        return jobAPI.list(PipelineContextKey.buildForProxy()).stream().map(each -> new LocalDataQueryResultRow(each.getJobMetaData().getJobId(),
+        return jobAPI.list(new PipelineContextKey(InstanceType.PROXY)).stream().map(each -> new LocalDataQueryResultRow(each.getJobMetaData().getJobId(),
                 ((TableBasedPipelineJobInfo) each).getDatabaseName(), ((TableBasedPipelineJobInfo) each).getTable(),
                 each.getJobMetaData().getJobItemCount(), each.getJobMetaData().isActive() ? Boolean.TRUE.toString() : Boolean.FALSE.toString(),
                 each.getJobMetaData().getCreateTime(), Optional.ofNullable(each.getJobMetaData().getStopTime()).orElse(""))).collect(Collectors.toList());

kernel/data-pipeline/distsql/handler/src/main/java/org/apache/shardingsphere/migration/distsql/handler/query/ShowMigrationListExecutor.java

@@ -21,6 +21,7 @@
 import org.apache.shardingsphere.data.pipeline.common.pojo.TableBasedPipelineJobInfo;
 import org.apache.shardingsphere.data.pipeline.scenario.migration.api.impl.MigrationJobAPI;
 import org.apache.shardingsphere.distsql.handler.ral.query.QueryableRALExecutor;
+import org.apache.shardingsphere.infra.instance.metadata.InstanceType;
 import org.apache.shardingsphere.infra.merge.result.impl.local.LocalDataQueryResultRow;
 import org.apache.shardingsphere.migration.distsql.statement.ShowMigrationListStatement;
 
@@ -37,7 +38,7 @@ public final class ShowMigrationListExecutor implements QueryableRALExecutor<Sho
 
     @Override
     public Collection<LocalDataQueryResultRow> getRows(final ShowMigrationListStatement sqlStatement) {
-        return jobAPI.list(PipelineContextKey.buildForProxy()).stream().map(each -> new LocalDataQueryResultRow(each.getJobMetaData().getJobId(),
+        return jobAPI.list(new PipelineContextKey(InstanceType.PROXY)).stream().map(each -> new LocalDataQueryResultRow(each.getJobMetaData().getJobId(),
                 ((TableBasedPipelineJobInfo) each).getTable(), each.getJobMetaData().getJobItemCount(),
                 each.getJobMetaData().isActive() ? Boolean.TRUE.toString() : Boolean.FALSE.toString(),
                 each.getJobMetaData().getCreateTime(), each.getJobMetaData().getStopTime())).collect(Collectors.toList());

kernel/data-pipeline/distsql/handler/src/main/java/org/apache/shardingsphere/migration/distsql/handler/query/ShowMigrationSourceStorageUnitsExecutor.java

@@ -20,6 +20,7 @@
 import org.apache.shardingsphere.data.pipeline.common.context.PipelineContextKey;
 import org.apache.shardingsphere.data.pipeline.scenario.migration.api.impl.MigrationJobAPI;
 import org.apache.shardingsphere.distsql.handler.ral.query.QueryableRALExecutor;
+import org.apache.shardingsphere.infra.instance.metadata.InstanceType;
 import org.apache.shardingsphere.infra.merge.result.impl.local.LocalDataQueryResultRow;
 import org.apache.shardingsphere.migration.distsql.statement.ShowMigrationSourceStorageUnitsStatement;
 
@@ -38,7 +39,7 @@ public final class ShowMigrationSourceStorageUnitsExecutor implements QueryableR
 
     @Override
     public Collection<LocalDataQueryResultRow> getRows(final ShowMigrationSourceStorageUnitsStatement sqlStatement) {
-        Iterator<Collection<Object>> data = jobAPI.listMigrationSourceResources(PipelineContextKey.buildForProxy()).iterator();
+        Iterator<Collection<Object>> data = jobAPI.listMigrationSourceResources(new PipelineContextKey(InstanceType.PROXY)).iterator();
         Collection<LocalDataQueryResultRow> result = new LinkedList<>();
         while (data.hasNext()) {
             result.add(new LocalDataQueryResultRow((List<Object>) data.next()));

kernel/data-pipeline/distsql/handler/src/main/java/org/apache/shardingsphere/migration/distsql/handler/update/MigrateTableUpdater.java

@@ -23,6 +23,7 @@
 import org.apache.shardingsphere.data.pipeline.scenario.migration.api.impl.MigrationJobAPI;
 import org.apache.shardingsphere.distsql.handler.ral.update.RALUpdater;
 import org.apache.shardingsphere.infra.exception.core.ShardingSpherePreconditions;
+import org.apache.shardingsphere.infra.instance.metadata.InstanceType;
 import org.apache.shardingsphere.migration.distsql.statement.MigrateTableStatement;
 
 /**
@@ -37,7 +38,7 @@ public final class MigrateTableUpdater implements RALUpdater<MigrateTableStateme
     public void executeUpdate(final String databaseName, final MigrateTableStatement sqlStatement) {
         String targetDatabaseName = null == sqlStatement.getTargetDatabaseName() ? databaseName : sqlStatement.getTargetDatabaseName();
         ShardingSpherePreconditions.checkNotNull(targetDatabaseName, MissingRequiredTargetDatabaseException::new);
-        jobAPI.createJobAndStart(PipelineContextKey.buildForProxy(), new MigrateTableStatement(sqlStatement.getSourceTargetEntries(), targetDatabaseName));
+        jobAPI.createJobAndStart(new PipelineContextKey(InstanceType.PROXY), new MigrateTableStatement(sqlStatement.getSourceTargetEntries(), targetDatabaseName));
     }
 
     @Override

kernel/data-pipeline/distsql/handler/src/main/java/org/apache/shardingsphere/migration/distsql/handler/update/RegisterMigrationSourceStorageUnitUpdater.java

@@ -30,6 +30,7 @@
 import org.apache.shardingsphere.infra.datasource.pool.props.domain.DataSourcePoolProperties;
 import org.apache.shardingsphere.infra.exception.core.ShardingSpherePreconditions;
 import org.apache.shardingsphere.infra.exception.core.external.sql.type.generic.UnsupportedSQLOperationException;
+import org.apache.shardingsphere.infra.instance.metadata.InstanceType;
 import org.apache.shardingsphere.migration.distsql.statement.RegisterMigrationSourceStorageUnitStatement;
 
 import java.util.ArrayList;
@@ -54,7 +55,7 @@ public void executeUpdate(final String databaseName, final RegisterMigrationSour
         DatabaseType databaseType = DatabaseTypeFactory.get(urlBasedDataSourceSegment.getUrl());
         Map<String, DataSourcePoolProperties> propsMap = DataSourceSegmentsConverter.convert(databaseType, dataSources);
         validateHandler.validate(propsMap);
-        jobAPI.addMigrationSourceResources(PipelineContextKey.buildForProxy(), propsMap);
+        jobAPI.addMigrationSourceResources(new PipelineContextKey(InstanceType.PROXY), propsMap);
     }
 
     @Override

kernel/data-pipeline/distsql/handler/src/main/java/org/apache/shardingsphere/migration/distsql/handler/update/UnregisterMigrationSourceStorageUnitUpdater.java

@@ -20,6 +20,7 @@
 import org.apache.shardingsphere.data.pipeline.common.context.PipelineContextKey;
 import org.apache.shardingsphere.data.pipeline.scenario.migration.api.impl.MigrationJobAPI;
 import org.apache.shardingsphere.distsql.handler.ral.update.RALUpdater;
+import org.apache.shardingsphere.infra.instance.metadata.InstanceType;
 import org.apache.shardingsphere.migration.distsql.statement.UnregisterMigrationSourceStorageUnitStatement;
 
 /**
@@ -31,7 +32,7 @@ public final class UnregisterMigrationSourceStorageUnitUpdater implements RALUpd
 
     @Override
     public void executeUpdate(final String databaseName, final UnregisterMigrationSourceStorageUnitStatement sqlStatement) {
-        jobAPI.dropMigrationSourceResources(PipelineContextKey.buildForProxy(), sqlStatement.getNames());
+        jobAPI.dropMigrationSourceResources(new PipelineContextKey(InstanceType.PROXY), sqlStatement.getNames());
     }
 
     @Override

kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/api/impl/CDCJobAPI.java

@@ -78,6 +78,7 @@
 import org.apache.shardingsphere.elasticjob.infra.pojo.JobConfigurationPOJO;
 import org.apache.shardingsphere.elasticjob.lite.api.bootstrap.impl.OneOffJobBootstrap;
 import org.apache.shardingsphere.infra.exception.core.ShardingSpherePreconditions;
+import org.apache.shardingsphere.infra.instance.metadata.InstanceType;
 import org.apache.shardingsphere.infra.metadata.database.ShardingSphereDatabase;
 import org.apache.shardingsphere.infra.metadata.database.resource.unit.StorageUnit;
 import org.apache.shardingsphere.infra.util.yaml.YamlEngine;
@@ -117,7 +118,7 @@ public final class CDCJobAPI extends AbstractInventoryIncrementalJobAPIImpl {
      * @return job id
      */
     public String createJob(final StreamDataParameter param, final CDCSinkType sinkType, final Properties sinkProps) {
-        PipelineContextKey contextKey = PipelineContextKey.buildForProxy(param.getDatabaseName());
+        PipelineContextKey contextKey = new PipelineContextKey(param.getDatabaseName(), InstanceType.PROXY);
         YamlCDCJobConfiguration yamlJobConfig = getYamlCDCJobConfiguration(param, sinkType, sinkProps, contextKey);
         extendYamlJobConfiguration(contextKey, yamlJobConfig);
         CDCJobConfiguration jobConfig = new YamlCDCJobConfigurationSwapper().swapToObject(yamlJobConfig);

kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/core/job/CDCJobIdTest.java

@@ -34,7 +34,7 @@ class CDCJobIdTest {
 
     @Test
     void assertParseJobType() {
-        PipelineContextKey contextKey = PipelineContextKey.build("sharding_db", InstanceType.PROXY);
+        PipelineContextKey contextKey = new PipelineContextKey("sharding_db", InstanceType.PROXY);
         CDCJobId pipelineJobId = new CDCJobId(contextKey, Arrays.asList("test", "t_order"), false, CDCSinkType.SOCKET.name());
         String jobId = PipelineJobIdUtils.marshalJobIdCommonPrefix(pipelineJobId) + "abcd";
         JobType actualJobType = PipelineJobIdUtils.parseJobType(jobId);

kernel/data-pipeline/scenario/migration/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/PipelineJobIdUtilsTest.java

@@ -40,7 +40,7 @@ void assertParse() {
     }
 
     private void assertParse0(final InstanceType instanceType) {
-        PipelineContextKey contextKey = PipelineContextKey.build("sharding_db", instanceType);
+        PipelineContextKey contextKey = new PipelineContextKey("sharding_db", instanceType);
         MigrationJobId pipelineJobId = new MigrationJobId(contextKey, Collections.singletonList("t_order:ds_0.t_order_0,ds_0.t_order_1"));
         String jobId = new MigrationJobAPI().marshalJobId(pipelineJobId);
         assertThat(PipelineJobIdUtils.parseJobType(jobId), instanceOf(MigrationJobType.class));

proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/handler/distsql/ral/queryable/ShowMigrationRuleExecutor.java

@@ -23,6 +23,7 @@
 import org.apache.shardingsphere.data.pipeline.core.job.service.PipelineJobAPI;
 import org.apache.shardingsphere.distsql.handler.ral.query.QueryableRALExecutor;
 import org.apache.shardingsphere.distsql.statement.ral.queryable.ShowMigrationRuleStatement;
+import org.apache.shardingsphere.infra.instance.metadata.InstanceType;
 import org.apache.shardingsphere.infra.merge.result.impl.local.LocalDataQueryResultRow;
 import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
 import org.apache.shardingsphere.infra.util.json.JsonUtils;
@@ -39,7 +40,7 @@ public final class ShowMigrationRuleExecutor implements QueryableRALExecutor<Sho
     @Override
     public Collection<LocalDataQueryResultRow> getRows(final ShowMigrationRuleStatement sqlStatement) {
         PipelineProcessConfiguration processConfig = ((InventoryIncrementalJobAPI) TypedSPILoader.getService(PipelineJobAPI.class, "MIGRATION"))
-                .showProcessConfiguration(PipelineContextKey.buildForProxy());
+                .showProcessConfiguration(new PipelineContextKey(InstanceType.PROXY));
         Collection<LocalDataQueryResultRow> result = new LinkedList<>();
         result.add(new LocalDataQueryResultRow(getString(processConfig.getRead()), getString(processConfig.getWrite()), getString(processConfig.getStreamChannel())));
         return result;

proxy/backend/core/src/main/java/org/apache/shardingsphere/proxy/backend/handler/distsql/ral/updatable/AlterInventoryIncrementalRuleUpdater.java

@@ -23,6 +23,7 @@
 import org.apache.shardingsphere.data.pipeline.core.job.service.PipelineJobAPI;
 import org.apache.shardingsphere.distsql.handler.ral.update.RALUpdater;
 import org.apache.shardingsphere.distsql.statement.ral.updatable.AlterInventoryIncrementalRuleStatement;
+import org.apache.shardingsphere.infra.instance.metadata.InstanceType;
 import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
 import org.apache.shardingsphere.proxy.backend.handler.distsql.ral.updatable.converter.InventoryIncrementalProcessConfigurationSegmentConverter;
 
@@ -35,7 +36,7 @@ public final class AlterInventoryIncrementalRuleUpdater implements RALUpdater<Al
     public void executeUpdate(final String databaseName, final AlterInventoryIncrementalRuleStatement sqlStatement) {
         InventoryIncrementalJobAPI jobAPI = (InventoryIncrementalJobAPI) TypedSPILoader.getService(PipelineJobAPI.class, sqlStatement.getJobTypeName());
         PipelineProcessConfiguration processConfig = InventoryIncrementalProcessConfigurationSegmentConverter.convert(sqlStatement.getProcessConfigSegment());
-        jobAPI.alterProcessConfiguration(PipelineContextKey.buildForProxy(), processConfig);
+        jobAPI.alterProcessConfiguration(new PipelineContextKey(InstanceType.PROXY), processConfig);
     }
 
     @Override

test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/util/JobConfigurationBuilder.java

@@ -78,7 +78,7 @@ public static YamlMigrationJobConfiguration createYamlMigrationJobConfiguration(
         result.setTargetTableSchemaMap(targetTableSchemaMap);
         result.setTablesFirstDataNodes("t_order:ds_0.t_order");
         result.setJobShardingDataNodes(Collections.singletonList("t_order:ds_0.t_order"));
-        PipelineContextKey contextKey = PipelineContextKey.build(RandomStringUtils.randomAlphabetic(32), InstanceType.PROXY);
+        PipelineContextKey contextKey = new PipelineContextKey(RandomStringUtils.randomAlphabetic(32), InstanceType.PROXY);
         result.setJobId(generateMigrationJobId(contextKey, result));
         Map<String, YamlPipelineDataSourceConfiguration> sources = new LinkedHashMap<>();
         String databaseNameSuffix = RandomStringUtils.randomAlphabetic(9);

test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/core/util/PipelineContextUtils.java

@@ -38,6 +38,7 @@
 import org.apache.shardingsphere.driver.api.yaml.YamlShardingSphereDataSourceFactory;
 import org.apache.shardingsphere.driver.jdbc.core.datasource.ShardingSphereDataSource;
 import org.apache.shardingsphere.infra.config.mode.ModeConfiguration;
+import org.apache.shardingsphere.infra.instance.metadata.InstanceType;
 import org.apache.shardingsphere.infra.metadata.database.schema.model.ShardingSphereColumn;
 import org.apache.shardingsphere.infra.metadata.database.schema.model.ShardingSphereTable;
 import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
@@ -65,7 +66,7 @@
  */
 public final class PipelineContextUtils {
 
-    private static final PipelineContextKey CONTEXT_KEY = PipelineContextKey.buildForProxy();
+    private static final PipelineContextKey CONTEXT_KEY = new PipelineContextKey(InstanceType.PROXY);
 
     private static final ExecuteEngine EXECUTE_ENGINE = ExecuteEngine.newCachedThreadInstance(PipelineContextUtils.class.getSimpleName());