Provide for correctly resolving app registration from version supplie…

…d during launch. Fixes #5467
spring-cloud · Sep 15, 2023 · 6f6a096 · 6f6a096
1 parent fa835fa
commit 6f6a096
Show file tree

Hide file tree

Showing 12 changed files with 251 additions and 109 deletions.
diff --git a/...ain/java/org/springframework/cloud/dataflow/aggregate/task/AggregateExecutionSupport.java b/...ain/java/org/springframework/cloud/dataflow/aggregate/task/AggregateExecutionSupport.java
@@ -43,12 +43,15 @@ public interface AggregateExecutionSupport {
 	 * @return The {@link SchemaVersionTarget} for the taskName specified.
 	 */
 	SchemaVersionTarget findSchemaVersionTarget(String taskName, TaskDefinitionReader taskDefinitionReader);
+	SchemaVersionTarget findSchemaVersionTarget(String taskName, String version, TaskDefinitionReader taskDefinitionReader);
 	SchemaVersionTarget findSchemaVersionTarget(String taskName, TaskDefinition taskDefinition);
+	SchemaVersionTarget findSchemaVersionTarget(String taskName, String version, TaskDefinition taskDefinition);
 
 	/**
 	 * Retrieve the {@link AppRegistration} for the registeredName.
 	 */
 	AppRegistration findTaskAppRegistration(String registeredName);
+	AppRegistration findTaskAppRegistration(String registeredName, String version);
 
 	/**
 	 * Return the {@link AggregateTaskExecution} for the {@link TaskExecution} and Schema Target name specified.

diff --git a/.../springframework/cloud/dataflow/aggregate/task/impl/DefaultAggregateExecutionSupport.java b/.../springframework/cloud/dataflow/aggregate/task/impl/DefaultAggregateExecutionSupport.java
@@ -42,7 +42,7 @@
  */
 
 public class DefaultAggregateExecutionSupport implements AggregateExecutionSupport {
-	private static Logger logger = LoggerFactory.getLogger(AggregateExecutionSupport.class);
+	private static final Logger logger = LoggerFactory.getLogger(AggregateExecutionSupport.class);
 
 	private final AppRegistryService registryService;
 
@@ -80,12 +80,29 @@ public SchemaVersionTarget findSchemaVersionTarget(String taskName, TaskDefiniti
 		return findSchemaVersionTarget(taskName, definition);
 	}
 
+	@Override
+	public SchemaVersionTarget findSchemaVersionTarget(String taskName, String version, TaskDefinitionReader taskDefinitionReader) {
+		logger.debug("findSchemaVersionTarget:{}:{}", taskName, version);
+		TaskDefinition definition = taskDefinitionReader.findTaskDefinition(taskName);
+		return findSchemaVersionTarget(taskName, version, definition);
+	}
+
 	@Override
 	public SchemaVersionTarget findSchemaVersionTarget(String taskName, TaskDefinition taskDefinition) {
+		return findSchemaVersionTarget(taskName, null, taskDefinition);
+	}
+
+	@Override
+	public SchemaVersionTarget findSchemaVersionTarget(String taskName, String version, TaskDefinition taskDefinition) {
+		logger.debug("findSchemaVersionTarget:{}:{}", taskName, version);
 		String registeredName = taskDefinition != null ? taskDefinition.getRegisteredAppName() : taskName;
-		AppRegistration registration = findTaskAppRegistration(registeredName);
+		AppRegistration registration = findTaskAppRegistration(registeredName, version);
 		if (registration == null) {
-			logger.warn("Cannot find AppRegistration for {}", taskName);
+			if(StringUtils.hasLength(version)) {
+				logger.warn("Cannot find AppRegistration for {}:{}", taskName, version);
+			} else {
+				logger.warn("Cannot find AppRegistration for {}", taskName);
+			}
 			return SchemaVersionTarget.defaultTarget();
 		}
 		final AppRegistration finalRegistration = registration;
@@ -101,16 +118,26 @@ public SchemaVersionTarget findSchemaVersionTarget(String taskName, TaskDefiniti
 			throw new IllegalStateException("Multiple SchemaVersionTargets for " + registration.getBootVersion());
 		}
 		SchemaVersionTarget schemaVersionTarget = versionTargets.get(0);
-		logger.debug("findSchemaVersionTarget:{}={},{}", taskName, registeredName, schemaVersionTarget);
+		logger.debug("findSchemaVersionTarget:{}:{}:{}={}", taskName, registeredName, version, schemaVersionTarget);
 		return schemaVersionTarget;
 	}
 
 	@Override
-	public AppRegistration findTaskAppRegistration(String registeredAppName) {
-		AppRegistration registration = registryService.find(registeredAppName, ApplicationType.task);
+	public AppRegistration findTaskAppRegistration(String registeredName) {
+		return findTaskAppRegistration(registeredName, null);
+	}
+
+	@Override
+	public AppRegistration findTaskAppRegistration(String registeredAppName, String version) {
+		AppRegistration registration = StringUtils.hasLength(version) ?
+			registryService.find(registeredAppName, ApplicationType.task, version) :
+			registryService.find(registeredAppName, ApplicationType.task);
 		if (registration == null) {
-			registration = registryService.find(registeredAppName, ApplicationType.app);
+			registration = StringUtils.hasLength(version) ?
+				registryService.find(registeredAppName, ApplicationType.app, version) :
+				registryService.find(registeredAppName, ApplicationType.app);
 		}
+		logger.debug("findTaskAppRegistration:{}:{}={}", registeredAppName, version, registration);
 		return registration;
 	}
 

diff --git a/...low-core/src/main/java/org/springframework/cloud/dataflow/core/DataFlowAppDefinition.java b/...low-core/src/main/java/org/springframework/cloud/dataflow/core/DataFlowAppDefinition.java
@@ -109,6 +109,8 @@ public ApplicationType getApplicationType() {
 		return applicationType;
 	}
 
+	public AppDefinition getAppDefinition() { return appDefinition; }
+
 	@Override
 	public boolean equals(Object obj) {
 		if (this == obj)

diff --git a/...n/java/org/springframework/cloud/dataflow/registry/service/DefaultAppRegistryService.java b/...n/java/org/springframework/cloud/dataflow/registry/service/DefaultAppRegistryService.java
@@ -106,7 +106,9 @@ public AppRegistration find(String name, ApplicationType type) {
 
 	@Override
 	public AppRegistration find(String name, ApplicationType type, String version) {
-		return this.appRegistrationRepository.findAppRegistrationByNameAndTypeAndVersion(name, type, version);
+		AppRegistration registration = this.appRegistrationRepository.findAppRegistrationByNameAndTypeAndVersion(name, type, version);
+		logger.debug("find:{}:{}:{}={}", name, type, version, registration);
+		return registration;
 	}
 
 	@Override

diff --git a/.../java/org/springframework/cloud/dataflow/server/service/TaskExecutionCreationService.java b/.../java/org/springframework/cloud/dataflow/server/service/TaskExecutionCreationService.java
@@ -30,5 +30,5 @@ public interface TaskExecutionCreationService {
 	 * @param taskName the name to be associated with the {@link TaskExecution}
 	 * @return {@link TaskExecution}
 	 */
-	TaskExecution createTaskExecution(String taskName);
+	TaskExecution createTaskExecution(String taskName, String version);
 }
diff --git a/...main/java/org/springframework/cloud/dataflow/server/service/TaskExecutionInfoService.java b/...main/java/org/springframework/cloud/dataflow/server/service/TaskExecutionInfoService.java
@@ -61,4 +61,5 @@ TaskExecutionInformation findTaskExecutionInformation(String taskName,
 	List<AppDeploymentRequest> createTaskDeploymentRequests(String taskName, String dslText);
 
 	Set<String> composedTaskChildNames(String taskName);
+	Set<String> taskNames(String taskName);
 }
diff --git a/.../java/org/springframework/cloud/dataflow/server/service/impl/DefaultSchedulerService.java b/.../java/org/springframework/cloud/dataflow/server/service/impl/DefaultSchedulerService.java
@@ -222,7 +222,12 @@ public void schedule(
 		}
 
 		String taskAppName = taskDefinition.getRegisteredAppName();
-		SchemaVersionTarget schemaVersionTarget = aggregateExecutionSupport.findSchemaVersionTarget(taskAppName, taskDefinitionReader);
+		String taskLabel = taskDefinition.getAppDefinition().getName();
+		if(!StringUtils.hasText(taskLabel)) {
+			taskLabel = taskAppName;
+		}
+		String version = taskDeploymentProperties.get("version." + taskLabel);
+		SchemaVersionTarget schemaVersionTarget = aggregateExecutionSupport.findSchemaVersionTarget(taskAppName, version, taskDefinition);
 		Assert.notNull(schemaVersionTarget, "schemaVersionTarget not found for " + taskAppName);
 		TaskParser taskParser = new TaskParser(taskDefinition.getName(), taskDefinition.getDslText(), true, true);
 		TaskNode taskNode = taskParser.parse();
@@ -258,8 +263,15 @@ public void schedule(
 				if (names.size() > 1) {
 					appId = names.get(1);
 				}
-				SchemaVersionTarget appSchemaTarget = this.aggregateExecutionSupport.findSchemaVersionTarget(registeredName, taskDefinitionReader);
-				logger.debug("ctr:{}:registeredName={}, schemaTarget={}", names, registeredName, appSchemaTarget.getName());
+				String appVersion = taskDeploymentProperties.get("version." + taskAppName + "-" + appId + "." + appId);
+				if(!StringUtils.hasText(appVersion)) {
+					appVersion = taskDeploymentProperties.get("version." + taskAppName + "-" + appId);
+				}
+				if(!StringUtils.hasText(appVersion)) {
+					appVersion = taskDeploymentProperties.get("version." + appId);
+				}
+				SchemaVersionTarget appSchemaTarget = this.aggregateExecutionSupport.findSchemaVersionTarget(registeredName, appVersion, taskDefinitionReader);
+				logger.debug("ctr:{}:registeredName={}, version={}, schemaTarget={}", names, registeredName, appVersion, appSchemaTarget.getName());
 				taskDeploymentProperties.put("app.composed-task-runner.composed-task-app-properties.app." + scheduleName + "-" + appId + ".spring.cloud.task.tablePrefix",
 						appSchemaTarget.getTaskPrefix());
 				taskDeploymentProperties.put("app.composed-task-runner.composed-task-app-properties.app." + appId + ".spring.cloud.task.tablePrefix",
@@ -437,9 +449,7 @@ private Launcher getTaskLauncher(String platformName) {
 	private List<Launcher> getLaunchers() {
 		List<Launcher> launchers = new ArrayList<>();
 		for (TaskPlatform taskPlatform : this.taskPlatforms) {
-			for (Launcher launcher : taskPlatform.getLaunchers()) {
-				launchers.add(launcher);
-			}
+			launchers.addAll(taskPlatform.getLaunchers());
 		}
 		return launchers;
 	}