User needs ability to specify app version when creating schedule

This update allows user to specify version.label=<version.number>
Tests were also updated because the original settings assumed that the
 appregistry was real instance instead of being mocked. Thus the find would always return null.
And in this case the tests returned a false positive.
Now that the mocks are in place it excercises all the code.
Also added explicit test if user does not set the version number.  Some of the tests do this,
but wanted an explicit test to verify this.

resolves spring-cloud#5705

spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/service/impl/DefaultSchedulerService.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2023 the original author or authors.
+ * Copyright 2018-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -297,7 +297,7 @@ public void schedule(
 				deployerDeploymentProperties,
 				commandLineArgs,
 				scheduleName,
-				getTaskResource(taskDefinitionName));
+				getTaskResource(taskDefinitionName, version));
 
 		launcher.getScheduler().schedule(scheduleRequest);
 
@@ -526,7 +526,7 @@ private static Map<String, String> extractAndQualifySchedulerProperties(Map<Stri
 						(fromWildcard, fromApp) -> fromApp));
 	}
 
-	protected Resource getTaskResource(String taskDefinitionName) {
+	protected Resource getTaskResource(String taskDefinitionName, String version) {
 		TaskDefinition taskDefinition = this.taskDefinitionRepository.findById(taskDefinitionName)
 				.orElseThrow(() -> new NoSuchTaskDefinitionException(taskDefinitionName));
 		AppRegistration appRegistration = null;
@@ -541,8 +541,14 @@ protected Resource getTaskResource(String taskDefinitionName) {
 			}
 			appRegistration = new AppRegistration(ComposedTaskRunnerConfigurationProperties.COMPOSED_TASK_RUNNER_NAME, ApplicationType.task, composedTaskUri);
 		} else {
-			appRegistration = this.registry.find(taskDefinition.getRegisteredAppName(),
+			if(version != null) {
+				appRegistration = this.registry.find(taskDefinition.getRegisteredAppName(),
+					ApplicationType.task, version);
+			}
+			else {
+				appRegistration = this.registry.find(taskDefinition.getRegisteredAppName(),
 					ApplicationType.task);
+			}
 		}
 		Assert.notNull(appRegistration, "Unknown task app: " + taskDefinition.getRegisteredAppName());
 		return this.registry.getAppResource(appRegistration);

spring-cloud-dataflow-server-core/src/test/java/org/springframework/cloud/dataflow/server/service/impl/DefaultSchedulerServiceMultiplatformTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2021 the original author or authors.
+ * Copyright 2020-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,6 +17,7 @@
 package org.springframework.cloud.dataflow.server.service.impl;
 
 import java.net.URI;
+import java.net.URISyntaxException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -73,6 +74,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -155,25 +157,20 @@ public class DefaultSchedulerServiceMultiplatformTests {
 
 	@Before
 	public void setup() throws Exception {
-		this.appRegistry.save("demo",
-				ApplicationType.task,
-				"1.0.0.",
-				new URI("file:src/test/resources/apps/foo-task"),
-				new URI("file:src/test/resources/apps/foo-task"));
-		this.appRegistry.save("demo2",
-				ApplicationType.task,
-				"1.0.0",
-				new URI("file:src/test/resources/apps/foo-task"),
-				new URI("file:src/test/resources/apps/foo-task"));
-
+		when(this.appRegistry.find(
+			eq("demo"), eq(ApplicationType.task), eq("1.0.0"))).thenReturn(new AppRegistration("demo",
+			ApplicationType.task, new URI("file:src/test/resources/apps/foo-task")));
+		when(this.appRegistry.find(
+			eq("demo2"), eq(ApplicationType.task), eq("1.0.0"))).thenReturn(new AppRegistration("demo2",
+			ApplicationType.task, new URI("file:src/test/resources/apps/foo-task")));
 		taskDefinitionRepository.save(new TaskDefinition(BASE_DEFINITION_NAME, "demo"));
 		taskDefinitionRepository.save(new TaskDefinition(CTR_DEFINITION_NAME, "demo && demo2"));
 		initializeSuccessfulRegistry();
 
 		this.testProperties = new HashMap<>();
 		this.testProperties.put(DATA_FLOW_SCHEDULER_PREFIX + "AAAA", "* * * * *");
 		this.testProperties.put(DATA_FLOW_SCHEDULER_PREFIX + "EXPRESSION", "* * * * *");
-		this.testProperties.put("version." + BASE_DEFINITION_NAME, "boot2");
+		this.testProperties.put("version." + BASE_DEFINITION_NAME, "1.0.0");
 		this.resolvedProperties = new HashMap<>();
 		this.resolvedProperties.put(DEPLOYER_PREFIX + "AAAA", "* * * * *");
 		this.resolvedProperties.put(DEPLOYER_PREFIX + "EXPRESSION", "* * * * *");
@@ -191,6 +188,13 @@ public void testSchedule() {
 		verifyScheduleExistsInScheduler(createScheduleInfo(BASE_SCHEDULE_NAME));
 	}
 
+	@Test
+	public void testScheduleWithNoVersion() {
+		this.testProperties.remove("version." + BASE_DEFINITION_NAME);
+		schedulerService.schedule(BASE_SCHEDULE_NAME, BASE_DEFINITION_NAME, this.testProperties, this.commandLineArgs, KUBERNETES_PLATFORM);
+		verifyScheduleExistsInScheduler(createScheduleInfo(BASE_SCHEDULE_NAME));
+	}
+
 	@Test(expected = IllegalArgumentException.class)
 	public void testScheduleWithLongNameOnKuberenetesPlatform() {
 		getMockedKubernetesSchedulerService().schedule(BASE_SCHEDULE_NAME +
@@ -397,15 +401,19 @@ public void testScheduleWithCommandLineArguments() throws Exception {
 	}
 
 	@Test
-	public void testScheduleWithoutCommandLineArguments() {
+	public void testScheduleWithoutCommandLineArguments() throws URISyntaxException {
 		List<String> args = getCommandLineArguments(new ArrayList<>());
 		assertThatCommandLineArgsHaveNonDefaultArgs(args, "--app.timestamp", new String[0]);
 	}
 
-	private List<String> getCommandLineArguments(List<String> commandLineArguments) {
+	private List<String> getCommandLineArguments(List<String> commandLineArguments) throws URISyntaxException {
 		Scheduler mockScheduler = mock(SimpleTestScheduler.class);
 		TaskDefinitionRepository mockTaskDefinitionRepository = mock(TaskDefinitionRepository.class);
 		AppRegistryService mockAppRegistryService = mock(AppRegistryService.class);
+		when(mockAppRegistryService.find(
+			eq("timestamp"), eq(ApplicationType.task), eq("1.0.0"))).
+			thenReturn(new AppRegistration("timestamp", ApplicationType.task,
+				new URI("file:src/test/resources/apps/timestamp-task")));
 
 
 		Launcher launcher = new Launcher("default", "defaultType", null, mockScheduler);