diff --git a/agent/conf/agent.properties b/agent/conf/agent.properties
index 0dc5b8211e0d..fcc5823059c8 100644
--- a/agent/conf/agent.properties
+++ b/agent/conf/agent.properties
@@ -457,3 +457,19 @@ iscsi.session.cleanup.enabled=false
 
 # Instance conversion VIRT_V2V_TMPDIR env var
 #convert.instance.env.virtv2v.tmpdir=
+
+# LIBGUESTFS backend to use for VMware to KVM conversion via VDDK (default: direct)
+#libguestfs.backend=direct
+
+# Path to the VDDK library directory for VMware to KVM conversion via VDDK,
+# passed to virt-v2v as -io vddk-libdir=<path>
+#vddk.lib.dir=
+
+# Ordered VDDK transport preference for VMware to KVM conversion via VDDK, passed as
+# -io vddk-transports=<value> to virt-v2v. Example: nbd:nbdssl
+#vddk.transports=
+
+# Optional vCenter SHA1 thumbprint for VMware to KVM conversion via VDDK, passed as
+# -io vddk-thumbprint=<value>. If unset, CloudStack computes it on the KVM host via openssl.
+#vddk.thumbprint=
+
diff --git a/agent/src/main/java/com/cloud/agent/properties/AgentProperties.java b/agent/src/main/java/com/cloud/agent/properties/AgentProperties.java
index 1561f0d5cfbc..1022e30aeab9 100644
--- a/agent/src/main/java/com/cloud/agent/properties/AgentProperties.java
+++ b/agent/src/main/java/com/cloud/agent/properties/AgentProperties.java
@@ -808,6 +808,37 @@ public Property<Integer> getWorkers() {
      */
     public static final Property<String> CONVERT_ENV_VIRTV2V_TMPDIR = new Property<>("convert.instance.env.virtv2v.tmpdir", null, String.class);
 
+    /**
+     * Path to the VDDK library directory on the KVM conversion host, used when converting VMs from VMware to KVM via VDDK.
+     * This directory is passed to virt-v2v as <code>-io vddk-libdir=&lt;path&gt;</code>.
+     * Data type: String.<br>
+     * Default value: <code>null</code>
+     */
+    public static final Property<String> VDDK_LIB_DIR = new Property<>("vddk.lib.dir", null, String.class);
+
+    /**
+     * Value for the LIBGUESTFS_BACKEND env var used during VMware to KVM conversion via VDDK.
+     * Data type: String.<br>
+     * Default value: <code>direct</code>
+     */
+    public static final Property<String> LIBGUESTFS_BACKEND = new Property<>("libguestfs.backend", "direct", String.class);
+
+    /**
+     * Ordered list of VDDK transports for virt-v2v, passed as <code>-io vddk-transports=&lt;value&gt;</code>.
+     * Example: <code>nbd:nbdssl</code>.
+     * Data type: String.<br>
+     * Default value: <code>null</code>
+     */
+    public static final Property<String> VDDK_TRANSPORTS = new Property<>("vddk.transports", null, String.class);
+
+    /**
+     * vCenter TLS certificate thumbprint used by virt-v2v VDDK mode, passed as <code>-io vddk-thumbprint=&lt;value&gt;</code>.
+     * If unset, the KVM host computes it at runtime from the vCenter endpoint.
+     * Data type: String.<br>
+     * Default value: <code>null</code>
+     */
+    public static final Property<String> VDDK_THUMBPRINT = new Property<>("vddk.thumbprint", null, String.class);
+
     /**
      * BGP controll CIDR
      * Data type: String.<br>
diff --git a/api/src/main/java/com/cloud/agent/api/to/RemoteInstanceTO.java b/api/src/main/java/com/cloud/agent/api/to/RemoteInstanceTO.java
index 18737c584b34..7daeb9649177 100644
--- a/api/src/main/java/com/cloud/agent/api/to/RemoteInstanceTO.java
+++ b/api/src/main/java/com/cloud/agent/api/to/RemoteInstanceTO.java
@@ -36,13 +36,17 @@ public class RemoteInstanceTO implements Serializable {
     private String vcenterPassword;
     private String vcenterHost;
     private String datacenterName;
+    private String clusterName;
+    private String hostName;
 
     public RemoteInstanceTO() {
     }
 
-    public RemoteInstanceTO(String instanceName) {
+    public RemoteInstanceTO(String instanceName, String clusterName, String hostName) {
         this.hypervisorType = Hypervisor.HypervisorType.VMware;
         this.instanceName = instanceName;
+        this.clusterName = clusterName;
+        this.hostName = hostName;
     }
 
     public RemoteInstanceTO(String instanceName, String instancePath, String vcenterHost, String vcenterUsername, String vcenterPassword, String datacenterName) {
@@ -55,6 +59,12 @@ public RemoteInstanceTO(String instanceName, String instancePath, String vcenter
         this.datacenterName = datacenterName;
     }
 
+    public RemoteInstanceTO(String instanceName, String instancePath, String vcenterHost, String vcenterUsername, String vcenterPassword, String datacenterName, String clusterName, String hostName) {
+        this(instanceName, instancePath, vcenterHost, vcenterUsername, vcenterPassword, datacenterName);
+        this.clusterName = clusterName;
+        this.hostName = hostName;
+    }
+
     public Hypervisor.HypervisorType getHypervisorType() {
         return this.hypervisorType;
     }
@@ -82,4 +92,12 @@ public String getVcenterHost() {
     public String getDatacenterName() {
         return datacenterName;
     }
+
+    public String getClusterName() {
+        return clusterName;
+    }
+
+    public String getHostName() {
+        return hostName;
+    }
 }
diff --git a/api/src/main/java/com/cloud/host/Host.java b/api/src/main/java/com/cloud/host/Host.java
index 9c011bac3190..5684c2647981 100644
--- a/api/src/main/java/com/cloud/host/Host.java
+++ b/api/src/main/java/com/cloud/host/Host.java
@@ -57,6 +57,7 @@ public static String[] toStrings(Host.Type... types) {
     String HOST_UEFI_ENABLE = "host.uefi.enable";
     String HOST_VOLUME_ENCRYPTION = "host.volume.encryption";
     String HOST_INSTANCE_CONVERSION = "host.instance.conversion";
+    String HOST_VDDK_SUPPORT = "host.vddk.support";
     String HOST_OVFTOOL_VERSION = "host.ovftool.version";
     String HOST_VIRTV2V_VERSION = "host.virtv2v.version";
     String HOST_SSH_PORT = "host.ssh.port";
diff --git a/api/src/main/java/org/apache/cloudstack/api/ApiConstants.java b/api/src/main/java/org/apache/cloudstack/api/ApiConstants.java
index 944b111eb70c..aa529d35679d 100644
--- a/api/src/main/java/org/apache/cloudstack/api/ApiConstants.java
+++ b/api/src/main/java/org/apache/cloudstack/api/ApiConstants.java
@@ -621,6 +621,7 @@ public class ApiConstants {
     public static final String USER_CONFIGURABLE = "userconfigurable";
     public static final String USER_SECURITY_GROUP_LIST = "usersecuritygrouplist";
     public static final String USER_SECRET_KEY = "usersecretkey";
+    public static final String USE_VDDK = "usevddk";
     public static final String USE_VIRTUAL_NETWORK = "usevirtualnetwork";
     public static final String USE_VIRTUAL_ROUTER_IP_RESOLVER = "userouteripresolver";
     public static final String UPDATE_IN_SEQUENCE = "updateinsequence";
diff --git a/api/src/main/java/org/apache/cloudstack/api/command/admin/vm/ImportVmCmd.java b/api/src/main/java/org/apache/cloudstack/api/command/admin/vm/ImportVmCmd.java
index 50ccfbd69c5e..dfefca7249a9 100644
--- a/api/src/main/java/org/apache/cloudstack/api/command/admin/vm/ImportVmCmd.java
+++ b/api/src/main/java/org/apache/cloudstack/api/command/admin/vm/ImportVmCmd.java
@@ -179,6 +179,14 @@ public class ImportVmCmd extends ImportUnmanagedInstanceCmd {
             description = "(only for importing VMs from VMware to KVM) optional - the ID of the guest OS for the imported VM.")
     private Long guestOsId;
 
+    @Parameter(name = ApiConstants.USE_VDDK,
+            type = CommandType.BOOLEAN,
+            since = "4.22.1",
+            description = "(only for importing VMs from VMware to KVM) optional - if true, uses VDDK on the KVM conversion host for converting the VM. " +
+                    "This parameter is mutually exclusive with " + ApiConstants.FORCE_MS_TO_IMPORT_VM_FILES + ".")
+    private Boolean useVddk;
+
+
     /////////////////////////////////////////////////////
     /////////////////// Accessors ///////////////////////
     /////////////////////////////////////////////////////
@@ -255,6 +263,10 @@ public Long getStoragePoolId() {
         return storagePoolId;
     }
 
+    public boolean getUseVddk() {
+        return BooleanUtils.toBooleanDefaultIfNull(useVddk, false);
+    }
+
     public String getTmpPath() {
         return tmpPath;
     }
diff --git a/core/src/main/java/com/cloud/agent/api/ConvertInstanceCommand.java b/core/src/main/java/com/cloud/agent/api/ConvertInstanceCommand.java
index 24336747ccf4..721173c20884 100644
--- a/core/src/main/java/com/cloud/agent/api/ConvertInstanceCommand.java
+++ b/core/src/main/java/com/cloud/agent/api/ConvertInstanceCommand.java
@@ -31,6 +31,11 @@ public class ConvertInstanceCommand extends Command {
     private boolean exportOvfToConversionLocation;
     private int threadsCountToExportOvf = 0;
     private String extraParams;
+    private boolean useVddk;
+    private String libguestfsBackend;
+    private String vddkLibDir;
+    private String vddkTransports;
+    private String vddkThumbprint;
 
     public ConvertInstanceCommand() {
     }
@@ -90,6 +95,46 @@ public void setExtraParams(String extraParams) {
         this.extraParams = extraParams;
     }
 
+    public boolean isUseVddk() {
+        return useVddk;
+    }
+
+    public void setUseVddk(boolean useVddk) {
+        this.useVddk = useVddk;
+    }
+
+    public String getLibguestfsBackend() {
+        return libguestfsBackend;
+    }
+
+    public void setLibguestfsBackend(String libguestfsBackend) {
+        this.libguestfsBackend = libguestfsBackend;
+    }
+
+    public String getVddkLibDir() {
+        return vddkLibDir;
+    }
+
+    public void setVddkLibDir(String vddkLibDir) {
+        this.vddkLibDir = vddkLibDir;
+    }
+
+    public String getVddkTransports() {
+        return vddkTransports;
+    }
+
+    public void setVddkTransports(String vddkTransports) {
+        this.vddkTransports = vddkTransports;
+    }
+
+    public String getVddkThumbprint() {
+        return vddkThumbprint;
+    }
+
+    public void setVddkThumbprint(String vddkThumbprint) {
+        this.vddkThumbprint = vddkThumbprint;
+    }
+
     @Override
     public boolean executeInSequence() {
         return false;
diff --git a/engine/orchestration/src/main/java/com/cloud/agent/manager/AgentManagerImpl.java b/engine/orchestration/src/main/java/com/cloud/agent/manager/AgentManagerImpl.java
index 9e85be87db34..2d8970a2846b 100644
--- a/engine/orchestration/src/main/java/com/cloud/agent/manager/AgentManagerImpl.java
+++ b/engine/orchestration/src/main/java/com/cloud/agent/manager/AgentManagerImpl.java
@@ -805,8 +805,9 @@ protected AgentAttache notifyMonitorsOfConnection(final AgentAttache attache, fi
                 String uefiEnabled = detailsMap.get(Host.HOST_UEFI_ENABLE);
                 String virtv2vVersion = detailsMap.get(Host.HOST_VIRTV2V_VERSION);
                 String ovftoolVersion = detailsMap.get(Host.HOST_OVFTOOL_VERSION);
+                String vddkSupport = detailsMap.get(Host.HOST_VDDK_SUPPORT);
                 logger.debug("Got HOST_UEFI_ENABLE [{}] for host [{}]:", uefiEnabled, host);
-                if (ObjectUtils.anyNotNull(uefiEnabled, virtv2vVersion, ovftoolVersion)) {
+                if (ObjectUtils.anyNotNull(uefiEnabled, virtv2vVersion, ovftoolVersion, vddkSupport)) {
                     _hostDao.loadDetails(host);
                     boolean updateNeeded = false;
                     if (StringUtils.isNotBlank(uefiEnabled) && !uefiEnabled.equals(host.getDetails().get(Host.HOST_UEFI_ENABLE))) {
@@ -821,6 +822,10 @@ protected AgentAttache notifyMonitorsOfConnection(final AgentAttache attache, fi
                         host.getDetails().put(Host.HOST_OVFTOOL_VERSION, ovftoolVersion);
                         updateNeeded = true;
                     }
+                    if (StringUtils.isNotBlank(vddkSupport) && !vddkSupport.equals(host.getDetails().get(Host.HOST_VDDK_SUPPORT))) {
+                        host.getDetails().put(Host.HOST_VDDK_SUPPORT, vddkSupport);
+                        updateNeeded = true;
+                    }
                     if (updateNeeded) {
                         _hostDao.saveDetails(host);
                     }
diff --git a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/LibvirtComputingResource.java b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/LibvirtComputingResource.java
index b561cedd0183..54d4145f0af2 100644
--- a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/LibvirtComputingResource.java
+++ b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/LibvirtComputingResource.java
@@ -18,6 +18,7 @@
 
 import static com.cloud.host.Host.HOST_INSTANCE_CONVERSION;
 import static com.cloud.host.Host.HOST_OVFTOOL_VERSION;
+import static com.cloud.host.Host.HOST_VDDK_SUPPORT;
 import static com.cloud.host.Host.HOST_VIRTV2V_VERSION;
 import static com.cloud.host.Host.HOST_VOLUME_ENCRYPTION;
 import static org.apache.cloudstack.utils.linux.KVMHostInfo.isHostS390x;
@@ -883,10 +884,15 @@ protected enum HealthCheckResult {
 
     private boolean convertInstanceVerboseMode = false;
     private Map<String, String> convertInstanceEnv = null;
+    private String vddkLibDir = null;
+    private String libguestfsBackend = "direct";
     protected boolean dpdkSupport = false;
     protected String dpdkOvsPath;
     protected String directDownloadTemporaryDownloadPath;
     protected String cachePath;
+    private String vddkTransports = null;
+    private String vddkThumbprint = null;
+    private String detectedPasswordFileOption = null;
     protected String javaTempDir = System.getProperty("java.io.tmpdir");
 
     private String getEndIpFromStartIp(final String startIp, final int numIps) {
@@ -951,6 +957,22 @@ public Map<String, String> getConvertInstanceEnv() {
         return convertInstanceEnv;
     }
 
+    public String getVddkLibDir() {
+        return vddkLibDir;
+    }
+
+    public String getLibguestfsBackend() {
+        return libguestfsBackend;
+    }
+
+    public String getVddkTransports() {
+        return vddkTransports;
+    }
+
+    public String getVddkThumbprint() {
+        return vddkThumbprint;
+    }
+
     /**
      * Defines resource's public and private network interface according to what is configured in agent.properties.
      */
@@ -1156,6 +1178,21 @@ public boolean configure(final String name, final Map<String, Object> params) th
 
         setConvertInstanceEnv(convertEnvTmpDir, convertEnvVirtv2vTmpDir);
 
+        vddkLibDir = AgentPropertiesFileHandler.getPropertyValue(AgentProperties.VDDK_LIB_DIR);
+        libguestfsBackend = StringUtils.defaultIfBlank(
+                AgentPropertiesFileHandler.getPropertyValue(AgentProperties.LIBGUESTFS_BACKEND), "direct");
+        vddkTransports = StringUtils.trimToNull(
+                AgentPropertiesFileHandler.getPropertyValue(AgentProperties.VDDK_TRANSPORTS));
+        vddkThumbprint = StringUtils.trimToNull(
+                AgentPropertiesFileHandler.getPropertyValue(AgentProperties.VDDK_THUMBPRINT));
+
+        detectedPasswordFileOption = detectPasswordFileOption();
+        if (StringUtils.isNotBlank(detectedPasswordFileOption)) {
+            LOGGER.info("Detected virt-v2v password option: {}", detectedPasswordFileOption);
+        } else {
+            LOGGER.warn("Could not detect virt-v2v password option, VDDK conversions may fail");
+        }
+
         pool = (String)params.get("pool");
         if (pool == null) {
             pool = "/root";
@@ -4217,6 +4254,7 @@ public StartupCommand[] initialize() {
         cmd.setHostTags(getHostTags());
         boolean instanceConversionSupported = hostSupportsInstanceConversion();
         cmd.getHostDetails().put(HOST_INSTANCE_CONVERSION, String.valueOf(instanceConversionSupported));
+        cmd.getHostDetails().put(HOST_VDDK_SUPPORT, String.valueOf(hostSupportsVddk()));
         if (instanceConversionSupported) {
             cmd.getHostDetails().put(HOST_VIRTV2V_VERSION, getHostVirtV2vVersion());
         }
@@ -5932,6 +5970,10 @@ public boolean hostSupportsInstanceConversion() {
         return exitValue == 0;
     }
 
+    public boolean hostSupportsVddk() {
+        return hostSupportsInstanceConversion() && StringUtils.isNotBlank(vddkLibDir) && new File(vddkLibDir).isDirectory();
+    }
+
     public boolean hostSupportsWindowsGuestConversion() {
         if (isUbuntuOrDebianHost()) {
             int exitValue = Script.runSimpleBashScriptForExitValue(UBUNTU_WINDOWS_GUEST_CONVERSION_SUPPORTED_CHECK_CMD);
@@ -5946,6 +5988,40 @@ public boolean hostSupportsOvfExport() {
         return exitValue == 0;
     }
 
+    /**
+     * Detect which password option virt-v2v supports by examining its --help output
+     * @return "-ip" if supported (virt-v2v >= 2.8.1), "--password-file" if older version, or null if detection fails
+     */
+    protected String detectPasswordFileOption() {
+        try {
+            ProcessBuilder pb = new ProcessBuilder("virt-v2v", "--help");
+            Process process = pb.start();
+
+            String output = new String(process.getInputStream().readAllBytes());
+            process.waitFor();
+
+            if (output.contains("-ip <filename>")) {
+                return "-ip";
+            } else if (output.contains("--password-file")) {
+                return "--password-file";
+            } else {
+                LOGGER.error("virt-v2v does not support -ip or --password-file");
+                return null;
+            }
+        } catch (Exception e) {
+            LOGGER.error("Failed to detect virt-v2v password option: {}", e.getMessage());
+            return null;
+        }
+    }
+
+    /**
+     * Get the detected password file option for virt-v2v
+     * @return the password option ("-ip" or "--password-file") or null if not detected
+     */
+    public String getDetectedPasswordFileOption() {
+        return detectedPasswordFileOption;
+    }
+
     public String getHostVirtV2vVersion() {
         if (!hostSupportsInstanceConversion()) {
             return "";
diff --git a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtConvertInstanceCommandWrapper.java b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtConvertInstanceCommandWrapper.java
index 66a5f5dd7d20..223eba6edcfa 100644
--- a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtConvertInstanceCommandWrapper.java
+++ b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtConvertInstanceCommandWrapper.java
@@ -20,10 +20,15 @@
 
 import java.net.URLEncoder;
 import java.nio.charset.Charset;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Locale;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 import org.apache.cloudstack.storage.to.PrimaryDataStoreTO;
 import org.apache.commons.collections4.MapUtils;
@@ -51,6 +56,7 @@ public class LibvirtConvertInstanceCommandWrapper extends CommandWrapper<Convert
 
     private static final List<Hypervisor.HypervisorType> supportedInstanceConvertSourceHypervisors =
             List.of(Hypervisor.HypervisorType.VMware);
+    private static final Pattern SHA1_FINGERPRINT_PATTERN = Pattern.compile("(?i)(?:SHA1\\s+)?Fingerprint\\s*=\\s*([0-9A-F:]+)");
 
     @Override
     public Answer execute(ConvertInstanceCommand cmd, LibvirtComputingResource serverResource) {
@@ -61,7 +67,8 @@ public Answer execute(ConvertInstanceCommand cmd, LibvirtComputingResource serve
         DataStoreTO conversionTemporaryLocation = cmd.getConversionTemporaryLocation();
         long timeout = (long) cmd.getWait() * 1000;
         String extraParams = cmd.getExtraParams();
-        String originalVMName = cmd.getOriginalVMName(); // For logging purposes, as the sourceInstance may have been cloned
+        boolean useVddk = cmd.isUseVddk();
+        String originalVMName = cmd.getOriginalVMName();
 
         if (cmd.getCheckConversionSupport() && !serverResource.hostSupportsInstanceConversion()) {
             String msg = String.format("Cannot convert the instance %s from VMware as the virt-v2v binary is not found. " +
@@ -84,61 +91,78 @@ public Answer execute(ConvertInstanceCommand cmd, LibvirtComputingResource serve
         logger.info(String.format("(%s) Attempting to convert the instance %s from %s to KVM",
                 originalVMName, sourceInstanceName, sourceHypervisorType));
         final String temporaryConvertPath = temporaryStoragePool.getLocalPath();
+        final String temporaryConvertUuid = UUID.randomUUID().toString();
+        boolean verboseModeEnabled = serverResource.isConvertInstanceVerboseModeEnabled();
 
-        String ovfTemplateDirOnConversionLocation;
-        String sourceOVFDirPath;
+        boolean cleanupSecondaryStorage = false;
         boolean ovfExported = false;
-        if (cmd.getExportOvfToConversionLocation()) {
-            String exportInstanceOVAUrl = getExportInstanceOVAUrl(sourceInstance, originalVMName);
-            if (StringUtils.isBlank(exportInstanceOVAUrl)) {
-                String err = String.format("Couldn't export OVA for the VM %s, due to empty url", sourceInstanceName);
-                logger.error(String.format("(%s) %s", originalVMName, err));
-                return new Answer(cmd, false, err);
-            }
+        String ovfTemplateDirOnConversionLocation = null;
 
-            int noOfThreads = cmd.getThreadsCountToExportOvf();
-            if (noOfThreads > 1 && !serverResource.ovfExportToolSupportsParallelThreads()) {
-                noOfThreads = 0;
-            }
-            ovfTemplateDirOnConversionLocation = UUID.randomUUID().toString();
-            temporaryStoragePool.createFolder(ovfTemplateDirOnConversionLocation);
-            sourceOVFDirPath = String.format("%s/%s/", temporaryConvertPath, ovfTemplateDirOnConversionLocation);
-            ovfExported = exportOVAFromVMOnVcenter(exportInstanceOVAUrl, sourceOVFDirPath, noOfThreads, originalVMName, timeout);
-            if (!ovfExported) {
-                String err = String.format("Export OVA for the VM %s failed", sourceInstanceName);
-                logger.error(String.format("(%s) %s", originalVMName, err));
-                return new Answer(cmd, false, err);
-            }
-            sourceOVFDirPath = String.format("%s%s/", sourceOVFDirPath, sourceInstanceName);
-        } else {
-            ovfTemplateDirOnConversionLocation = cmd.getTemplateDirOnConversionLocation();
-            sourceOVFDirPath = String.format("%s/%s/", temporaryConvertPath, ovfTemplateDirOnConversionLocation);
-        }
+        try {
+            boolean result;
+            if (useVddk) {
+                logger.info("({}) Using VDDK-based conversion (direct from VMware)", originalVMName);
+                String vddkLibDir = resolveVddkSetting(cmd.getVddkLibDir(), serverResource.getVddkLibDir());
+                if (StringUtils.isBlank(vddkLibDir)) {
+                    String err = String.format("VDDK lib dir is not configured on the host. " +
+                            "Set '%s' in agent.properties to use VDDK-based conversion.", "vddk.lib.dir");
+                    logger.error("({}) {}", originalVMName, err);
+                    return new Answer(cmd, false, err);
+                }
+                String libguestfsBackend = StringUtils.defaultIfBlank(resolveVddkSetting(cmd.getLibguestfsBackend(), serverResource.getLibguestfsBackend()), "direct");
+                String vddkTransports = resolveVddkSetting(cmd.getVddkTransports(), serverResource.getVddkTransports());
+                String configuredVddkThumbprint = resolveVddkSetting(cmd.getVddkThumbprint(), serverResource.getVddkThumbprint());
+                String passwordOption = serverResource.getDetectedPasswordFileOption();
+                result = performInstanceConversionVddk(sourceInstance, originalVMName, temporaryConvertPath,
+                        vddkLibDir, libguestfsBackend, vddkTransports, configuredVddkThumbprint,
+                        timeout, verboseModeEnabled, extraParams, temporaryConvertUuid, passwordOption);
+            } else {
+                logger.info("({}) Using OVF-based conversion (export + local convert)", originalVMName);
+                String sourceOVFDirPath;
+                if (cmd.getExportOvfToConversionLocation()) {
+                    String exportInstanceOVAUrl = getExportInstanceOVAUrl(sourceInstance, originalVMName);
 
-        logger.info(String.format("(%s) Attempting to convert the OVF %s of the instance %s from %s to KVM",
-                originalVMName, ovfTemplateDirOnConversionLocation, sourceInstanceName, sourceHypervisorType));
+                    if (StringUtils.isBlank(exportInstanceOVAUrl)) {
+                        String err = String.format("Couldn't export OVA for the VM %s, due to empty url", sourceInstanceName);
+                        logger.error("({}) {}", originalVMName, err);
+                        return new Answer(cmd, false, err);
+                    }
 
-        final String temporaryConvertUuid = UUID.randomUUID().toString();
-        boolean verboseModeEnabled = serverResource.isConvertInstanceVerboseModeEnabled();
+                    int noOfThreads = cmd.getThreadsCountToExportOvf();
+                    if (noOfThreads > 1 && !serverResource.ovfExportToolSupportsParallelThreads()) {
+                        noOfThreads = 0;
+                    }
+                    ovfTemplateDirOnConversionLocation = UUID.randomUUID().toString();
+                    temporaryStoragePool.createFolder(ovfTemplateDirOnConversionLocation);
+                    sourceOVFDirPath = String.format("%s/%s/", temporaryConvertPath, ovfTemplateDirOnConversionLocation);
+                    ovfExported = exportOVAFromVMOnVcenter(exportInstanceOVAUrl, sourceOVFDirPath, noOfThreads, originalVMName, timeout);
+
+                    if (!ovfExported) {
+                        String err = String.format("Export OVA for the VM %s failed", sourceInstanceName);
+                        logger.error("({}) {}", originalVMName, err);
+                        return new Answer(cmd, false, err);
+                    }
+                    sourceOVFDirPath = String.format("%s%s/", sourceOVFDirPath, sourceInstanceName);
+                } else {
+                    ovfTemplateDirOnConversionLocation = cmd.getTemplateDirOnConversionLocation();
+                    sourceOVFDirPath = String.format("%s/%s/", temporaryConvertPath, ovfTemplateDirOnConversionLocation);
+                }
+
+                result = performInstanceConversion(originalVMName, sourceOVFDirPath, temporaryConvertPath, temporaryConvertUuid,
+                        timeout, verboseModeEnabled, extraParams, serverResource);
+            }
 
-        boolean cleanupSecondaryStorage = false;
-        try {
-            boolean result = performInstanceConversion(originalVMName, sourceOVFDirPath, temporaryConvertPath, temporaryConvertUuid,
-                    timeout, verboseModeEnabled, extraParams, serverResource);
             if (!result) {
-                String err = String.format(
-                        "The virt-v2v conversion for the OVF %s failed. Please check the agent logs " +
-                                "for the virt-v2v output. Please try on a different kvm host which " +
-                                "has a different virt-v2v version.",
-                        ovfTemplateDirOnConversionLocation);
-                logger.error(String.format("(%s) %s", originalVMName, err));
+                String err = String.format("Instance conversion failed for VM %s. Please check virt-v2v logs.", sourceInstanceName);
+                logger.error("({}) {}", originalVMName, err);
                 return new Answer(cmd, false, err);
             }
+
             return new ConvertInstanceAnswer(cmd, temporaryConvertUuid);
+
         } catch (Exception e) {
-            String error = String.format("Error converting instance %s from %s, due to: %s",
-                    sourceInstanceName, sourceHypervisorType, e.getMessage());
-            logger.error(String.format("(%s) %s", originalVMName, error), e);
+            String error = String.format("Error converting instance %s from %s, due to: %s", sourceInstanceName, sourceHypervisorType, e.getMessage());
+            logger.error("({}) {}", originalVMName, error, e);
             cleanupSecondaryStorage = true;
             return new Answer(cmd, false, error);
         } finally {
@@ -275,4 +299,194 @@ protected void addExtraParamsToScript(String extraParams, Script script) {
     protected String encodeUsername(String username) {
         return URLEncoder.encode(username, Charset.defaultCharset());
     }
+
+    private String resolveVddkSetting(String commandValue, String agentValue) {
+        return StringUtils.defaultIfBlank(StringUtils.trimToNull(commandValue), StringUtils.trimToNull(agentValue));
+    }
+
+    protected boolean performInstanceConversionVddk(RemoteInstanceTO vmwareInstance, String originalVMName,
+                                                    String temporaryConvertFolder, String vddkLibDir,
+                                                    String libguestfsBackend, String vddkTransports,
+                                                    String configuredVddkThumbprint,
+                                                    long timeout, boolean verboseModeEnabled, String extraParams,
+                                                    String temporaryConvertUuid, String passwordOption) {
+
+        String vcenterPassword = vmwareInstance.getVcenterPassword();
+        if (StringUtils.isBlank(vcenterPassword)) {
+            logger.error("({}) Could not determine vCenter password for {}", originalVMName, vmwareInstance.getVcenterHost());
+            return false;
+        }
+
+        String passwordFilePath = String.format("/root/v2v.pass.cloud.%s",
+                StringUtils.defaultIfBlank(vmwareInstance.getVcenterHost(), "unknown"));
+        try {
+            Files.writeString(Path.of(passwordFilePath), vcenterPassword);
+            logger.debug("({}) Written vCenter password to {}", originalVMName, passwordFilePath);
+        } catch (Exception e) {
+            logger.error("({}) Failed to write vCenter password file {}: {}", originalVMName, passwordFilePath, e.getMessage());
+            return false;
+        }
+
+        String vpxUrl = buildVpxUrl(vmwareInstance, originalVMName);
+
+        StringBuilder cmd = new StringBuilder();
+
+        String effectiveLibguestfsBackend = StringUtils.defaultIfBlank(libguestfsBackend, "direct");
+        cmd.append("export LIBGUESTFS_BACKEND=").append(effectiveLibguestfsBackend).append(" && ");
+
+        cmd.append("virt-v2v ");
+        cmd.append("--root first ");
+        cmd.append("-ic '").append(vpxUrl).append("' ");
+        if (StringUtils.isBlank(passwordOption)) {
+            logger.error("({}) Could not determine supported password file option for virt-v2v", originalVMName);
+            return false;
+        }
+
+        cmd.append(passwordOption).append(" ").append(passwordFilePath).append(" ");
+        cmd.append("-it vddk ");
+        cmd.append("-io vddk-libdir=").append(vddkLibDir).append(" ");
+        String vddkThumbprint = StringUtils.trimToNull(configuredVddkThumbprint);
+        if (StringUtils.isBlank(vddkThumbprint)) {
+            vddkThumbprint = getVcenterThumbprint(vmwareInstance.getVcenterHost(), timeout, originalVMName);
+        }
+        if (StringUtils.isBlank(vddkThumbprint)) {
+            logger.error("({}) Could not determine vCenter thumbprint for {}", originalVMName, vmwareInstance.getVcenterHost());
+            return false;
+        }
+        cmd.append("-io vddk-thumbprint=").append(vddkThumbprint).append(" ");
+        if (StringUtils.isNotBlank(vddkTransports)) {
+            cmd.append("-io vddk-transports=").append(vddkTransports).append(" ");
+        }
+        cmd.append(originalVMName).append(" ");
+        cmd.append("-o local ");
+        cmd.append("-os ").append(temporaryConvertFolder).append(" ");
+        cmd.append("-of qcow2 ");
+        cmd.append("-on ").append(temporaryConvertUuid).append(" ");
+
+        if (verboseModeEnabled) {
+            cmd.append("-v ");
+        }
+
+        if (StringUtils.isNotBlank(extraParams)) {
+            cmd.append(extraParams).append(" ");
+        }
+
+        Script script = new Script("/bin/bash", timeout, logger);
+        script.add("-c");
+        script.add(cmd.toString());
+
+        String logPrefix = String.format("(%s) virt-v2v vddk import", originalVMName);
+        OutputInterpreter.LineByLineOutputLogger outputLogger =
+                new OutputInterpreter.LineByLineOutputLogger(logger, logPrefix);
+
+        logger.info("({}) Starting virt-v2v VDDK conversion", originalVMName);
+        script.execute(outputLogger);
+
+        int exitValue = script.getExitValue();
+        if (exitValue != 0) {
+            logger.error("({}) virt-v2v failed with exit code {}", originalVMName, exitValue);
+        }
+        try {
+            Files.deleteIfExists(Path.of(passwordFilePath));
+            logger.debug("({}) Deleted password file {}", originalVMName, passwordFilePath);
+        } catch (Exception e) {
+            logger.warn("({}) Failed to delete password file {}: {}", originalVMName, passwordFilePath, e.getMessage());
+        }
+
+        return exitValue == 0;
+    }
+
+
+    protected String getVcenterThumbprint(String vcenterHost, long timeout, String originalVMName) {
+        if (StringUtils.isBlank(vcenterHost)) {
+            return null;
+        }
+
+        String endpoint = String.format("%s:443", vcenterHost);
+        String command = String.format("openssl s_client -connect '%s' </dev/null 2>/dev/null | " +
+                "openssl x509 -fingerprint -sha1 -noout", endpoint);
+
+        Script script = new Script("/bin/bash", timeout, logger);
+        script.add("-c");
+        script.add(command);
+
+        OutputInterpreter.AllLinesParser parser = new OutputInterpreter.AllLinesParser();
+        script.execute(parser);
+
+        String output = parser.getLines();
+        if (script.getExitValue() != 0) {
+            logger.error("({}) Failed to fetch vCenter thumbprint for {}", originalVMName, vcenterHost);
+            return null;
+        }
+
+        String thumbprint = extractSha1Fingerprint(output);
+        if (StringUtils.isBlank(thumbprint)) {
+            logger.error("({}) Failed to parse vCenter thumbprint from output for {}", originalVMName, vcenterHost);
+            return null;
+        }
+        return thumbprint;
+    }
+
+    private String extractSha1Fingerprint(String output) {
+        String parsedOutput = StringUtils.trimToEmpty(output);
+        if (StringUtils.isBlank(parsedOutput)) {
+            return null;
+        }
+
+        for (String line : parsedOutput.split("\\R")) {
+            String trimmedLine = StringUtils.trimToEmpty(line);
+            if (StringUtils.isBlank(trimmedLine)) {
+                continue;
+            }
+
+            Matcher matcher = SHA1_FINGERPRINT_PATTERN.matcher(trimmedLine);
+            if (matcher.find()) {
+                return matcher.group(1).toUpperCase(Locale.ROOT);
+            }
+
+            // Fallback for raw fingerprint-only output.
+            if (trimmedLine.matches("(?i)[0-9a-f]{2}(:[0-9a-f]{2})+")) {
+                return trimmedLine.toUpperCase(Locale.ROOT);
+            }
+        }
+        return null;
+    }
+
+    /**
+     * Build vpx:// URL for virt-v2v
+     *
+     * Format:
+     * vpx://user@vcenter/DC/cluster/host?no_verify=1
+     */
+    private String buildVpxUrl(RemoteInstanceTO vmwareInstance, String originalVMName) {
+
+        String vcenter = vmwareInstance.getVcenterHost();
+        String username = vmwareInstance.getVcenterUsername();
+        String datacenter = vmwareInstance.getDatacenterName();
+        String cluster = vmwareInstance.getClusterName();
+        String host = vmwareInstance.getHostName();
+
+        String encodedUsername = encodeUsername(username);
+
+        StringBuilder url = new StringBuilder();
+        url.append("vpx://")
+                .append(encodedUsername)
+                .append("@")
+                .append(vcenter)
+                .append("/")
+                .append(datacenter);
+
+        if (StringUtils.isNotBlank(cluster)) {
+            url.append("/").append(cluster);
+        }
+
+        if (StringUtils.isNotBlank(host)) {
+            url.append("/").append(host);
+        }
+
+        url.append("?no_verify=1");
+
+        logger.info("({}) Using VPX URL: {}", originalVMName, url);
+        return url.toString();
+    }
 }
diff --git a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtReadyCommandWrapper.java b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtReadyCommandWrapper.java
index e74923b281f6..2ff3b1bb51cc 100644
--- a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtReadyCommandWrapper.java
+++ b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtReadyCommandWrapper.java
@@ -50,6 +50,7 @@ public Answer execute(final ReadyCommand command, final LibvirtComputingResource
         if (libvirtComputingResource.hostSupportsInstanceConversion()) {
             hostDetails.put(Host.HOST_VIRTV2V_VERSION, libvirtComputingResource.getHostVirtV2vVersion());
         }
+        hostDetails.put(Host.HOST_VDDK_SUPPORT, Boolean.toString(libvirtComputingResource.hostSupportsVddk()));
 
         if (libvirtComputingResource.hostSupportsOvfExport()) {
             hostDetails.put(Host.HOST_OVFTOOL_VERSION, libvirtComputingResource.getHostOvfToolVersion());
diff --git a/plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtConvertInstanceCommandWrapperTest.java b/plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtConvertInstanceCommandWrapperTest.java
index 4d55ac2bc73a..05117c140c7b 100644
--- a/plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtConvertInstanceCommandWrapperTest.java
+++ b/plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtConvertInstanceCommandWrapperTest.java
@@ -18,6 +18,8 @@
 //
 package com.cloud.hypervisor.kvm.resource.wrapper;
 
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.util.List;
 import java.util.UUID;
 
@@ -189,4 +191,123 @@ public void testAddExtraParamsToScriptDifferentArgs() {
         Mockito.verify(script).add("-x");
         Mockito.verify(script).add("-v");
     }
+
+    @Test
+    public void testPerformInstanceConversionVddkUsesConfiguredLibguestfsBackend() {
+        RemoteInstanceTO remoteInstanceTO = Mockito.mock(RemoteInstanceTO.class);
+        Mockito.when(remoteInstanceTO.getVcenterHost()).thenReturn("vcenter.local");
+        Mockito.when(remoteInstanceTO.getVcenterUsername()).thenReturn("administrator@vsphere.local");
+        Mockito.when(remoteInstanceTO.getVcenterPassword()).thenReturn("secret");
+        Mockito.when(remoteInstanceTO.getDatacenterName()).thenReturn("dc1");
+        Mockito.when(remoteInstanceTO.getClusterName()).thenReturn("cluster1");
+        Mockito.when(remoteInstanceTO.getHostName()).thenReturn("host1");
+        Mockito.doReturn("28:19:A6:1C:90:ED:46:D7:1C:86:BC:F6:13:52:F0:B9:19:81:0D:81")
+                .when(convertInstanceCommandWrapper).getVcenterThumbprint(Mockito.anyString(), Mockito.anyLong(), Mockito.anyString());
+
+        try (MockedStatic<Files> filesMock = Mockito.mockStatic(Files.class);
+             MockedConstruction<Script> ignored = Mockito.mockConstruction(Script.class, (mock, context) -> {
+                 Mockito.when(mock.execute(Mockito.any())).thenReturn("");
+                 Mockito.when(mock.getExitValue()).thenReturn(0);
+             })) {
+            Path passwordFilePath = Path.of("/root/v2v.pass.cloud.vcenter.local");
+            filesMock.when(() -> Files.writeString(passwordFilePath, "secret")).thenReturn(passwordFilePath);
+            filesMock.when(() -> Files.deleteIfExists(passwordFilePath)).thenReturn(true);
+
+            boolean result = convertInstanceCommandWrapper.performInstanceConversionVddk(
+                    remoteInstanceTO, vmName, "/tmp/convert", "/opt/vddk", "libvirt", null, null, 1000L, false, null, "tmp-uuid", "-ip");
+
+            Assert.assertTrue(result);
+            Script scriptMock = ignored.constructed().get(0);
+            Mockito.verify(scriptMock).add("-c");
+            Mockito.verify(scriptMock).add(Mockito.contains("export LIBGUESTFS_BACKEND=libvirt &&"));
+            Mockito.verify(scriptMock).add(Mockito.contains("-ip /root/v2v.pass.cloud.vcenter.local "));
+            Mockito.verify(scriptMock).add(Mockito.contains(" -on tmp-uuid "));
+            Mockito.verify(scriptMock).add(Mockito.contains("-io vddk-thumbprint=28:19:A6:1C:90:ED:46:D7:1C:86:BC:F6:13:52:F0:B9:19:81:0D:81 "));
+        }
+    }
+
+    @Test
+    public void testPerformInstanceConversionVddkUsesConfiguredTransportsOrder() {
+        RemoteInstanceTO remoteInstanceTO = Mockito.mock(RemoteInstanceTO.class);
+        Mockito.when(remoteInstanceTO.getVcenterHost()).thenReturn("vcenter.local");
+        Mockito.when(remoteInstanceTO.getVcenterUsername()).thenReturn("administrator@vsphere.local");
+        Mockito.when(remoteInstanceTO.getVcenterPassword()).thenReturn("secret");
+        Mockito.when(remoteInstanceTO.getDatacenterName()).thenReturn("dc1");
+        Mockito.when(remoteInstanceTO.getClusterName()).thenReturn("cluster1");
+        Mockito.when(remoteInstanceTO.getHostName()).thenReturn("host1");
+        Mockito.doReturn("28:19:A6:1C:90:ED:46:D7:1C:86:BC:F6:13:52:F0:B9:19:81:0D:81")
+                .when(convertInstanceCommandWrapper).getVcenterThumbprint(Mockito.anyString(), Mockito.anyLong(), Mockito.anyString());
+
+        try (MockedStatic<Files> filesMock = Mockito.mockStatic(Files.class);
+             MockedConstruction<Script> ignored = Mockito.mockConstruction(Script.class, (mock, context) -> {
+                 Mockito.when(mock.execute(Mockito.any())).thenReturn("");
+                 Mockito.when(mock.getExitValue()).thenReturn(0);
+             })) {
+            Path passwordFilePath = Path.of("/root/v2v.pass.cloud.vcenter.local");
+            filesMock.when(() -> Files.writeString(passwordFilePath, "secret")).thenReturn(passwordFilePath);
+            filesMock.when(() -> Files.deleteIfExists(passwordFilePath)).thenReturn(true);
+
+            boolean result = convertInstanceCommandWrapper.performInstanceConversionVddk(
+                    remoteInstanceTO, vmName, "/tmp/convert", "/opt/vddk", "direct", "nbd:nbdssl", null, 1000L, false, null, "tmp-uuid", "-ip");
+
+            Assert.assertTrue(result);
+            Script scriptMock = ignored.constructed().get(0);
+            Mockito.verify(scriptMock).add(Mockito.contains("-io vddk-transports=nbd:nbdssl "));
+        }
+    }
+
+    @Test
+    public void testPerformInstanceConversionVddkFailsWhenThumbprintUnavailable() {
+        RemoteInstanceTO remoteInstanceTO = Mockito.mock(RemoteInstanceTO.class);
+        Mockito.when(remoteInstanceTO.getVcenterHost()).thenReturn("vcenter.local");
+        Mockito.when(remoteInstanceTO.getVcenterUsername()).thenReturn("administrator@vsphere.local");
+        Mockito.when(remoteInstanceTO.getVcenterPassword()).thenReturn("secret");
+        Mockito.when(remoteInstanceTO.getDatacenterName()).thenReturn("dc1");
+        Mockito.when(remoteInstanceTO.getClusterName()).thenReturn("cluster1");
+        Mockito.when(remoteInstanceTO.getHostName()).thenReturn("host1");
+        Mockito.doReturn(null)
+                .when(convertInstanceCommandWrapper).getVcenterThumbprint(Mockito.anyString(), Mockito.anyLong(), Mockito.anyString());
+
+        try (MockedStatic<Files> filesMock = Mockito.mockStatic(Files.class)) {
+            Path passwordFilePath = Path.of("/root/v2v.pass.cloud.vcenter.local");
+            filesMock.when(() -> Files.writeString(passwordFilePath, "secret")).thenReturn(passwordFilePath);
+            filesMock.when(() -> Files.deleteIfExists(passwordFilePath)).thenReturn(true);
+
+            boolean result = convertInstanceCommandWrapper.performInstanceConversionVddk(
+                    remoteInstanceTO, vmName, "/tmp/convert", "/opt/vddk", "direct", null, null, 1000L, false, null, "tmp-uuid", "-ip");
+
+            Assert.assertFalse(result);
+        }
+    }
+
+    @Test
+    public void testPerformInstanceConversionVddkUsesConfiguredThumbprintFromAgentProperty() {
+        RemoteInstanceTO remoteInstanceTO = Mockito.mock(RemoteInstanceTO.class);
+        Mockito.when(remoteInstanceTO.getVcenterHost()).thenReturn("vcenter.local");
+        Mockito.when(remoteInstanceTO.getVcenterUsername()).thenReturn("administrator@vsphere.local");
+        Mockito.when(remoteInstanceTO.getVcenterPassword()).thenReturn("secret");
+        Mockito.when(remoteInstanceTO.getDatacenterName()).thenReturn("dc1");
+        Mockito.when(remoteInstanceTO.getClusterName()).thenReturn("cluster1");
+        Mockito.when(remoteInstanceTO.getHostName()).thenReturn("host1");
+
+        try (MockedStatic<Files> filesMock = Mockito.mockStatic(Files.class);
+             MockedConstruction<Script> ignored = Mockito.mockConstruction(Script.class, (mock, context) -> {
+                 Mockito.when(mock.execute(Mockito.any())).thenReturn("");
+                 Mockito.when(mock.getExitValue()).thenReturn(0);
+             })) {
+            Path passwordFilePath = Path.of("/root/v2v.pass.cloud.vcenter.local");
+            filesMock.when(() -> Files.writeString(passwordFilePath, "secret")).thenReturn(passwordFilePath);
+            filesMock.when(() -> Files.deleteIfExists(passwordFilePath)).thenReturn(true);
+
+            boolean result = convertInstanceCommandWrapper.performInstanceConversionVddk(
+                    remoteInstanceTO, vmName, "/tmp/convert", "/opt/vddk", "direct", null,
+                    "AA:BB:CC:DD:EE", 1000L, false, null, "tmp-uuid", "-ip");
+
+            Assert.assertTrue(result);
+            Script scriptMock = ignored.constructed().get(0);
+            Mockito.verify(scriptMock).add(Mockito.contains("-io vddk-thumbprint=AA:BB:CC:DD:EE "));
+            Mockito.verify(convertInstanceCommandWrapper, Mockito.never())
+                    .getVcenterThumbprint(Mockito.anyString(), Mockito.anyLong(), Mockito.anyString());
+        }
+    }
 }
diff --git a/server/src/main/java/org/apache/cloudstack/vm/UnmanagedVMsManagerImpl.java b/server/src/main/java/org/apache/cloudstack/vm/UnmanagedVMsManagerImpl.java
index e043791c6bfe..cc03b55c398b 100644
--- a/server/src/main/java/org/apache/cloudstack/vm/UnmanagedVMsManagerImpl.java
+++ b/server/src/main/java/org/apache/cloudstack/vm/UnmanagedVMsManagerImpl.java
@@ -208,6 +208,10 @@ public class UnmanagedVMsManagerImpl implements UnmanagedVMsManager {
     private static final List<Storage.StoragePoolType> forceConvertToPoolAllowedTypes =
             Arrays.asList(Storage.StoragePoolType.NetworkFilesystem, Storage.StoragePoolType.Filesystem,
                     Storage.StoragePoolType.SharedMountPoint);
+    private static final String DETAIL_LIBGUESTFS_BACKEND = "libguestfs.backend";
+    private static final String DETAIL_VDDK_LIB_DIR = "vddk.lib.dir";
+    private static final String DETAIL_VDDK_TRANSPORTS = "vddk.transports";
+    private static final String DETAIL_VDDK_THUMBPRINT = "vddk.thumbprint";
 
     ConfigKey<Boolean> ConvertVmwareInstanceToKvmExtraParamsAllowed = new ConfigKey<>(Boolean.class,
             "convert.vmware.instance.to.kvm.extra.params.allowed",
@@ -1651,6 +1655,8 @@ protected UserVm importUnmanagedInstanceFromVmwareToKvm(DataCenter zone, Cluster
         String extraParams = cmd.getExtraParams();
         boolean forceConvertToPool = cmd.getForceConvertToPool();
         Long guestOsId = cmd.getGuestOsId();
+        boolean forceMsToImportVmFiles = Boolean.TRUE.equals(cmd.getForceMsToImportVmFiles());
+        boolean useVddk = cmd.getUseVddk();
 
         if ((existingVcenterId == null && vcenter == null) || (existingVcenterId != null && vcenter != null)) {
             throw new ServerApiException(ApiErrorCode.PARAM_ERROR,
@@ -1662,9 +1668,16 @@ protected UserVm importUnmanagedInstanceFromVmwareToKvm(DataCenter zone, Cluster
         }
 
         checkConversionStoragePool(convertStoragePoolId, forceConvertToPool);
+        validateSelectedConversionStoragePoolForVddk(useVddk, convertStoragePoolId, serviceOffering, dataDiskOfferingMap);
 
         checkExtraParamsAllowed(extraParams);
 
+        if (forceMsToImportVmFiles && useVddk) {
+            throw new ServerApiException(ApiErrorCode.PARAM_ERROR,
+                    String.format("Parameters %s and %s are mutually exclusive",
+                            ApiConstants.FORCE_MS_TO_IMPORT_VM_FILES, ApiConstants.USE_VDDK));
+        }
+
         if (existingVcenterId != null) {
             VmwareDatacenterVO existingDC = vmwareDatacenterDao.findById(existingVcenterId);
             if (existingDC == null) {
@@ -1713,7 +1726,7 @@ protected UserVm importUnmanagedInstanceFromVmwareToKvm(DataCenter zone, Cluster
 
             checkNetworkingBeforeConvertingVmwareInstance(zone, owner, displayName, hostName, sourceVMwareInstance, nicNetworkMap, nicIpAddressMap, forced);
             UnmanagedInstanceTO convertedInstance;
-            if (cmd.getForceMsToImportVmFiles() || !conversionSupportAnswer.isOvfExportSupported()) {
+            if (!useVddk && (forceMsToImportVmFiles || !conversionSupportAnswer.isOvfExportSupported())) {
                 // Uses MS for OVF export to temporary conversion location
                 int noOfThreads = UnmanagedVMsManager.ThreadsOnMSToImportVMwareVMFiles.value();
                 importVmTasksManager.updateImportVMTaskStep(importVMTask, zone, owner, convertHost, importHost, null, ConvertingInstance);
@@ -1730,7 +1743,7 @@ protected UserVm importUnmanagedInstanceFromVmwareToKvm(DataCenter zone, Cluster
                 convertedInstance = convertVmwareInstanceToKVMAfterExportingOVFToConvertLocation(
                         sourceVMName, sourceVMwareInstance, convertHost, importHost,
                         convertStoragePools, serviceOffering, dataDiskOfferingMap,
-                        temporaryConvertLocation, vcenter, username, password, datacenterName, forceConvertToPool, extraParams);
+                        temporaryConvertLocation, vcenter, username, password, datacenterName, forceConvertToPool, extraParams, useVddk, details);
             }
 
             sanitizeConvertedInstance(convertedInstance, sourceVMwareInstance);
@@ -1785,6 +1798,41 @@ protected void checkConversionStoragePool(Long convertStoragePoolId, boolean for
         }
     }
 
+    protected void validateSelectedConversionStoragePoolForVddk(boolean useVddk, Long convertStoragePoolId,
+                                                                ServiceOfferingVO serviceOffering, Map<String, Long> dataDiskOfferingMap) {
+        if (!useVddk || convertStoragePoolId == null) {
+            return;
+        }
+
+        StoragePoolVO selectedStoragePool = primaryDataStoreDao.findById(convertStoragePoolId);
+        if (selectedStoragePool == null) {
+            return;
+        }
+
+        if (serviceOffering.getDiskOfferingId() != null) {
+            DiskOfferingVO rootDiskOffering = diskOfferingDao.findById(serviceOffering.getDiskOfferingId());
+            if (rootDiskOffering == null) {
+                throw new InvalidParameterValueException(String.format("Cannot find disk offering with ID %s that belongs to the service offering %s",
+                        serviceOffering.getDiskOfferingId(), serviceOffering.getName()));
+            }
+            if (!volumeApiService.doesStoragePoolSupportDiskOffering(selectedStoragePool, rootDiskOffering)) {
+                throw new InvalidParameterValueException("Using VDDK, multiple storage pools cannot be used if a storage pool is selected for conversion");
+            }
+        }
+
+        if (MapUtils.isNotEmpty(dataDiskOfferingMap)) {
+            for (Long diskOfferingId : dataDiskOfferingMap.values()) {
+                DiskOfferingVO diskOffering = diskOfferingDao.findById(diskOfferingId);
+                if (diskOffering == null) {
+                    throw new InvalidParameterValueException(String.format("Cannot find disk offering with ID %s", diskOfferingId));
+                }
+                if (!volumeApiService.doesStoragePoolSupportDiskOffering(selectedStoragePool, diskOffering)) {
+                    throw new InvalidParameterValueException("Using VDDK, multiple storage pools cannot be used if a storage pool is selected for conversion");
+                }
+            }
+        }
+    }
+
     private void checkNetworkingBeforeConvertingVmwareInstance(DataCenter zone, Account owner, String displayName,
                                                                String hostName, UnmanagedInstanceTO sourceVMwareInstance,
                                                                Map<String, Long> nicNetworkMap,
@@ -2032,7 +2080,7 @@ private UnmanagedInstanceTO convertVmwareInstanceToKVMWithOVFOnConvertLocation(
         logger.debug("Delegating the conversion of instance {} from VMware to KVM to the host {} using OVF {} on conversion datastore",
                 sourceVM, convertHost, ovfTemplateDirConvertLocation);
 
-        RemoteInstanceTO remoteInstanceTO = new RemoteInstanceTO(sourceVM);
+        RemoteInstanceTO remoteInstanceTO = new RemoteInstanceTO(sourceVM, sourceVMwareInstance.getClusterName(), sourceVMwareInstance.getHostName());
         List<String> destinationStoragePools = selectInstanceConversionStoragePools(convertStoragePools, sourceVMwareInstance.getDisks(), serviceOffering, dataDiskOfferingMap);
         ConvertInstanceCommand cmd = new ConvertInstanceCommand(remoteInstanceTO,
                 Hypervisor.HypervisorType.KVM, temporaryConvertLocation,
@@ -2052,10 +2100,11 @@ private UnmanagedInstanceTO convertVmwareInstanceToKVMAfterExportingOVFToConvert
             HostVO importHost, List<StoragePoolVO> convertStoragePools,
             ServiceOfferingVO serviceOffering, Map<String, Long> dataDiskOfferingMap,
             DataStoreTO temporaryConvertLocation, String vcenterHost, String vcenterUsername,
-            String vcenterPassword, String datacenterName, boolean forceConvertToPool, String extraParams) {
+            String vcenterPassword, String datacenterName, boolean forceConvertToPool, String extraParams,
+            boolean useVddk, Map<String, String> details) {
         logger.debug("Delegating the conversion of instance {} from VMware to KVM to the host {} after OVF export through ovftool", sourceVM, convertHost);
 
-        RemoteInstanceTO remoteInstanceTO = new RemoteInstanceTO(sourceVMwareInstance.getName(), sourceVMwareInstance.getPath(), vcenterHost, vcenterUsername, vcenterPassword, datacenterName);
+        RemoteInstanceTO remoteInstanceTO = new RemoteInstanceTO(sourceVMwareInstance.getName(), sourceVMwareInstance.getPath(), vcenterHost, vcenterUsername, vcenterPassword, datacenterName, sourceVMwareInstance.getClusterName(), sourceVMwareInstance.getHostName());
         List<String> destinationStoragePools = selectInstanceConversionStoragePools(convertStoragePools, sourceVMwareInstance.getDisks(), serviceOffering, dataDiskOfferingMap);
         ConvertInstanceCommand cmd = new ConvertInstanceCommand(remoteInstanceTO,
                 Hypervisor.HypervisorType.KVM, temporaryConvertLocation, null, false, true, sourceVM);
@@ -2070,10 +2119,23 @@ private UnmanagedInstanceTO convertVmwareInstanceToKVMAfterExportingOVFToConvert
         if (StringUtils.isNotBlank(extraParams)) {
             cmd.setExtraParams(extraParams);
         }
+        cmd.setUseVddk(useVddk);
+        applyVddkOverridesFromDetails(cmd, details);
         return convertAndImportToKVM(cmd, convertHost, importHost, sourceVM,
                 remoteInstanceTO, destinationStoragePools, temporaryConvertLocation, forceConvertToPool);
     }
 
+    private void applyVddkOverridesFromDetails(ConvertInstanceCommand cmd, Map<String, String> details) {
+        if (MapUtils.isEmpty(details)) {
+            return;
+        }
+
+        cmd.setLibguestfsBackend(StringUtils.trimToNull(details.get(DETAIL_LIBGUESTFS_BACKEND)));
+        cmd.setVddkLibDir(StringUtils.trimToNull(details.get(DETAIL_VDDK_LIB_DIR)));
+        cmd.setVddkTransports(StringUtils.trimToNull(details.get(DETAIL_VDDK_TRANSPORTS)));
+        cmd.setVddkThumbprint(StringUtils.trimToNull(details.get(DETAIL_VDDK_THUMBPRINT)));
+    }
+
     private UnmanagedInstanceTO convertAndImportToKVM(ConvertInstanceCommand convertInstanceCommand, HostVO convertHost, HostVO importHost,
                                                       String sourceVM,
                                                       RemoteInstanceTO remoteInstanceTO,
diff --git a/server/src/test/java/org/apache/cloudstack/vm/UnmanagedVMsManagerImplTest.java b/server/src/test/java/org/apache/cloudstack/vm/UnmanagedVMsManagerImplTest.java
index 98e6388f3d6b..184702428ad2 100644
--- a/server/src/test/java/org/apache/cloudstack/vm/UnmanagedVMsManagerImplTest.java
+++ b/server/src/test/java/org/apache/cloudstack/vm/UnmanagedVMsManagerImplTest.java
@@ -714,7 +714,17 @@ public void testGetTemplateForImportInstanceDefaultTemplate() {
     }
 
     private enum VcenterParameter {
-        EXISTING, EXTERNAL, BOTH, NONE, EXISTING_INVALID, AGENT_UNAVAILABLE, CONVERT_FAILURE
+        EXISTING,
+        EXTERNAL,
+        BOTH,
+        NONE,
+        EXISTING_INVALID,
+        AGENT_UNAVAILABLE,
+        CONVERT_FAILURE,
+        FORCE_MS_AND_USE_VDDK,
+        USE_VDDK_OVF_UNSUPPORTED,
+        USE_VDDK_OVF_SUPPORTED,
+        USE_VDDK_DETAILS_OVERRIDES
     }
 
     private void baseTestImportVmFromVmwareToKvm(VcenterParameter vcenterParameter, boolean selectConvertHost,
@@ -751,6 +761,35 @@ private void baseTestImportVmFromVmwareToKvm(VcenterParameter vcenterParameter,
         when(importVmCmd.getConvertInstanceHostId()).thenReturn(null);
         when(importVmCmd.getImportInstanceHostId()).thenReturn(null);
         when(importVmCmd.getConvertStoragePoolId()).thenReturn(null);
+        when(importVmCmd.getExistingVcenterId()).thenReturn(null);
+        when(importVmCmd.getVcenter()).thenReturn(null);
+        when(importVmCmd.getDatacenterName()).thenReturn(null);
+        when(importVmCmd.getUsername()).thenReturn(null);
+        when(importVmCmd.getPassword()).thenReturn(null);
+        when(importVmCmd.getDetails()).thenReturn(new HashMap<>());
+
+        boolean forceMsToImportVmFiles = false;
+        boolean useVddk = false;
+        boolean ovfExportSupported = false;
+        if (VcenterParameter.FORCE_MS_AND_USE_VDDK == vcenterParameter) {
+            forceMsToImportVmFiles = true;
+            useVddk = true;
+        } else if (VcenterParameter.USE_VDDK_OVF_UNSUPPORTED == vcenterParameter) {
+            useVddk = true;
+        } else if (VcenterParameter.USE_VDDK_OVF_SUPPORTED == vcenterParameter) {
+            useVddk = true;
+            ovfExportSupported = true;
+        } else if (VcenterParameter.USE_VDDK_DETAILS_OVERRIDES == vcenterParameter) {
+            useVddk = true;
+            ovfExportSupported = true;
+            when(importVmCmd.getDetails()).thenReturn(Map.of(
+                    "libguestfs.backend", "libvirt",
+                    "vddk.lib.dir", "/opt/vmware-vddk/override",
+                    "vddk.transports", "nbd:nbdssl",
+                    "vddk.thumbprint", "AA:BB:CC:DD:EE"));
+        }
+        when(importVmCmd.getForceMsToImportVmFiles()).thenReturn(forceMsToImportVmFiles);
+        when(importVmCmd.getUseVddk()).thenReturn(useVddk);
 
         NetworkVO networkVO = Mockito.mock(NetworkVO.class);
         when(networkVO.getGuestType()).thenReturn(Network.GuestType.L2);
@@ -812,11 +851,6 @@ private void baseTestImportVmFromVmwareToKvm(VcenterParameter vcenterParameter,
             when(datacenterVO.getPassword()).thenReturn(password);
             when(importVmCmd.getExistingVcenterId()).thenReturn(existingDatacenterId);
             when(vmwareDatacenterDao.findById(existingDatacenterId)).thenReturn(datacenterVO);
-        } else if (VcenterParameter.EXTERNAL == vcenterParameter) {
-            when(importVmCmd.getVcenter()).thenReturn(vcenterHost);
-            when(importVmCmd.getDatacenterName()).thenReturn(datacenter);
-            when(importVmCmd.getUsername()).thenReturn(username);
-            when(importVmCmd.getPassword()).thenReturn(password);
         }
 
         if (VcenterParameter.BOTH == vcenterParameter) {
@@ -830,8 +864,20 @@ private void baseTestImportVmFromVmwareToKvm(VcenterParameter vcenterParameter,
             when(vmwareDatacenterDao.findById(existingDatacenterId)).thenReturn(null);
         }
 
+        if (VcenterParameter.FORCE_MS_AND_USE_VDDK == vcenterParameter
+                || VcenterParameter.USE_VDDK_OVF_UNSUPPORTED == vcenterParameter
+                || VcenterParameter.USE_VDDK_OVF_SUPPORTED == vcenterParameter
+                || VcenterParameter.USE_VDDK_DETAILS_OVERRIDES == vcenterParameter) {
+            Mockito.doReturn((Long) null).when(importVmCmd).getExistingVcenterId();
+            Mockito.doReturn(vcenterHost).when(importVmCmd).getVcenter();
+            Mockito.doReturn(datacenter).when(importVmCmd).getDatacenterName();
+            Mockito.doReturn(username).when(importVmCmd).getUsername();
+            Mockito.doReturn(password).when(importVmCmd).getPassword();
+        }
+
         CheckConvertInstanceAnswer checkConvertInstanceAnswer = mock(CheckConvertInstanceAnswer.class);
         when(checkConvertInstanceAnswer.getResult()).thenReturn(vcenterParameter != VcenterParameter.CONVERT_FAILURE);
+        when(checkConvertInstanceAnswer.isOvfExportSupported()).thenReturn(ovfExportSupported);
         if (VcenterParameter.AGENT_UNAVAILABLE != vcenterParameter) {
             when(agentManager.send(Mockito.eq(convertHostId), Mockito.any(CheckConvertInstanceCommand.class))).thenReturn(checkConvertInstanceAnswer);
         }
@@ -853,9 +899,30 @@ private void baseTestImportVmFromVmwareToKvm(VcenterParameter vcenterParameter,
         try (MockedStatic<UsageEventUtils> ignored = Mockito.mockStatic(UsageEventUtils.class)) {
             unmanagedVMsManager.importVm(importVmCmd);
             verify(vmwareGuru).getHypervisorVMOutOfBandAndCloneIfRequired(Mockito.eq(host), Mockito.eq(vmName), anyMap());
-            verify(vmwareGuru).createVMTemplateOutOfBand(Mockito.eq(host), Mockito.eq(vmName), anyMap(), any(DataStoreTO.class), anyInt());
+            if (VcenterParameter.USE_VDDK_OVF_SUPPORTED == vcenterParameter) {
+                verify(vmwareGuru, Mockito.never()).createVMTemplateOutOfBand(anyString(), anyString(), anyMap(), any(DataStoreTO.class), anyInt());
+                verify(agentManager).send(Mockito.eq(convertHostId), Mockito.<com.cloud.agent.api.Command>argThat(command ->
+                        command instanceof ConvertInstanceCommand && ((ConvertInstanceCommand) command).isUseVddk()));
+                verify(vmwareGuru, Mockito.never()).removeVMTemplateOutOfBand(any(DataStoreTO.class), anyString());
+            } else if (VcenterParameter.USE_VDDK_DETAILS_OVERRIDES == vcenterParameter) {
+                verify(vmwareGuru, Mockito.never()).createVMTemplateOutOfBand(anyString(), anyString(), anyMap(), any(DataStoreTO.class), anyInt());
+                verify(agentManager).send(Mockito.eq(convertHostId), Mockito.<com.cloud.agent.api.Command>argThat(command -> {
+                    if (!(command instanceof ConvertInstanceCommand)) {
+                        return false;
+                    }
+                    ConvertInstanceCommand convertCmd = (ConvertInstanceCommand) command;
+                    return convertCmd.isUseVddk()
+                            && "libvirt".equals(convertCmd.getLibguestfsBackend())
+                            && "/opt/vmware-vddk/override".equals(convertCmd.getVddkLibDir())
+                            && "nbd:nbdssl".equals(convertCmd.getVddkTransports())
+                            && "AA:BB:CC:DD:EE".equals(convertCmd.getVddkThumbprint());
+                }));
+                verify(vmwareGuru, Mockito.never()).removeVMTemplateOutOfBand(any(DataStoreTO.class), anyString());
+            } else {
+                verify(vmwareGuru).createVMTemplateOutOfBand(Mockito.eq(host), Mockito.eq(vmName), anyMap(), any(DataStoreTO.class), anyInt());
+                verify(vmwareGuru).removeVMTemplateOutOfBand(any(DataStoreTO.class), anyString());
+            }
             verify(vmwareGuru).removeClonedHypervisorVMOutOfBand(Mockito.eq(host), Mockito.eq(vmName), anyMap());
-            verify(vmwareGuru).removeVMTemplateOutOfBand(any(DataStoreTO.class), anyString());
         }
     }
 
@@ -948,6 +1015,59 @@ public void testImportVmFromVmwareToKvmExistingVcenterConvertFailure() throws Op
         baseTestImportVmFromVmwareToKvm(VcenterParameter.CONVERT_FAILURE, false, false);
     }
 
+    @Test(expected = ServerApiException.class)
+    public void testImportVmFromVmwareToKvmForceMsMutuallyExclusiveWithUseVddk() throws OperationTimedoutException, AgentUnavailableException {
+        baseTestImportVmFromVmwareToKvm(VcenterParameter.FORCE_MS_AND_USE_VDDK, false, false);
+    }
+
+    @Test
+    public void testImportVmFromVmwareToKvmUseVddkIsPassedToConvertCommand() throws OperationTimedoutException, AgentUnavailableException {
+        baseTestImportVmFromVmwareToKvm(VcenterParameter.USE_VDDK_OVF_SUPPORTED, false, false);
+    }
+
+    @Test
+    public void testImportVmFromVmwareToKvmDetailsOverrideVddkSettings() throws OperationTimedoutException, AgentUnavailableException {
+        baseTestImportVmFromVmwareToKvm(VcenterParameter.USE_VDDK_DETAILS_OVERRIDES, false, false);
+    }
+
+    @Test(expected = InvalidParameterValueException.class)
+    public void testValidateSelectedConversionStoragePoolForVddkFailsWhenPoolDoesNotSupportDiskOfferings() {
+        long poolId = 11L;
+        StoragePoolVO selectedPool = mock(StoragePoolVO.class);
+        ServiceOfferingVO serviceOffering = mock(ServiceOfferingVO.class);
+        DiskOfferingVO rootDiskOffering = mock(DiskOfferingVO.class);
+        DiskOfferingVO dataDiskOffering = mock(DiskOfferingVO.class);
+
+        when(serviceOffering.getDiskOfferingId()).thenReturn(21L);
+        when(primaryDataStoreDao.findById(poolId)).thenReturn(selectedPool);
+        when(diskOfferingDao.findById(21L)).thenReturn(rootDiskOffering);
+        when(diskOfferingDao.findById(22L)).thenReturn(dataDiskOffering);
+        when(volumeApiService.doesStoragePoolSupportDiskOffering(selectedPool, rootDiskOffering)).thenReturn(true);
+        when(volumeApiService.doesStoragePoolSupportDiskOffering(selectedPool, dataDiskOffering)).thenReturn(false);
+
+        unmanagedVMsManager.validateSelectedConversionStoragePoolForVddk(true, poolId,
+                serviceOffering, Map.of("1000-2", 22L));
+    }
+
+    @Test
+    public void testValidateSelectedConversionStoragePoolForVddkPassesWhenPoolSupportsAllDiskOfferings() {
+        long poolId = 12L;
+        StoragePoolVO selectedPool = mock(StoragePoolVO.class);
+        ServiceOfferingVO serviceOffering = mock(ServiceOfferingVO.class);
+        DiskOfferingVO rootDiskOffering = mock(DiskOfferingVO.class);
+        DiskOfferingVO dataDiskOffering = mock(DiskOfferingVO.class);
+
+        when(serviceOffering.getDiskOfferingId()).thenReturn(31L);
+        when(primaryDataStoreDao.findById(poolId)).thenReturn(selectedPool);
+        when(diskOfferingDao.findById(31L)).thenReturn(rootDiskOffering);
+        when(diskOfferingDao.findById(32L)).thenReturn(dataDiskOffering);
+        when(volumeApiService.doesStoragePoolSupportDiskOffering(selectedPool, rootDiskOffering)).thenReturn(true);
+        when(volumeApiService.doesStoragePoolSupportDiskOffering(selectedPool, dataDiskOffering)).thenReturn(true);
+
+        unmanagedVMsManager.validateSelectedConversionStoragePoolForVddk(true, poolId,
+                serviceOffering, Map.of("1000-2", 32L));
+    }
+
     private ClusterVO getClusterForTests() {
         ClusterVO cluster = mock(ClusterVO.class);
         when(cluster.getId()).thenReturn(1L);
diff --git a/ui/public/locales/en.json b/ui/public/locales/en.json
index 6494995bf223..9bf0e9270c6e 100644
--- a/ui/public/locales/en.json
+++ b/ui/public/locales/en.json
@@ -2373,6 +2373,7 @@
 "label.user.data.policy.tooltip": "User Data linked to the Template can be overridden by User Data provided during Instance deploy. Select the override policy as required.",
 "label.user.data": "User Data",
 "label.user.data.library": "User Data Library",
+"label.use.vddk": "Use VDDK",
 "label.ssh.port": "SSH port",
 "label.sshkeypair": "New SSH key pair",
 "label.sshkeypairs": "SSH key pairs",
diff --git a/ui/src/views/tools/ImportUnmanagedInstance.vue b/ui/src/views/tools/ImportUnmanagedInstance.vue
index cbe0dc7d5a02..d54d34d754ec 100644
--- a/ui/src/views/tools/ImportUnmanagedInstance.vue
+++ b/ui/src/views/tools/ImportUnmanagedInstance.vue
@@ -152,7 +152,13 @@
                   </a-row>
                 </a-radio-group>
               </a-form-item>
-              <a-form-item name="forceconverttopool" ref="forceconverttopool" v-if="selectedVmwareVcenter">
+              <a-form-item name="usevddk" ref="usevddk" v-if="selectedVmwareVcenter">
+                <template #label>
+                  <tooltip-label :title="$t('label.use.vddk')" :tooltip="apiParams.usevddk ? apiParams.usevddk.description : ''"/>
+                </template>
+                <a-switch v-model:checked="form.usevddk" @change="onUseVddkChange" />
+              </a-form-item>
+              <a-form-item name="forceconverttopool" ref="forceconverttopool" v-if="selectedVmwareVcenter && !form.usevddk">
                 <template #label>
                   <tooltip-label :title="$t('label.force.convert.to.pool')" :tooltip="apiParams.forceconverttopool.description"/>
                 </template>
@@ -184,12 +190,13 @@
               </a-form-item>
               <a-form-item name="convertstorageoption" ref="convertstorageoption">
                 <check-box-select-pair
+                  :key="`convertstorageoption-${form.usevddk ? 'vddk' : 'default'}-${switches.forceConvertToPool ? 'pool' : 'tmp'}`"
                   layout="vertical"
                   v-if="cluster.hypervisortype === 'KVM' && selectedVmwareVcenter"
                   :resourceKey="cluster.id"
                   :selectOptions="storageOptionsForConversion"
                   :checkBoxLabel="switches.forceConvertToPool ? $t('message.select.destination.storage.instance.conversion') : $t('message.select.temporary.storage.instance.conversion')"
-                  :defaultCheckBoxValue="false"
+                  :defaultCheckBoxValue="form.usevddk"
                   :reversed="false"
                   @handle-checkselectpair-change="updateSelectedStorageOptionForConversion"
                 />
@@ -226,8 +233,7 @@
                   :placeholder="$t('label.extra')"
                 />
               </a-form-item>
-              <a-form-item name="forcemstoimportvmfiles" ref="forcemstoimportvmfiles" v-if="selectedVmwareVcenter">
-                <template #label>
+              <a-form-item name="forcemstoimportvmfiles" ref="forcemstoimportvmfiles" v-if="selectedVmwareVcenter && !form.usevddk">                <template #label>
                   <tooltip-label :title="$t('label.force.ms.to.import.vm.files')" :tooltip="apiParams.forcemstoimportvmfiles.description"/>
                 </template>
                 <a-switch v-model:checked="form.forcemstoimportvmfiles" @change="val => { switches.forceMsToImportVmFiles = val }" />
@@ -778,6 +784,7 @@ export default {
       this.formRef = ref()
       this.form = reactive({
         rootdiskid: 0,
+        usevddk: false,
         migrateallowed: this.switches.migrateAllowed,
         forced: this.switches.forced,
         forcemstoimportvmfiles: this.switches.forceMsToImportVmFiles,
@@ -1117,6 +1124,21 @@ export default {
       this.switches.forceConvertToPool = val
       this.resetStorageOptionsForConversion()
     },
+    onUseVddkChange (val) {
+      if (val) {
+        this.form.forceconverttopool = true
+        this.form.forcemstoimportvmfiles = false
+        this.switches.forceConvertToPool = true
+        this.switches.forceMsToImportVmFiles = false
+      } else {
+        this.form.forceconverttopool = false
+        this.switches.forceConvertToPool = false
+        this.selectedStorageOptionForConversion = null
+        this.selectedStoragePoolForConversion = null
+        this.showStoragePoolsForConversion = false
+      }
+      this.resetStorageOptionsForConversion()
+    },
     updateSelectedRootDisk () {
       var rootDisk = this.resource.disk[this.selectedRootDiskIndex]
       rootDisk.size = rootDisk.capacity / (1024 * 1024 * 1024)
@@ -1235,12 +1257,18 @@ export default {
           if (this.vmwareToKvmExtraParams) {
             params.extraparams = this.vmwareToKvmExtraParams
           }
-          params.forcemstoimportvmfiles = values.forcemstoimportvmfiles
-          if (values.forceconverttopool) {
+          if (values.usevddk) {
+            params.usevddk = true
+            params.forceconverttopool = true
+            params.forcemstoimportvmfiles = false
+          } else {
+            params.forcemstoimportvmfiles = values.forcemstoimportvmfiles
+          }
+          if (!values.usevddk && values.forceconverttopool) {
             params.forceconverttopool = values.forceconverttopool
           }
         }
-        var keys = ['hostname', 'domainid', 'projectid', 'account', 'migrateallowed', 'forced', 'forcemstoimportvmfiles', 'osid']
+        var keys = ['hostname', 'domainid', 'projectid', 'account', 'migrateallowed', 'forced', 'osid']
         if (this.templateType !== 'auto') {
           keys.push('templateid')
         }
@@ -1354,6 +1382,10 @@ export default {
       this.templateType = this.defaultTemplateType()
       this.updateComputeOffering(undefined)
       this.switches = {}
+      this.form.usevddk = false
+      this.form.forceconverttopool = false
+      this.form.forcemstoimportvmfiles = false
+      this.resetStorageOptionsForConversion()
     },
     closeAction () {
       this.$emit('close-action')