使用DeployerPartitionHandler在spring cloud任务中运行的spring批处理的launchWorker中获取空指针异常

Question

我正在使用DeployerPartitionHandler（本地变体）对Spring批处理作业进行分区。当我运行作业时，在worker的启动步骤中得到一个空指针异常，如下所示

at org.springframework.cloud.task.batch.partition.DeployerPartitionHandler.launchWorkers(DeployerPartitionHandler.java:313)
at org.springframework.cloud.task.batch.partition.DeployerPartitionHandler.handle(DeployerPartitionHandler.java:302)
at org.springframework.batch.core.partition.support.PartitionStep.doExecute(PartitionStep.java:106)
at org.springframework.batch.core.step.AbstractStep.execute(AbstractStep.java:208)
at org.springframework.batch.core.job.SimpleStepHandler.handleStep(SimpleStepHandler.java:148)
at org.springframework.batch.core.job.AbstractJob.handleStep(AbstractJob.java:410)
at org.springframework.batch.core.job.SimpleJob.doExecute(SimpleJob.java:136)
at org.springframework.batch.core.job.AbstractJob.execute(AbstractJob.java:319)
at org.springframework.batch.core.launch.support.SimpleJobLauncher$1.run(SimpleJobLauncher.java:147)
at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
at org.springframework.batch.core.launch.support.SimpleJobLauncher.run(SimpleJobLauncher.java:140)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:566)
at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:343)
at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:198)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
at org.springframework.batch.core.configuration.annotation.SimpleBatchConfiguration$PassthruAdvice.invoke(SimpleBatchConfiguration.java:127)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:212)
at com.sun.proxy.$Proxy248.run(Unknown Source)
at org.springframework.boot.autoconfigure.batch.JobLauncherCommandLineRunner.execute(JobLauncherCommandLineRunner.java:207)
at org.springframework.boot.autoconfigure.batch.JobLauncherCommandLineRunner.executeLocalJobs(JobLauncherCommandLineRunner.java:181)
at org.springframework.boot.autoconfigure.batch.JobLauncherCommandLineRunner.launchJobFromProperties(JobLauncherCommandLineRunner.java:168)
at org.springframework.boot.autoconfigure.batch.JobLauncherCommandLineRunner.run(JobLauncherCommandLineRunner.java:163)
at org.springframework.boot.SpringApplication.callRunner(SpringApplication.java:781)
at org.springframework.boot.SpringApplication.callRunners(SpringApplication.java:765)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:319)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1215)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1204).

当作业开始时，我会看到下面的错误消息

调试[O.S.C.task.repository.support.SimpleTaskrepository]创建：TaskExecution{executionid=0,parentexecutionid=null,exitcode=null,taskname='mytask',starttime=tue nov24 17:14:48 IST 2020,endtime=null,exitmessage='null“,exteralexecutionid='null”,errormessage='null“,arguments=[--spring.profiles.active=local,

DEBUG[O.S.C.T.Batch.Partition.DeployerPartitionHandler]返回3个分区[main][]

此外，我在TaskRepository中看到了另一个条目，但这一次它在大多数列DEBUG[o.s.c.task.repository.support.simpleTaskRepository]中都是空值：TaskExecution{executionid=65,parentexecutionid=null,exitcode=null,taskname='null',starttime=null,endtime=null,exitmessage='null',exitmessage='null',externalexecutionid='null',

我认为，由于spring批处理是一个单一的任务，所以我希望TaskRepository中只有一个条目，但无法理解为什么要进行第二个条目。我使用Postgres并遵循示例代码中提到的相同步骤，但无法理解https://github.com/spring-cloud/spring-cloud-task/tree/master/spring-cloud-task-samples/partitioned-batch-job问题

下面是代码，以供参考

@Bean
@Profile("!worker")
public PartitionHandler partitionHandler(TaskLauncher taskLauncher, JobExplorer jobExplorer,
        TaskRepository taskRepository) throws Exception {
    

    Resource resource = this.resourceLoader.getResource("maven://XXX:YYYY:0.0.2-SNAPSHOT");

    DeployerPartitionHandler partitionHandler =
                new DeployerPartitionHandler(taskLauncher, jobExplorer, resource, "workerStep", taskRepository);
      
      
    List<String> commandLineArgs = new ArrayList<>(3);
    commandLineArgs.add("--spring.profiles.active=worker");
    commandLineArgs.add("--spring.cloud.task.initialize-enabled=false");
    commandLineArgs.add("--spring.batch.initializer.enabled=false");
    partitionHandler.setCommandLineArgsProvider(new PassThroughCommandLineArgsProvider(commandLineArgs));
    partitionHandler.setEnvironmentVariablesProvider(new SimpleEnvironmentVariablesProvider(this.environment));
    partitionHandler.setMaxWorkers(2);
    partitionHandler.setApplicationName("MYTASK");

    return partitionHandler;
}


@Bean
@Profile("!worker")
public Job partitionedJob(PartitionHandler partitionHandler) throws Exception {
    
    Random random = new Random();
    return this.jobBuilderFactory.get("partitionedJob" + random.nextInt())
        .start(step1(partitionHandler))
        .build();
}

@Bean
@Profile("!worker")
public Step step1(PartitionHandler partitionHandler) throws Exception {

    return this.stepBuilderFactory.get("step1")
        .partitioner(workerStep().getName(), partitioner())         
        .step(workerStep())
        .partitionHandler(partitionHandler)
        .build();
}


@Bean
@Profile("!worker")
public Partitioner partitioner() {
    return new Partitioner() {
        @Override
        public Map<String, ExecutionContext> partition(int gridSize) {
            System.out.println("In partitioner");
            Map<String, ExecutionContext> partitions = new HashMap<>(gridSize);

            for (int i = 0; i < 3; i++) {
                ExecutionContext context1 = new ExecutionContext();
                context1.put("partitionNumber", i);

                partitions.put("partition" + i, context1);
            }
            
            return partitions;
        }
    };
}

@Bean
@Profile("worker")
public DeployerStepExecutionHandler stepExecutionHandler(JobExplorer jobExplorer) {
    
    return new DeployerStepExecutionHandler(this.context, jobExplorer, this.jobRepository);
}

@Bean

public Step workerStep() {

    return this.stepBuilderFactory.get("workerStep")
        .tasklet(workerTasklet(null))
    
        .build();
}

@Bean
@StepScope
public Tasklet workerTasklet(
    final @Value("#{stepExecutionContext['partitionNumber']}") Integer partitionNumber) {

    return new Tasklet() {
        @Override
        public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {

            return RepeatStatus.FINISHED;
        }
    };
}

Answer 1

因为代码似乎有点不完整，所以不能百分之百确定。

但我可以解决类似的问题后，添加

1. @enableTask

任务2和任务3将在另一个表中有一个映射，以说明它们与任务#1相关联