CTR : not working in SNAPSHOT-3.0.0

[cbonami]

Description:

Our journey began when trying to have a Composed Task with a Boot 3 based applications. The CTR of the latest released version of SCDF is still based on Boot 2. So the CTR itself started reading/writing in BOOT3-prefixed tables, while it is a Boot 2 application. That's why this didn't work. We also assumed that it is even not supposed to work due to the combination of Boot2 and Boot3. A bit odd, as Boot 3 is available already for such a long time.

So we hoped that the Boot 3 version of the CTR, available in SCDF github branch 'main3' would solve our issue. We also assumed that this branch corresponds to the image on dockerhub springcloud/spring-cloud-dataflow-composed-task-runner:3.0.0-SNAPSHOT.
We ran our composed task with SCDF version: bitnami/spring-cloud-dataflow:2.11.5-debian-12-r3.

But then, even before sub-tasks are launched, we encounter this exception in the container running the composed task:

2024-11-14 10:57:39.320  INFO 1 --- [           main] o.s.c.d.c.ComposedTaskRunner             : Started ComposedTaskRunner in 27.109 seconds (process running for 28.389)
2024-11-14 10:57:39.326  INFO 1 --- [           main] o.s.b.a.b.JobLauncherApplicationRunner   : Running default command line with: []
2024-11-14 10:57:39.538  INFO 1 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=migrate-employer]] launched with the following parameters: [{'ctr.id':'{value=15d0d628-c9ea-4c42-8b25-ac66711b573b, type=class java.lang.String, identifying=true}'}]
2024-11-14 10:57:39.620 ERROR 1 --- [           main] o.s.batch.core.job.AbstractJob           : Encountered fatal error executing job

org.springframework.dao.EmptyResultDataAccessException: Incorrect result size: expected 1, actual 0
	at org.springframework.dao.support.DataAccessUtils.nullableSingleResult(DataAccessUtils.java:190) ~[spring-tx-6.1.14.jar:6.1.14]
	at org.springframework.jdbc.core.JdbcTemplate.queryForObject(JdbcTemplate.java:890) ~[spring-jdbc-6.1.14.jar:6.1.14]
	at org.springframework.jdbc.core.JdbcTemplate.queryForObject(JdbcTemplate.java:916) ~[spring-jdbc-6.1.14.jar:6.1.14]
	at org.springframework.batch.core.repository.dao.JdbcJobExecutionDao.synchronizeStatus(JdbcJobExecutionDao.java:394) ~[spring-batch-core-5.1.2.jar:5.1.2]
	at org.springframework.batch.core.repository.support.SimpleJobRepository.update(SimpleJobRepository.java:194) ~[spring-batch-core-5.1.2.jar:5.1.2]
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:na]
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) ~[na:na]
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) ~[na:na]
	at java.base/java.lang.reflect.Method.invoke(Unknown Source) ~[na:na]
	at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:355) ~[spring-aop-6.1.14.jar:6.1.14]
	at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:196) ~[spring-aop-6.1.14.jar:6.1.14]
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163) ~[spring-aop-6.1.14.jar:6.1.14]
	at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:379) ~[spring-tx-6.1.14.jar:6.1.14]
	at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:119) ~[spring-tx-6.1.14.jar:6.1.14]
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:184) ~[spring-aop-6.1.14.jar:6.1.14]
	at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:223) ~[spring-aop-6.1.14.jar:6.1.14]
	at jdk.proxy2/jdk.proxy2.$Proxy108.update(Unknown Source) ~[na:na]
	at org.springframework.batch.core.job.AbstractJob.updateStatus(AbstractJob.java:445) ~[spring-batch-core-5.1.2.jar:5.1.2]
	at org.springframework.batch.core.job.AbstractJob.execute(AbstractJob.java:312) ~[spring-batch-core-5.1.2.jar:5.1.2]
	at org.springframework.batch.core.launch.support.SimpleJobLauncher$1.run(SimpleJobLauncher.java:157) ~[spring-batch-core-5.1.2.jar:5.1.2]
	at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50) ~[spring-core-6.1.14.jar:6.1.14]
	at org.springframework.batch.core.launch.support.SimpleJobLauncher.run(SimpleJobLauncher.java:148) ~[spring-batch-core-5.1.2.jar:5.1.2]
	at org.springframework.batch.core.launch.support.TaskExecutorJobLauncher.run(TaskExecutorJobLauncher.java:59) ~[spring-batch-core-5.1.2.jar:5.1.2]
	at org.springframework.boot.autoconfigure.batch.JobLauncherApplicationRunner.execute(JobLauncherApplicationRunner.java:210) ~[spring-boot-autoconfigure-3.3.5.jar:3.3.5]
	at org.springframework.boot.autoconfigure.batch.JobLauncherApplicationRunner.executeLocalJobs(JobLauncherApplicationRunner.java:194) ~[spring-boot-autoconfigure-3.3.5.jar:3.3.5]
	at org.springframework.boot.autoconfigure.batch.JobLauncherApplicationRunner.launchJobFromProperties(JobLauncherApplicationRunner.java:174) ~[spring-boot-autoconfigure-3.3.5.jar:3.3.5]
	at org.springframework.boot.autoconfigure.batch.JobLauncherApplicationRunner.run(JobLauncherApplicationRunner.java:169) ~[spring-boot-autoconfigure-3.3.5.jar:3.3.5]
	at org.springframework.boot.autoconfigure.batch.JobLauncherApplicationRunner.run(JobLauncherApplicationRunner.java:164) ~[spring-boot-autoconfigure-3.3.5.jar:3.3.5]
	at org.springframework.cloud.task.configuration.observation.ObservationApplicationRunner.run(ObservationApplicationRunner.java:59) ~[spring-cloud-task-core-3.1.2.jar:3.1.2]
	at org.springframework.boot.SpringApplication.lambda$callRunner$4(SpringApplication.java:786) ~[spring-boot-3.3.5.jar:3.3.5]
	at org.springframework.util.function.ThrowingConsumer$1.acceptWithException(ThrowingConsumer.java:83) ~[spring-core-6.1.14.jar:6.1.14]
	at org.springframework.util.function.ThrowingConsumer.accept(ThrowingConsumer.java:60) ~[spring-core-6.1.14.jar:6.1.14]
	at org.springframework.util.function.ThrowingConsumer$1.accept(ThrowingConsumer.java:88) ~[spring-core-6.1.14.jar:6.1.14]
	at org.springframework.boot.SpringApplication.callRunner(SpringApplication.java:798) ~[spring-boot-3.3.5.jar:3.3.5]
	at org.springframework.boot.SpringApplication.callRunner(SpringApplication.java:786) ~[spring-boot-3.3.5.jar:3.3.5]
	at org.springframework.boot.SpringApplication.lambda$callRunners$3(SpringApplication.java:774) ~[spring-boot-3.3.5.jar:3.3.5]
	at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(Unknown Source) ~[na:na]
	at java.base/java.util.stream.SortedOps$SizedRefSortingSink.end(Unknown Source) ~[na:na]
	at java.base/java.util.stream.AbstractPipeline.copyInto(Unknown Source) ~[na:na]
	at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(Unknown Source) ~[na:na]
	at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(Unknown Source) ~[na:na]
	at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(Unknown Source) ~[na:na]
	at java.base/java.util.stream.AbstractPipeline.evaluate(Unknown Source) ~[na:na]
	at java.base/java.util.stream.ReferencePipeline.forEach(Unknown Source) ~[na:na]
	at org.springframework.boot.SpringApplication.callRunners(SpringApplication.java:774) ~[spring-boot-3.3.5.jar:3.3.5]
	at org.springframework.boot.SpringApplication.run(SpringApplication.java:342) ~[spring-boot-3.3.5.jar:3.3.5]
	at org.springframework.boot.SpringApplication.run(SpringApplication.java:1363) ~[spring-boot-3.3.5.jar:3.3.5]
	at org.springframework.boot.SpringApplication.run(SpringApplication.java:1352) ~[spring-boot-3.3.5.jar:3.3.5]
	at org.springframework.cloud.dataflow.composedtaskrunner.ComposedTaskRunner.main(ComposedTaskRunner.java:31) ~[classes/:na]
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:na]
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) ~[na:na]
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) ~[na:na]
	at java.base/java.lang.reflect.Method.invoke(Unknown Source) ~[na:na]
	at org.springframework.boot.loader.MainMethodRunner.run(MainMethodRunner.java:49) ~[workspace/:na]
	at org.springframework.boot.loader.Launcher.launch(Launcher.java:95) ~[workspace/:na]
	at org.springframework.boot.loader.Launcher.launch(Launcher.java:58) ~[workspace/:na]
	at org.springframework.boot.loader.JarLauncher.main(JarLauncher.java:65) ~[workspace/:na]
	at org.springframework.boot.loader.launch.JarLauncher.main(JarLauncher.java:31) ~[workspace/:na]

2024-11-14 10:57:39.630  INFO 1 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=migrate-employer]] failed unexpectedly and fatally with the following parameters: [{'ctr.id':'{value=15d0d628-c9ea-4c42-8b25-ac66711b573b, type=class java.lang.String, identifying=true}'}]`

No clue why this happens.

Is this (still) as expected? Do we have to wait for the final 3.0.0 version of the CTR?

Release versions:

• SCDF: bitnami/spring-cloud-dataflow:2.11.5-debian-12-r3
• Composed Task Runner: springcloud/spring-cloud-dataflow-composed-task-runner:3.0.0-SNAPSHOT

[cppwfs]

Hello,
SCDF 2.11.5 only supports the Boot 2 implementation of CTR. Looking at the stack trace above it looks like it is from the BOOT-3 implementation of CTR.

If you were to create the following CTR definition, t1: timestamp && t2: timestamp do you get the same results?
If so, what properties are you setting when you launch your CTR?

[cbonami]

I will try to do what you asked asap. But in the mean time, I can confirm: we deployed a CTR Boot 3 version with a SCDF 2.11.5. Because we are so desperate to get our Boot 3 tasks/jobs running in SCDF. We really need them to be Boot 3 (and not Boot 2 anymore) because we want to use Boot 3's support for improved metrics with Micrometer and new distributed tracing support with Micrometer Tracing.

So we configured the SCDF Helm chart as follows:

spring-cloud-dataflow:
  ...
  server:
    composedTaskRunner:
      image:
        registry: docker.io
        repository: springcloud/spring-cloud-dataflow-composed-task-runner
        tag: 3.0.0-SNAPSHOT
        digest: ""
    image:
      registry: docker.io
      repository: bitnami/spring-cloud-dataflow
      tag: 2.11.5-debian-12-r3
      digest: ""

Using the default Boot 2 CTR in combination with SB3 jobs simply didn't work, as that CTR will write in different tables. Juggling with table prefixes didn't lead anywhere.

But from what I understood from your answer, running a Boot 3 CTR on SCDF 2.11.5 won't be possible anyway, right?

[cppwfs]

CTR 2.11.5 can launch Boot 3 applications. So don't use CTR 3.0.0 with SCDF 2.11.5.

[cbonami]

We didn't get CTR 2.11.5 working with Boot 3 BATCH applications. The CTR itself started reading/writing in BOOT3-prefixed tables, while it is a Boot 2 application. We received SQL exceptions about jobs not existing etc. We will check this again and give more details.

[klopfdreh]

Hey @cbonami,

in your task application are you defining a custom JobExplorer or JobRepository?

If so try to autowire org.springframework.boot.autoconfigure.batch.BatchProperties into the bean definition and create the beans like this

        JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
        // ....
        Optional.ofNullable(batchProperties)
            .map(BatchProperties::getJdbc)
            .map(BatchProperties.Jdbc::getTablePrefix)
            .ifPresent(factory::setTablePrefix);
        factory.afterPropertiesSet();
        return Objects.requireNonNull(factory.getObject());

        JobExplorerFactoryBean jobExplorerFactoryBean = new JobExplorerFactoryBean();
        // ....
        Optional.ofNullable(batchProperties)
            .map(BatchProperties::getJdbc)
            .map(BatchProperties.Jdbc::getTablePrefix)
            .ifPresent(jobExplorerFactoryBean::setTablePrefix);
        jobExplorerFactoryBean.afterPropertiesSet();
        return Objects.requireNonNull(jobExplorerFactoryBean.getObject());

[cbonami]

@klopfdreh , thanks for your reply. But no, we did not provide a custom JobExplorer or JobRepository.

Anyway, in the meantime we could get it running via a hack. At least, we consider this a hack.

We reverted everything to SCDF 2.11.5 and the standard Boot2-based CTR.

We deployed our Composed Task comprising 2 subtasks: a Boot3 task, and a Boot3 batch app.
Like in our initial experiments, the FIRST run of the CT worked just fine.
However, when launching the CT for the second time, we received this:

Caused by: org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException: A job instance already exists and is complete for parameters={run.id=1}.  If you want to run this job again, change the parameters.

This is what put us on the wrong foot during our first experiments when we tried the CTR (Boot2) with the 2 subtasks that where Boot2 before, but that we converted to Boot3. At that time we received exactly the same error, something that we never encountered in the Boot2-world. This made us conclude that the CTR also needed to be Boot3, which apparently is not the case.

So now, we circumvented this problem by adding an argument (called 'unique') when starting the CT. The argument contains a random UUID.

Now we can run the CT multiple times without any problem; this is what we see in the logs:

Job: [FlowJob: [name=migrate-employer]] launched with the following parameters: [{run.id=1, unique=fd1ed4b1-c9a6-40a2-9f9f-ac77a2414da0}]

As you can see, run.id always remains 1 for some reason, but we make the arguments unique with our 'hack'.
Of course, we don't think that this is how this is intended to be done. The run.id should auto-increment, just like it did (we suppose) when using the Boot2-subtasks.

[cppwfs]

To clarify. Its CTR that is throwing the JobInstanceAlreadyCompleteException exception?