diff --git a/agent/conf/agent.properties b/agent/conf/agent.properties
index 0dc5b8211e0..fcc5823059c 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 1561f0d5cfb..1022e30aeab 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 class AgentProperties{
      */
     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 18737c584b3..7daeb964917 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 class RemoteInstanceTO implements Serializable {
         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 class RemoteInstanceTO implements Serializable {
     public String getDatacenterName() {
         return datacenterName;
     }
+
+    public String getClusterName() {
+        return clusterName;
+    }
+
+    public String getHostName() {
+        return hostName;
+    }
 }
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 944b111eb70..aa529d35679 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 50ccfbd69c5..dfefca7249a 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 class ImportVmCmd extends ImportUnmanagedInstanceCmd {
         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 24336747ccf..721173c2088 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 class ConvertInstanceCommand extends Command {
         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/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 b561cedd018..e2794782d66 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
@@ -883,10 +883,14 @@ public class LibvirtComputingResource extends ServerResourceBase implements Serv
 
     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;
     protected String javaTempDir = System.getProperty("java.io.tmpdir");
 
     private String getEndIpFromStartIp(final String startIp, final int numIps) {
@@ -951,6 +955,22 @@ public class LibvirtComputingResource extends ServerResourceBase implements Serv
         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 +1176,14 @@ public class LibvirtComputingResource extends ServerResourceBase implements Serv
 
         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));
+
         pool = (String)params.get("pool");
         if (pool == null) {
             pool = "/root";
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 66a5f5dd7d2..d8eb7d39aa2 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 @@ package com.cloud.hypervisor.kvm.resource.wrapper;
 
 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 class LibvirtConvertInstanceCommandWrapper extends CommandWrapper<Convert
         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,77 @@ public class LibvirtConvertInstanceCommandWrapper extends CommandWrapper<Convert
         logger.info(String.format("(%s) Attempting to convert the instance %s from %s to KVM",
                 originalVMName, sourceInstanceName, sourceHypervisorType));
         final String temporaryConvertPath = temporaryStoragePool.getLocalPath();
-
-        String ovfTemplateDirOnConversionLocation;
-        String sourceOVFDirPath;
-        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);
-            }
-
-            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);
-        }
-
-        logger.info(String.format("(%s) Attempting to convert the OVF %s of the instance %s from %s to KVM",
-                originalVMName, ovfTemplateDirOnConversionLocation, sourceInstanceName, sourceHypervisorType));
-
         final String temporaryConvertUuid = UUID.randomUUID().toString();
         boolean verboseModeEnabled = serverResource.isConvertInstanceVerboseModeEnabled();
 
         boolean cleanupSecondaryStorage = false;
+        boolean ovfExported = false;
+        String ovfTemplateDirOnConversionLocation = null;
+
         try {
-            boolean result = performInstanceConversion(originalVMName, sourceOVFDirPath, temporaryConvertPath, temporaryConvertUuid,
-                    timeout, verboseModeEnabled, extraParams, serverResource);
+            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());
+                result = performInstanceConversionVddk(sourceInstance, originalVMName, temporaryConvertPath,
+                        vddkLibDir, libguestfsBackend, vddkTransports, configuredVddkThumbprint,
+                        timeout, verboseModeEnabled, extraParams);
+            } else {
+                logger.info("({}) Using OVF-based conversion (export + local convert)", originalVMName);
+                String sourceOVFDirPath;
+                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("({}) {}", originalVMName, err);
+                        return new Answer(cmd, false, err);
+                    }
+
+                    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);
+            }
+
             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 +298,185 @@ public class LibvirtConvertInstanceCommandWrapper extends CommandWrapper<Convert
     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 vcenterPassword = vmwareInstance.getVcenterPassword();
+        if (StringUtils.isBlank(vcenterPassword)) {
+            logger.error("({}) Could not determine vCenter password for {}", originalVMName, vmwareInstance.getVcenterHost());
+            return false;
+        }
+
+        String passwordFilePath = "/root/v2v.pass.cloud";
+        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("' ");
+        cmd.append("--password-file ").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 ");
+
+        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/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 4d55ac2bc73..fc508255608 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,122 @@ public class LibvirtConvertInstanceCommandWrapperTest {
         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");
+            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);
+
+            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("--password-file /root/v2v.pass.cloud "));
+            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");
+            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);
+
+            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");
+            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);
+
+            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");
+            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);
+
+            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 e043791c6bf..06a1e82e454 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 @@ public class UnmanagedVMsManagerImpl implements UnmanagedVMsManager {
         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,
@@ -1665,6 +1671,12 @@ public class UnmanagedVMsManagerImpl implements UnmanagedVMsManager {
 
         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 +1725,7 @@ public class UnmanagedVMsManagerImpl implements UnmanagedVMsManager {
 
             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 +1742,7 @@ public class UnmanagedVMsManagerImpl implements UnmanagedVMsManager {
                 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);
@@ -2032,7 +2044,7 @@ public class UnmanagedVMsManagerImpl implements UnmanagedVMsManager {
         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 +2064,11 @@ public class UnmanagedVMsManagerImpl implements UnmanagedVMsManager {
             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 +2083,23 @@ public class UnmanagedVMsManagerImpl implements UnmanagedVMsManager {
         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 98e6388f3d6..bc1cfa85ee3 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 class UnmanagedVMsManagerImplTest {
     }
 
     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 @@ public class UnmanagedVMsManagerImplTest {
         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 @@ public class UnmanagedVMsManagerImplTest {
             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 @@ public class UnmanagedVMsManagerImplTest {
             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 @@ public class UnmanagedVMsManagerImplTest {
         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,21 @@ public class UnmanagedVMsManagerImplTest {
         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);
+    }
+
     private ClusterVO getClusterForTests() {
         ClusterVO cluster = mock(ClusterVO.class);
         when(cluster.getId()).thenReturn(1L);
diff --git a/ui/src/views/tools/ImportUnmanagedInstance.vue b/ui/src/views/tools/ImportUnmanagedInstance.vue
index cbe0dc7d5a0..776750c84f2 100644
--- a/ui/src/views/tools/ImportUnmanagedInstance.vue
+++ b/ui/src/views/tools/ImportUnmanagedInstance.vue
@@ -189,13 +189,13 @@
                   :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="true"
                   :reversed="false"
                   @handle-checkselectpair-change="updateSelectedStorageOptionForConversion"
                 />
               </a-form-item>
               <a-form-item
-                v-if="showStoragePoolsForConversion"
+                v-if="selectedVmwareVcenter && showStoragePoolsForConversion"
                 name="convertstoragepool"
                 ref="convertstoragepool"
                 :label="$t('label.storagepool')"
@@ -226,7 +226,13 @@
                   :placeholder="$t('label.extra')"
                 />
               </a-form-item>
-              <a-form-item name="forcemstoimportvmfiles" ref="forcemstoimportvmfiles" v-if="selectedVmwareVcenter">
+              <a-form-item name="usevddk" ref="usevddk" v-if="selectedVmwareVcenter">
+                <template #label>
+                  <tooltip-label title="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="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>
@@ -552,7 +558,11 @@ export default {
       memoryKey: 'memory',
       minIopsKey: 'minIops',
       maxIopsKey: 'maxIops',
-      switches: {},
+      switches: {
+        forceConvertToPool: true,
+        forceMsToImportVmFiles: false,
+        useVddk: false
+      },
       loading: false,
       kvmHostsForConversion: [],
       kvmHostsForImporting: [],
@@ -560,17 +570,14 @@ export default {
       selectedKvmHostForImporting: null,
       storageOptionsForConversion: [
         {
-          id: 'secondary',
-          name: 'Secondary Storage'
-        }, {
           id: 'primary',
           name: 'Primary Storage'
         }
       ],
       storagePoolsForConversion: [],
-      selectedStorageOptionForConversion: null,
+      selectedStorageOptionForConversion: 'primary',
       selectedStoragePoolForConversion: null,
-      showStoragePoolsForConversion: false,
+      showStoragePoolsForConversion: true,
       selectedRootDiskColumns: [
         {
           key: 'name',
@@ -782,6 +789,7 @@ export default {
         forced: this.switches.forced,
         forcemstoimportvmfiles: this.switches.forceMsToImportVmFiles,
         forceconverttopool: this.switches.forceConvertToPool,
+        usevddk: this.switches.useVddk,
         domainid: null,
         account: null,
         osid: null
@@ -805,6 +813,10 @@ export default {
       })
       this.fetchKvmHostsForConversion()
       this.fetchKvmHostsForImporting()
+      if (this.cluster.hypervisortype === 'KVM' && this.selectedVmwareVcenter) {
+        this.resetStorageOptionsForConversion()
+        this.fetchStoragePoolsForConversion()
+      }
       if (this.resource?.disk?.length > 1) {
         this.updateSelectedRootDisk()
       }
@@ -1095,6 +1107,7 @@ export default {
         this.fetchStoragePoolsForConversion()
         this.showStoragePoolsForConversion = value !== 'secondary'
       } else {
+        this.selectedStorageOptionForConversion = null
         this.showStoragePoolsForConversion = false
         this.selectedStoragePoolForConversion = null
       }
@@ -1108,6 +1121,8 @@ export default {
         id: 'primary',
         name: 'Primary Storage'
       })
+      this.selectedStorageOptionForConversion = 'primary'
+      this.showStoragePoolsForConversion = true
     },
     onSelectRootDisk (val) {
       this.selectedRootDiskIndex = val
@@ -1116,6 +1131,14 @@ export default {
     onForceConvertToPoolChange (val) {
       this.switches.forceConvertToPool = val
       this.resetStorageOptionsForConversion()
+      this.fetchStoragePoolsForConversion()
+    },
+    onUseVddkChange (val) {
+      this.switches.useVddk = val
+      if (val) {
+        this.switches.forceMsToImportVmFiles = false
+        this.form.forcemstoimportvmfiles = false
+      }
     },
     updateSelectedRootDisk () {
       var rootDisk = this.resource.disk[this.selectedRootDiskIndex]
@@ -1235,6 +1258,7 @@ export default {
           if (this.vmwareToKvmExtraParams) {
             params.extraparams = this.vmwareToKvmExtraParams
           }
+          params.usevddk = !!values.usevddk
           params.forcemstoimportvmfiles = values.forcemstoimportvmfiles
           if (values.forceconverttopool) {
             params.forceconverttopool = values.forceconverttopool
@@ -1353,7 +1377,18 @@ export default {
       }
       this.templateType = this.defaultTemplateType()
       this.updateComputeOffering(undefined)
-      this.switches = {}
+      this.switches = {
+        forceConvertToPool: true,
+        forceMsToImportVmFiles: false,
+        useVddk: false
+      }
+      this.form.forcemstoimportvmfiles = this.switches.forceMsToImportVmFiles
+      this.form.forceconverttopool = this.switches.forceConvertToPool
+      this.form.usevddk = this.switches.useVddk
+      if (this.cluster.hypervisortype === 'KVM' && this.selectedVmwareVcenter) {
+        this.resetStorageOptionsForConversion()
+        this.fetchStoragePoolsForConversion()
+      }
     },
     closeAction () {
       this.$emit('close-action')