diff --git a/projects/control-service/projects/pipelines_control_service/src/integration-test/java/com/vmware/taurus/datajobs/it/DataJobDeploymentCrudIT.java b/projects/control-service/projects/pipelines_control_service/src/integration-test/java/com/vmware/taurus/datajobs/it/DataJobDeploymentCrudIT.java
index 5e0c34debe..178b824528 100644
--- a/projects/control-service/projects/pipelines_control_service/src/integration-test/java/com/vmware/taurus/datajobs/it/DataJobDeploymentCrudIT.java
+++ b/projects/control-service/projects/pipelines_control_service/src/integration-test/java/com/vmware/taurus/datajobs/it/DataJobDeploymentCrudIT.java
@@ -41,7 +41,7 @@ private void setVdkVersionForDeployment() throws Exception {
                 .with(user("user"))
                 .content(getDataJobDeploymentVdkVersionRequestBody("new_vdk_version_tag"))
                 .contentType(MediaType.APPLICATION_JSON))
-        .andExpect(status().isBadRequest());
+        .andExpect(status().isAccepted());
   }
 
   private void disableDeployment() throws Exception {
@@ -80,7 +80,7 @@ private void resetVdkDeploymentVersion() throws Exception {
                 .with(user("user"))
                 .content(getDataJobDeploymentVdkVersionRequestBody(""))
                 .contentType(MediaType.APPLICATION_JSON))
-        .andExpect(status().isBadRequest());
+        .andExpect(status().isAccepted());
   }
 
   private MvcResult getDeployment() throws Exception {
diff --git a/projects/control-service/projects/pipelines_control_service/src/integration-test/java/com/vmware/taurus/datajobs/it/TestJobImageBuilderDynamicVdkImageIT.java b/projects/control-service/projects/pipelines_control_service/src/integration-test/java/com/vmware/taurus/datajobs/it/TestJobImageBuilderDynamicVdkImageIT.java
index de5e775fad..303a3bf7e1 100644
--- a/projects/control-service/projects/pipelines_control_service/src/integration-test/java/com/vmware/taurus/datajobs/it/TestJobImageBuilderDynamicVdkImageIT.java
+++ b/projects/control-service/projects/pipelines_control_service/src/integration-test/java/com/vmware/taurus/datajobs/it/TestJobImageBuilderDynamicVdkImageIT.java
@@ -212,9 +212,9 @@ public void testDataJobDeploymentDynamicVdkVersion() throws Exception {
                 .with(user("user"))
                 .content(getDataJobDeploymentVdkVersionRequestBody("new_vdk_version_tag"))
                 .contentType(MediaType.APPLICATION_JSON))
-        .andExpect(status().isBadRequest());
+        .andExpect(status().isAccepted());
 
-    // verify vdk version is not changed
+    // verify vdk version is changed
     mockMvc
         .perform(
             get(String.format(
@@ -223,7 +223,7 @@ public void testDataJobDeploymentDynamicVdkVersion() throws Exception {
                 .with(user("user"))
                 .contentType(MediaType.APPLICATION_JSON))
         .andExpect(status().isOk())
-        .andExpect(jsonPath("$.vdk_version", is("release")));
+        .andExpect(jsonPath("$.vdk_version", is("new_vdk_version_tag")));
 
     // Execute change python version and set corresponding vdk version for deployment
     mockMvc
diff --git a/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/datajobs/DeploymentModelConverter.java b/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/datajobs/DeploymentModelConverter.java
index ad536d3818..353df76a25 100644
--- a/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/datajobs/DeploymentModelConverter.java
+++ b/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/datajobs/DeploymentModelConverter.java
@@ -62,6 +62,7 @@ public static DesiredDataJobDeployment toDesiredDataJobDeployment(JobDeployment
 
     deployment.setGitCommitSha(jobDeployment.getGitCommitSha());
     deployment.setPythonVersion(jobDeployment.getPythonVersion());
+    deployment.setVdkVersion(jobDeployment.getVdkVersion());
 
     return deployment;
   }
@@ -222,6 +223,10 @@ public static DesiredDataJobDeployment mergeDeployments(
         newDeployment.getPythonVersion() != null
             ? newDeployment.getPythonVersion()
             : oldDeployment.getPythonVersion());
+    mergedDeployment.setVdkVersion(
+        newDeployment.getVdkVersion() != null
+            ? newDeployment.getVdkVersion()
+            : oldDeployment.getVdkVersion());
     mergedDeployment.setLastDeployedBy(
         userDeployer != null ? userDeployer : oldDeployment.getLastDeployedBy());
     mergedDeployment.setSchedule(
@@ -236,6 +241,8 @@ public static DesiredDataJobDeployment mergeDeployments(
             ? newDeployment.getEnabled()
             : oldDeployment.getEnabled());
 
+    resetVdkVersionIfPythonVersionChange(oldDeployment, mergedDeployment);
+
     return mergedDeployment;
   }
 
@@ -260,6 +267,15 @@ private static void checkDeploymentsCanBeMerged(
     }
   }
 
+  private static void resetVdkVersionIfPythonVersionChange(
+      DesiredDataJobDeployment oldDeployment, DesiredDataJobDeployment newDeployment) {
+    if (newDeployment.getPythonVersion() != null
+        && oldDeployment.getPythonVersion() != null
+        && !oldDeployment.getPythonVersion().equals(newDeployment.getPythonVersion())) {
+      newDeployment.setVdkVersion(null);
+    }
+  }
+
   private static void mergeDeploymentResources(
       DesiredDataJobDeployment mergedDeployment,
       JobDeployment newDeployment,
@@ -311,6 +327,7 @@ public static DataJobDeploymentStatus toJobDeploymentStatus(
     var deploymentStatus = new DataJobDeploymentStatus();
     deploymentStatus.setJobVersion(actualDataJobDeployment.getGitCommitSha());
     deploymentStatus.setPythonVersion(actualDataJobDeployment.getPythonVersion());
+    deploymentStatus.setVdkVersion(actualDataJobDeployment.getVdkVersion());
     deploymentStatus.setId(actualDataJobDeployment.getDataJobName());
     deploymentStatus.setEnabled(actualDataJobDeployment.getEnabled());
     deploymentStatus.setContacts(getContactsFromJob(job));
diff --git a/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/deploy/DeploymentService.java b/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/deploy/DeploymentService.java
index cde6f3dc78..9afa76eb5b 100644
--- a/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/deploy/DeploymentService.java
+++ b/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/deploy/DeploymentService.java
@@ -72,6 +72,7 @@ public void patchDeployment(DataJob dataJob, JobDeployment jobDeployment) {
               dataJob.getJobConfig().getTeam(), dataJob.getName(), deploymentStatus.get());
       var mergedDeployment =
           DeploymentModelConverter.mergeDeployments(oldDeployment, jobDeployment);
+      resetVdkVersionIfPythonVersionChange(oldDeployment, mergedDeployment);
       validateFieldsCanBePatched(oldDeployment, mergedDeployment);
 
       // we are setting sendNotification to false since it's not necessary. If something fails we'd
@@ -122,17 +123,6 @@ private void validateFieldsCanBePatched(
           mergedDeployment.getPythonVersion(),
           "Use POST HTTP request to change python version.");
     }
-
-    if (mergedDeployment.getVdkVersion() != null
-        && !mergedDeployment.getVdkVersion().equals(oldDeployment.getVdkVersion())) {
-      throw new ApiConstraintError(
-          "vdk_version",
-          String.format(
-              "same as the current vdk version -- %s -- when using PATCH request.",
-              oldDeployment.getVdkVersion()),
-          mergedDeployment.getPythonVersion(),
-          "Use POST HTTP request to change vdk version.");
-    }
   }
 
   /**
@@ -172,6 +162,7 @@ public void updateDeployment(
                 dataJob.getJobConfig().getTeam(), dataJob.getName(), deploymentStatus.get());
         setPythonVersionIfNull(oldDeployment, jobDeployment);
         jobDeployment = DeploymentModelConverter.mergeDeployments(oldDeployment, jobDeployment);
+        resetVdkVersionIfPythonVersionChange(oldDeployment, jobDeployment);
       }
 
       if (jobDeployment.getPythonVersion() == null) {
@@ -280,4 +271,13 @@ private void setPythonVersionIfNull(JobDeployment oldDeployment, JobDeployment n
       newDeployment.setPythonVersion(supportedPythonVersions.getDefaultPythonVersion());
     }
   }
+
+  private void resetVdkVersionIfPythonVersionChange(
+      JobDeployment oldDeployment, JobDeployment newDeployment) {
+    if (newDeployment.getPythonVersion() != null
+        && oldDeployment.getPythonVersion() != null
+        && !oldDeployment.getPythonVersion().equals(newDeployment.getPythonVersion())) {
+      newDeployment.setVdkVersion(null);
+    }
+  }
 }
diff --git a/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/deploy/JobImageDeployer.java b/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/deploy/JobImageDeployer.java
index 3b560f9b5b..9433b5495c 100644
--- a/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/deploy/JobImageDeployer.java
+++ b/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/deploy/JobImageDeployer.java
@@ -254,7 +254,10 @@ private void updateCronJob(DataJob dataJob, JobDeployment jobDeployment, String
             "-c",
             "cp -r $(python -c \"from distutils.sysconfig import get_python_lib;"
                 + " print(get_python_lib())\") /vdk/. && cp /usr/local/bin/vdk /vdk/.");
-    var jobVdkImage = supportedPythonVersions.getVdkImage(jobDeployment.getPythonVersion());
+    var jobVdkImage =
+        isVdkVersionPassedDifferentFromOneSetByPythonVersion(jobDeployment)
+            ? supportedPythonVersions.replaceVdkVersionInImage(jobDeployment.getVdkVersion())
+            : supportedPythonVersions.getVdkImage(jobDeployment.getPythonVersion());
     var jobInitContainer =
         KubernetesService.container(
             "vdk",
@@ -372,4 +375,16 @@ private Map<String, String> getJobAnnotations(
   public static String getCronJobName(String jobName) {
     return jobName;
   }
+
+  private boolean isVdkVersionPassedDifferentFromOneSetByPythonVersion(
+      JobDeployment jobDeployment) {
+    var passedVdkVersion = jobDeployment.getVdkVersion();
+    var vdkVersionSetByPythonVersion =
+        DockerImageName.getTag(
+            supportedPythonVersions.getVdkImage(jobDeployment.getPythonVersion()));
+
+    return passedVdkVersion != null
+        && !passedVdkVersion.isEmpty()
+        && !passedVdkVersion.equals(vdkVersionSetByPythonVersion);
+  }
 }
diff --git a/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/deploy/JobImageDeployerV2.java b/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/deploy/JobImageDeployerV2.java
index 7fe0ddd936..7c9862b629 100644
--- a/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/deploy/JobImageDeployerV2.java
+++ b/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/deploy/JobImageDeployerV2.java
@@ -373,7 +373,10 @@ private KubernetesService.CronJob getCronJob(
             "-c",
             "cp -r $(python -c \"from distutils.sysconfig import get_python_lib;"
                 + " print(get_python_lib())\") /vdk/. && cp /usr/local/bin/vdk /vdk/.");
-    var jobVdkImage = supportedPythonVersions.getVdkImage(jobDeployment.getPythonVersion());
+    var jobVdkImage =
+        isVdkVersionPassedDifferentFromOneSetByPythonVersion(jobDeployment)
+            ? supportedPythonVersions.replaceVdkVersionInImage(jobDeployment.getVdkVersion())
+            : supportedPythonVersions.getVdkImage(jobDeployment.getPythonVersion());
     var jobInitContainer =
         KubernetesService.container(
             "vdk",
@@ -481,4 +484,16 @@ private void handleResourcesException(ParseException e) {
             "Verify the string can be parsed to a number");
     log.error(errorMessage.toString(), e);
   }
+
+  private boolean isVdkVersionPassedDifferentFromOneSetByPythonVersion(
+      DesiredDataJobDeployment jobDeployment) {
+    var passedVdkVersion = jobDeployment.getVdkVersion();
+    var vdkVersionSetByPythonVersion =
+        DockerImageName.getTag(
+            supportedPythonVersions.getVdkImage(jobDeployment.getPythonVersion()));
+
+    return passedVdkVersion != null
+        && !passedVdkVersion.isEmpty()
+        && !passedVdkVersion.equals(vdkVersionSetByPythonVersion);
+  }
 }
diff --git a/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/deploy/SupportedPythonVersions.java b/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/deploy/SupportedPythonVersions.java
index 3775a88ff3..2f5092a78e 100644
--- a/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/deploy/SupportedPythonVersions.java
+++ b/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/deploy/SupportedPythonVersions.java
@@ -134,4 +134,9 @@ public String getDefaultPythonVersion() {
   private String getBuilderImage(Map<String, String> supportedPythonVersion) {
     return supportedPythonVersion.getOrDefault(BUILDER_IMAGE, dockerRegistryService.builderImage());
   }
+
+  public String replaceVdkVersionInImage(String vdkVersion) {
+    String defaultVdkImage = getDefaultVdkImage();
+    return defaultVdkImage.replace(DockerImageName.getTag(defaultVdkImage), vdkVersion);
+  }
 }
diff --git a/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/model/BaseDataJobDeployment.java b/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/model/BaseDataJobDeployment.java
index ef4a3ca2da..9a248ca1c9 100644
--- a/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/model/BaseDataJobDeployment.java
+++ b/projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/model/BaseDataJobDeployment.java
@@ -25,6 +25,8 @@ public abstract class BaseDataJobDeployment {
 
   private String pythonVersion;
 
+  private String vdkVersion;
+
   private String gitCommitSha;
 
   private String schedule;
diff --git a/projects/control-service/projects/pipelines_control_service/src/main/resources/db/migration/V20231124124300__add_column_vdk_version_to_desired_data_job_deployment_table.sql b/projects/control-service/projects/pipelines_control_service/src/main/resources/db/migration/V20231124124300__add_column_vdk_version_to_desired_data_job_deployment_table.sql
new file mode 100644
index 0000000000..da0b820f36
--- /dev/null
+++ b/projects/control-service/projects/pipelines_control_service/src/main/resources/db/migration/V20231124124300__add_column_vdk_version_to_desired_data_job_deployment_table.sql
@@ -0,0 +1,2 @@
+alter table if exists desired_data_job_deployment
+    add column if not exists vdk_version varchar;
diff --git a/projects/control-service/projects/pipelines_control_service/src/main/resources/db/migration/V20231124140200__add_column_vdk_version_to_actual_data_job_deployment_table.sql b/projects/control-service/projects/pipelines_control_service/src/main/resources/db/migration/V20231124140200__add_column_vdk_version_to_actual_data_job_deployment_table.sql
new file mode 100644
index 0000000000..5a01b6ace7
--- /dev/null
+++ b/projects/control-service/projects/pipelines_control_service/src/main/resources/db/migration/V20231124140200__add_column_vdk_version_to_actual_data_job_deployment_table.sql
@@ -0,0 +1,2 @@
+alter table if exists actual_data_job_deployment
+    add column if not exists vdk_version varchar;
diff --git a/projects/control-service/projects/pipelines_control_service/src/test/java/com/vmware/taurus/service/JobImageDeployerTest.java b/projects/control-service/projects/pipelines_control_service/src/test/java/com/vmware/taurus/service/JobImageDeployerTest.java
index 899635026c..9bfd311831 100644
--- a/projects/control-service/projects/pipelines_control_service/src/test/java/com/vmware/taurus/service/JobImageDeployerTest.java
+++ b/projects/control-service/projects/pipelines_control_service/src/test/java/com/vmware/taurus/service/JobImageDeployerTest.java
@@ -83,6 +83,7 @@ public void testScheduleLabelsAnnotations() {
         .thenReturn(new KubernetesService.Resources("1.5", "200"));
     Mockito.when(kubernetesResources.dataJobInitContainerRequests())
         .thenReturn(new KubernetesService.Resources("1.5", "200"));
+    Mockito.when(supportedPythonVersions.getVdkImage("3.9")).thenReturn("release");
 
     var annotationCaptor = ArgumentCaptor.forClass(Map.class);
     var labelCaptor = ArgumentCaptor.forClass(Map.class);
diff --git a/projects/control-service/projects/pipelines_control_service/src/test/java/com/vmware/taurus/service/deploy/DeploymentServiceTest.java b/projects/control-service/projects/pipelines_control_service/src/test/java/com/vmware/taurus/service/deploy/DeploymentServiceTest.java
index 8a470f1d75..0ce0333aed 100644
--- a/projects/control-service/projects/pipelines_control_service/src/test/java/com/vmware/taurus/service/deploy/DeploymentServiceTest.java
+++ b/projects/control-service/projects/pipelines_control_service/src/test/java/com/vmware/taurus/service/deploy/DeploymentServiceTest.java
@@ -142,6 +142,7 @@ public void setUp() {
 
     when(kubernetesService.readCronJob(TEST_CRONJOB_NAME))
         .thenReturn(Optional.of(TestUtils.getJobDeploymentStatus()));
+    when(supportedPythonVersions.getVdkImage(any())).thenReturn("release");
   }
 
   @Test
diff --git a/projects/control-service/projects/pipelines_control_service/src/test/java/com/vmware/taurus/service/deploy/JobImageDeployerTest.java b/projects/control-service/projects/pipelines_control_service/src/test/java/com/vmware/taurus/service/deploy/JobImageDeployerTest.java
index f8c7852804..c2362cb319 100644
--- a/projects/control-service/projects/pipelines_control_service/src/test/java/com/vmware/taurus/service/deploy/JobImageDeployerTest.java
+++ b/projects/control-service/projects/pipelines_control_service/src/test/java/com/vmware/taurus/service/deploy/JobImageDeployerTest.java
@@ -94,7 +94,8 @@ public void setUp() {
     when(defaultConfigurations.dataJobLimits())
         .thenReturn(new KubernetesService.Resources("2000m", "1G"));
 
-    when(supportedPythonVersions.getDefaultVdkImage()).thenReturn("release");
+    when(supportedPythonVersions.getDefaultVdkImage()).thenReturn("domain:5000/name:release");
+    when(supportedPythonVersions.getVdkImage(any())).thenReturn("domain:5000/name:release");
 
     JobConfig jobConfig = new JobConfig();
     jobConfig.setSchedule(TEST_JOB_SCHEDULE);
diff --git a/projects/control-service/projects/pipelines_control_service/src/test/java/com/vmware/taurus/service/deploy/SupportedPythonVersionsTest.java b/projects/control-service/projects/pipelines_control_service/src/test/java/com/vmware/taurus/service/deploy/SupportedPythonVersionsTest.java
index bf41250425..3d595ef420 100644
--- a/projects/control-service/projects/pipelines_control_service/src/test/java/com/vmware/taurus/service/deploy/SupportedPythonVersionsTest.java
+++ b/projects/control-service/projects/pipelines_control_service/src/test/java/com/vmware/taurus/service/deploy/SupportedPythonVersionsTest.java
@@ -231,7 +231,7 @@ public void getVdkImage_defaultImage() {
         supportedPythonVersions, SUPPORTED_PYTHON_VERSIONS, supportedVersions);
     ReflectionTestUtils.setField(supportedPythonVersions, DEFAULT_PYTHON_VERSION, "3.7");
 
-    final String defaultVdkImage = "test_vdk_image_3.7";
+    final String defaultVdkImage = "domain:5000/name:test_vdk_image_3.7";
 
     Assertions.assertEquals(defaultVdkImage, supportedPythonVersions.getVdkImage("3.11"));
   }
@@ -240,17 +240,31 @@ public void getVdkImage_defaultImage() {
   public void getVdkImage_multipleSupportedVersions() {
     var supportedVersions = generateSupportedPythonVersionsConf();
 
-    final String resultVdkImg = "test_vdk_image_3.8";
+    final String resultVdkImg = "domain:5000/name:test_vdk_image_3.8";
     ReflectionTestUtils.setField(
         supportedPythonVersions, SUPPORTED_PYTHON_VERSIONS, supportedVersions);
 
     Assertions.assertEquals(resultVdkImg, supportedPythonVersions.getVdkImage("3.8"));
   }
 
+  @Test
+  public void replaceVdkVersionInImage_replaceTheImage() {
+    var supportedVersions = generateSupportedPythonVersionsConf();
+    ReflectionTestUtils.setField(
+        supportedPythonVersions, SUPPORTED_PYTHON_VERSIONS, supportedVersions);
+    ReflectionTestUtils.setField(supportedPythonVersions, DEFAULT_PYTHON_VERSION, "3.8");
+    String resImage = "domain:5000/name:replaced_vdk_image";
+
+    Assertions.assertEquals(
+        resImage, supportedPythonVersions.replaceVdkVersionInImage("replaced_vdk_image"));
+  }
+
   private static Map<String, Map<String, String>> generateSupportedPythonVersionsConf() {
     return Map.of(
-        "3.7", Map.of(BASE_IMAGE, "python:3.7-slim", VDK_IMAGE, "test_vdk_image_3.7"),
-        "3.8", Map.of(BASE_IMAGE, "python:3.8-slim", VDK_IMAGE, "test_vdk_image_3.8"),
+        "3.7",
+            Map.of(BASE_IMAGE, "python:3.7-slim", VDK_IMAGE, "domain:5000/name:test_vdk_image_3.7"),
+        "3.8",
+            Map.of(BASE_IMAGE, "python:3.8-slim", VDK_IMAGE, "domain:5000/name:test_vdk_image_3.8"),
         "3.9", Map.of(BASE_IMAGE, "python:3.9-slim", VDK_IMAGE, "test_vdk_image_3.9"));
   }
 }