Refactor job type declarations to use generics for improved type safety (apache#37173)

Author: terrymanu

features/sharding/core/src/test/java/org/apache/shardingsphere/sharding/route/engine/condition/generator/impl/ConditionValueInOperatorGeneratorTest.java

@@ -52,6 +52,7 @@ class ConditionValueInOperatorGeneratorTest {
 
     private final TimestampServiceRule timestampServiceRule = new TimestampServiceRule(new TimestampServiceRuleConfiguration("System", new Properties()));
 
+    @SuppressWarnings("UseOfObsoleteDateTimeApi")
     @Test
     void assertNowExpression() {
         ListExpression listExpression = new ListExpression(0, 0);
@@ -71,7 +72,7 @@ void assertNullExpression() {
         InExpression inExpression = new InExpression(0, 0, null, listExpression, false);
         Optional<ShardingConditionValue> shardingConditionValue = generator.generate(inExpression, column, new LinkedList<>(), timestampServiceRule);
         assertTrue(shardingConditionValue.isPresent());
-        assertThat(((ListShardingConditionValue) shardingConditionValue.get()).getValues(), is(Arrays.asList(null, null)));
+        assertThat(((ListShardingConditionValue<?>) shardingConditionValue.get()).getValues(), is(Arrays.asList(null, null)));
         assertTrue(shardingConditionValue.get().getParameterMarkerIndexes().isEmpty());
         assertThat(shardingConditionValue.get().toString(), is("tbl.id in (,)"));
     }
@@ -86,7 +87,7 @@ void assertNullAndCommonExpression() {
         InExpression inExpression = new InExpression(0, 0, null, listExpression, false);
         Optional<ShardingConditionValue> shardingConditionValue = generator.generate(inExpression, column, new LinkedList<>(), timestampServiceRule);
         assertTrue(shardingConditionValue.isPresent());
-        assertThat(((ListShardingConditionValue) shardingConditionValue.get()).getValues(), is(Arrays.asList("test1", null, null, "test2")));
+        assertThat(((ListShardingConditionValue<?>) shardingConditionValue.get()).getValues(), is(Arrays.asList("test1", null, null, "test2")));
         assertTrue(shardingConditionValue.get().getParameterMarkerIndexes().isEmpty());
         assertThat(shardingConditionValue.get().toString(), is("tbl.id in (test1,,,test2)"));
     }

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

@@ -68,7 +68,7 @@ public void execute(final ShardingContext shardingContext) {
             log.info("Job is stopping, ignore.");
             return;
         }
-        PipelineJobType jobType = PipelineJobIdUtils.parseJobType(jobId);
+        PipelineJobType<?> jobType = PipelineJobIdUtils.parseJobType(jobId);
         PipelineContextKey contextKey = PipelineJobIdUtils.parseContextKey(jobId);
         PipelineJobConfiguration jobConfig = jobType.getOption().getYamlJobConfigurationSwapper().swapToObject(shardingContext.getJobParameter());
         PipelineJobItemManager<PipelineJobItemProgress> jobItemManager = new PipelineJobItemManager<>(jobType.getOption().getYamlJobItemProgressSwapper());
@@ -111,7 +111,7 @@ private boolean execute(final PipelineJobItemContext jobItemContext, final Pipel
         return true;
     }
 
-    private TransmissionProcessContext createTransmissionProcessContext(final String jobId, final PipelineJobType jobType, final PipelineContextKey contextKey) {
+    private TransmissionProcessContext createTransmissionProcessContext(final String jobId, final PipelineJobType<?> jobType, final PipelineContextKey contextKey) {
         if (!jobType.getOption().isTransmissionJob()) {
             return null;
         }

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

@@ -36,7 +36,7 @@ public interface PipelineJobId {
      *
      * @return pipeline job type
      */
-    PipelineJobType getJobType();
+    PipelineJobType<?> getJobType();
 
     /**
      * Get pipeline context key.

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

@@ -71,7 +71,7 @@ private void dispatchEnablePipelineJobStartEvent(final PipelineContextKey contex
                 .stream().filter(each -> !each.getJobName().startsWith("_")).collect(Collectors.toList());
         log.info("All job names: {}", allJobsBriefInfo.stream().map(JobBriefInfo::getJobName).collect(Collectors.joining(",")));
         for (JobBriefInfo each : allJobsBriefInfo) {
-            PipelineJobType jobType;
+            PipelineJobType<?> jobType;
             try {
                 jobType = PipelineJobIdUtils.parseJobType(each.getJobName());
             } catch (final IllegalArgumentException ex) {

kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/task/runner/TransmissionTasksRunner.java

@@ -54,7 +54,7 @@ public final class TransmissionTasksRunner implements PipelineTasksRunner {
 
     private final Collection<PipelineTask> incrementalTasks;
 
-    private final PipelineJobType jobType;
+    private final PipelineJobType<?> jobType;
 
     private final PipelineJobItemManager<TransmissionJobItemProgress> jobItemManager;
 

kernel/data-pipeline/distsql/handler/src/test/java/org/apache/shardingsphere/data/pipeline/distsql/handler/transmission/update/AlterTransmissionRuleExecutorTest.java

@@ -72,7 +72,7 @@ void setUp() throws ReflectiveOperationException {
 
     @Test
     void assertExecuteUpdate() {
-        PipelineJobType jobType = mock(PipelineJobType.class);
+        PipelineJobType<?> jobType = mock(PipelineJobType.class);
         when(jobType.getType()).thenReturn(JOB_TYPE);
         TransmissionRuleSegment segment = new TransmissionRuleSegment();
         segment.setReadSegment(new ReadOrWriteSegment(5, 1000, 200, new AlgorithmSegment("READ_LIMITER", PropertiesBuilder.build(new Property("qps", "50")))));
@@ -102,7 +102,7 @@ void assertExecuteUpdate() {
 
     @Test
     void assertExecuteUpdatePersistWhenStreamChannelIsNull() {
-        PipelineJobType jobType = mock(PipelineJobType.class);
+        PipelineJobType<?> jobType = mock(PipelineJobType.class);
         when(jobType.getType()).thenReturn(JOB_TYPE);
         AlterTransmissionRuleStatement sqlStatement = new AlterTransmissionRuleStatement(JOB_TYPE, new TransmissionRuleSegment());
         try (MockedStatic<TypedSPILoader> mockedStatic = mockStatic(TypedSPILoader.class)) {

kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJob.java

@@ -101,7 +101,7 @@ public CDCJob(final PipelineSink sink) {
     public void execute(final ShardingContext shardingContext) {
         String jobId = shardingContext.getJobName();
         log.info("Execute job {}", jobId);
-        PipelineJobType jobType = PipelineJobIdUtils.parseJobType(jobId);
+        PipelineJobType<?> jobType = PipelineJobIdUtils.parseJobType(jobId);
         PipelineContextKey contextKey = PipelineJobIdUtils.parseContextKey(jobId);
         CDCJobConfiguration jobConfig = (CDCJobConfiguration) jobType.getOption().getYamlJobConfigurationSwapper().swapToObject(shardingContext.getJobParameter());
         PipelineJobItemManager<TransmissionJobItemProgress> jobItemManager = new PipelineJobItemManager<>(jobType.getOption().getYamlJobItemProgressSwapper());

kernel/data-pipeline/scenario/cdc/core/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/CDCJobId.java

@@ -22,7 +22,6 @@
 import org.apache.shardingsphere.data.pipeline.cdc.constant.CDCSinkType;
 import org.apache.shardingsphere.data.pipeline.core.context.PipelineContextKey;
 import org.apache.shardingsphere.data.pipeline.core.job.id.PipelineJobId;
-import org.apache.shardingsphere.data.pipeline.core.job.type.PipelineJobType;
 
 import java.util.List;
 
@@ -33,7 +32,7 @@
 @Getter
 public final class CDCJobId implements PipelineJobId {
 
-    private final PipelineJobType jobType = new CDCJobType();
+    private final CDCJobType jobType = new CDCJobType();
 
     private final PipelineContextKey contextKey;
 

kernel/data-pipeline/scenario/consistency-check/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/ConsistencyCheckJobId.java

@@ -20,7 +20,6 @@
 import lombok.Getter;
 import org.apache.shardingsphere.data.pipeline.core.context.PipelineContextKey;
 import org.apache.shardingsphere.data.pipeline.core.job.id.PipelineJobId;
-import org.apache.shardingsphere.data.pipeline.core.job.type.PipelineJobType;
 import org.apache.shardingsphere.data.pipeline.scenario.consistencycheck.util.ConsistencyCheckSequence;
 
 /**
@@ -29,7 +28,7 @@
 @Getter
 public final class ConsistencyCheckJobId implements PipelineJobId {
 
-    private final PipelineJobType jobType = new ConsistencyCheckJobType();
+    private final ConsistencyCheckJobType jobType = new ConsistencyCheckJobType();
 
     private final PipelineContextKey contextKey;
 

kernel/data-pipeline/scenario/consistency-check/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/task/ConsistencyCheckTasksRunner.java

@@ -55,7 +55,7 @@
 @Slf4j
 public final class ConsistencyCheckTasksRunner implements PipelineTasksRunner {
 
-    private final PipelineJobType jobType = new ConsistencyCheckJobType();
+    private final ConsistencyCheckJobType jobType = new ConsistencyCheckJobType();
 
     private final PipelineJobManager jobManager = new PipelineJobManager(jobType);