强制终止后 Spring Batch 不更新作业存储库

Question

我正在使用带有 SpringBatch 4.3.x 的 SpringBoot 2.4.x 应用程序。我创建了一个简单的工作。 我有从 CSV 文件读取的 FlatFileItemReader。我有 ImportKafkaItemWriter 写入 Kafka 主题。我将这些结合在一起的一步。我正在使用 SimpleJobLauncher，并将 ThreadPoolTask​​Executor 设置为 JobLauncher 的 TasKExecutor。正如我所料，它工作正常。但是我有一个弹性用例，如果我终止应用程序然后重新启动应用程序并触发作业，那么它将继续并完成剩余的作业。不幸的是，它没有发生。我做了进一步调查，发现当我强行关闭应用程序时，SpringBatch 作业存储库键表如下所示：

job_execution_id	version	job_instance_id	create_time	start_time	end_time	status	exit_code	exit_message	last_updated	job_configuration_location
1	1	1	2021-06-16 09:32:43	2021-06-16 09:32:43		STARTED	UNKNOWN		2021-06-16 09:32:43

和

step_execution_id	version	step_name	job_execution_id	start_time	end_time	status	commit_count	read_count	filter_count	write_count	read_skip_count	write_skip_count	process_skip_count	rollback_count	exit_code	exit_message	last_updated
1	4	productImportStep	1	2021-06-16 09:32:43		STARTED	3	6	0	6	0	0	0	0	EXECUTING		2021-06-16 09:32:50

如果我手动更新这些表，其中我将有效的 end_time 和状态设置为 FAILED，那么我可以重新启动作业并且工作正常。我可以知道我需要做什么，以便 Spring Batch 可以适当地更新那些相关的存储库，并且我可以避免这个手动步骤。如果需要，我可以提供有关代码的更多信息。

Answer 1

如果我手动更新这些表，我将有效的 end_time 和状态设置为 FAILED，那么我可以重新启动作业并且工作正常。我可以知道我需要做什么，以便 Spring Batch 可以适当地更新那些相关的存储库，并且我可以避免这个手动步骤

当一个作业被突然终止时，Spring Batch 将没有机会更新其在 Job 存储库中的状态，因此状态停留在 STARTED。现在，当作业重新启动时，Spring Batch 拥有的唯一信息是作业存储库中的状态。仅通过查看数据库中的状态，Spring Batch 无法区分有效运行的作业和已突然终止的作业（在这两种情况下，状态都是STARTED）。

确实要手动更新表以将状态标记为FAILED 以便能够重新启动作业或ABANDONED 以放弃它。这是您必须做出的业务决策，并且无法在框架方面实现自动化。更多详细信息，请参阅此处的参考文档：Aborting a Job。

Answer 2

您可以添加一个伪造的参数示例 为每次执行新作业时增加一个计数器版本，这样您就不必检查表数据库作业。

我的意思是 mvn clean package 然后你尝试像这样启动程序： java my-jarfile.jar dest=/tmp/foo Version="0" java my-jarfile.jar dest=/tmp/foo Version="1" java my-jarfile.jar dest=/tmp/foo Version="2" 等等……或者 您可以使用 jobParameters 通过 jobLauncher 以编程方式启动作业并使用日期参数 date = new Date().toString() 在每次执行新作业时提供带有新标记的日期

Answer 3

你可以使用“JVM Shutdown Hook”： 像这样的：

Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    if (jobExecution.isRunning()) {
        jobExecution.setEndTime(new Date());
        jobExecution.setStatus(BatchStatus.FAILED);
        jobExecution.setExitStatus(ExitStatus.FAILED);
        jobRepository.update(jobExecution);
    }
}));