diff --git a/core/src/org/apache/cloudstack/storage/to/PrimaryDataStoreTO.java b/core/src/org/apache/cloudstack/storage/to/PrimaryDataStoreTO.java
index 9b711bc3b3a..67ff0d71795 100644
--- a/core/src/org/apache/cloudstack/storage/to/PrimaryDataStoreTO.java
+++ b/core/src/org/apache/cloudstack/storage/to/PrimaryDataStoreTO.java
@@ -51,6 +51,7 @@ public class PrimaryDataStoreTO implements DataStoreTO {
     private final String url;
     private Map<String, String> details;
     private static final String pathSeparator = "/";
+    private Boolean fullCloneFlag;
 
     public PrimaryDataStoreTO(PrimaryDataStore dataStore) {
         this.uuid = dataStore.getUuid();
@@ -144,4 +145,12 @@ public class PrimaryDataStoreTO implements DataStoreTO {
             .append("]")
             .toString();
     }
+
+    public Boolean isFullCloneFlag() {
+        return fullCloneFlag;
+    }
+
+    public void setFullCloneFlag(Boolean fullCloneFlag) {
+        this.fullCloneFlag = fullCloneFlag;
+    }
 }
diff --git a/engine/components-api/src/com/cloud/capacity/CapacityManager.java b/engine/components-api/src/com/cloud/capacity/CapacityManager.java
index 3db6e578152..d190d78a8a2 100644
--- a/engine/components-api/src/com/cloud/capacity/CapacityManager.java
+++ b/engine/components-api/src/com/cloud/capacity/CapacityManager.java
@@ -35,6 +35,7 @@ public interface CapacityManager {
     static final String StorageCapacityDisableThresholdCK = "pool.storage.capacity.disablethreshold";
     static final String StorageOverprovisioningFactorCK = "storage.overprovisioning.factor";
     static final String StorageAllocatedCapacityDisableThresholdCK = "pool.storage.allocated.capacity.disablethreshold";
+    static final String VmwareCreateCloneFullCK = "vmware.create.full.clone";
 
     static final ConfigKey<Float> CpuOverprovisioningFactor = new ConfigKey<Float>(Float.class, CpuOverprovisioningFactorCK, "Advanced", "1.0",
         "Used for CPU overprovisioning calculation; available CPU will be (actualCpuCapacity * cpu.overprovisioning.factor)", true, ConfigKey.Scope.Cluster, null);
@@ -63,6 +64,15 @@ public interface CapacityManager {
                     true,
                     ConfigKey.Scope.Cluster,
                     null);
+    static final ConfigKey<Boolean> VmwareCreateCloneFull =
+            new ConfigKey<Boolean>(
+                    "Storage",
+                    Boolean.class,
+                    VmwareCreateCloneFullCK,
+                    "false",
+                    "If set to true, creates VMs as full clones on ESX hypervisor",
+                    true,
+                    ConfigKey.Scope.StoragePool);
 
     public boolean releaseVmCapacity(VirtualMachine vm, boolean moveFromReserved, boolean moveToReservered, Long hostId);
 
diff --git a/engine/orchestration/src/org/apache/cloudstack/engine/orchestration/VolumeOrchestrator.java b/engine/orchestration/src/org/apache/cloudstack/engine/orchestration/VolumeOrchestrator.java
index ca4ef4fd2de..32cb19b170b 100644
--- a/engine/orchestration/src/org/apache/cloudstack/engine/orchestration/VolumeOrchestrator.java
+++ b/engine/orchestration/src/org/apache/cloudstack/engine/orchestration/VolumeOrchestrator.java
@@ -69,6 +69,7 @@ import com.cloud.agent.api.to.DataTO;
 import com.cloud.agent.api.to.DiskTO;
 import com.cloud.agent.api.to.VirtualMachineTO;
 import com.cloud.agent.manager.allocator.PodAllocator;
+import com.cloud.capacity.CapacityManager;
 import com.cloud.cluster.ClusterManager;
 import com.cloud.configuration.Resource.ResourceType;
 import com.cloud.dc.DataCenter;
@@ -120,6 +121,7 @@ import com.cloud.utils.exception.CloudRuntimeException;
 import com.cloud.utils.fsm.NoTransitionException;
 import com.cloud.utils.fsm.StateMachine2;
 import com.cloud.vm.DiskProfile;
+import com.cloud.vm.UserVmCloneSettingVO;
 import com.cloud.vm.UserVmVO;
 import com.cloud.vm.VirtualMachine;
 import com.cloud.vm.VirtualMachine.State;
@@ -129,9 +131,15 @@ import com.cloud.vm.VmWorkAttachVolume;
 import com.cloud.vm.VmWorkMigrateVolume;
 import com.cloud.vm.VmWorkSerializer;
 import com.cloud.vm.VmWorkTakeVolumeSnapshot;
+import com.cloud.vm.dao.UserVmCloneSettingDao;
 import com.cloud.vm.dao.UserVmDao;
 
 public class VolumeOrchestrator extends ManagerBase implements VolumeOrchestrationService, Configurable {
+
+    public enum UserVmCloneType {
+        full, linked
+    }
+
     private static final Logger s_logger = Logger.getLogger(VolumeOrchestrator.class);
 
     @Inject
@@ -178,6 +186,8 @@ public class VolumeOrchestrator extends ManagerBase implements VolumeOrchestrati
     ClusterManager clusterManager;
     @Inject
     StorageManager storageMgr;
+    @Inject
+    protected UserVmCloneSettingDao _vmCloneSettingDao;
 
     private final StateMachine2<Volume.State, Volume.Event, Volume> _volStateMachine;
     protected List<StoragePoolAllocator> _storagePoolAllocators;
@@ -1353,6 +1363,33 @@ public class VolumeOrchestrator extends ManagerBase implements VolumeOrchestrati
             disk.setDetails(getDetails(volumeInfo, dataStore));
 
             vm.addDisk(disk);
+
+            // If hypervisor is vSphere, check for clone type setting.
+            if (vm.getHypervisorType().equals(HypervisorType.VMware)) {
+                // retrieve clone flag.
+                UserVmCloneType cloneType = UserVmCloneType.linked;
+                Boolean value = CapacityManager.VmwareCreateCloneFull.valueIn(vol.getPoolId());
+                if (value != null && value) {
+                    cloneType = UserVmCloneType.full;
+                }
+                try {
+                    UserVmCloneSettingVO cloneSettingVO = _vmCloneSettingDao.findByVmId(vm.getId());
+                    if (cloneSettingVO != null){
+                        if (! cloneSettingVO.getCloneType().equals(cloneType.toString())){
+                            cloneSettingVO.setCloneType(cloneType.toString());
+                            _vmCloneSettingDao.update(cloneSettingVO.getVmId(), cloneSettingVO);
+                        }
+                    }
+                    else {
+                        UserVmCloneSettingVO vmCloneSettingVO = new UserVmCloneSettingVO(vm.getId(), cloneType.toString());
+                        _vmCloneSettingDao.persist(vmCloneSettingVO);
+                    }
+                }
+                catch (Throwable e){
+                    s_logger.debug("[NSX_PLUGIN_LOG] ERROR: " + e.getMessage());
+                }
+            }
+
         }
     }
 
diff --git a/engine/schema/src/com/cloud/vm/UserVmCloneSettingVO.java b/engine/schema/src/com/cloud/vm/UserVmCloneSettingVO.java
index 91b4918625d..adca686db57 100644
--- a/engine/schema/src/com/cloud/vm/UserVmCloneSettingVO.java
+++ b/engine/schema/src/com/cloud/vm/UserVmCloneSettingVO.java
@@ -46,4 +46,8 @@ public class UserVmCloneSettingVO {
     public String getCloneType() {
         return this.cloneType;
     }
+
+    public void setCloneType(String cloneType) {
+        this.cloneType = cloneType;
+    }
 }
diff --git a/engine/schema/src/com/cloud/vm/dao/UserVmCloneSettingDaoImpl.java b/engine/schema/src/com/cloud/vm/dao/UserVmCloneSettingDaoImpl.java
index d76f6d46d4e..0761f56917b 100644
--- a/engine/schema/src/com/cloud/vm/dao/UserVmCloneSettingDaoImpl.java
+++ b/engine/schema/src/com/cloud/vm/dao/UserVmCloneSettingDaoImpl.java
@@ -45,7 +45,7 @@ public class UserVmCloneSettingDaoImpl extends GenericDaoBase<UserVmCloneSetting
     public void init() {
         // Initialize the search builders.
         vmIdSearch = createSearchBuilder();
-        vmIdSearch.and("vmId", vmIdSearch.entity().getCloneType(), Op.EQ);
+        vmIdSearch.and("vmId", vmIdSearch.entity().getVmId(), Op.EQ);
         vmIdSearch.done();
 
         cloneTypeSearch = createSearchBuilder();
diff --git a/engine/storage/datamotion/src/org/apache/cloudstack/storage/motion/AncientDataMotionStrategy.java b/engine/storage/datamotion/src/org/apache/cloudstack/storage/motion/AncientDataMotionStrategy.java
index 3124666db19..57e4181aaf9 100644
--- a/engine/storage/datamotion/src/org/apache/cloudstack/storage/motion/AncientDataMotionStrategy.java
+++ b/engine/storage/datamotion/src/org/apache/cloudstack/storage/motion/AncientDataMotionStrategy.java
@@ -45,6 +45,7 @@ import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
 import org.apache.cloudstack.storage.RemoteHostEndPoint;
 import org.apache.cloudstack.storage.command.CopyCommand;
 import org.apache.cloudstack.storage.image.datastore.ImageStoreEntity;
+import org.apache.cloudstack.storage.to.PrimaryDataStoreTO;
 import org.apache.log4j.Logger;
 import org.springframework.stereotype.Component;
 
@@ -56,8 +57,10 @@ import com.cloud.agent.api.to.DataStoreTO;
 import com.cloud.agent.api.to.DataTO;
 import com.cloud.agent.api.to.NfsTO;
 import com.cloud.agent.api.to.VirtualMachineTO;
+import com.cloud.capacity.CapacityManager;
 import com.cloud.configuration.Config;
 import com.cloud.host.Host;
+import com.cloud.hypervisor.Hypervisor;
 import com.cloud.storage.DataStoreRole;
 import com.cloud.storage.Storage.StoragePoolType;
 import com.cloud.storage.StoragePool;
@@ -153,7 +156,7 @@ public class AncientDataMotionStrategy implements DataMotionStrategy {
                 srcForCopy = cacheData = cacheMgr.createCacheObject(srcData, destScope);
             }
 
-            CopyCommand cmd = new CopyCommand(srcForCopy.getTO(), destData.getTO(), _primaryStorageDownloadWait, VirtualMachineManager.ExecuteInSequence.value());
+            CopyCommand cmd = new CopyCommand(srcForCopy.getTO(), addFullCloneFlagOnVMwareDest(destData.getTO()), _primaryStorageDownloadWait, VirtualMachineManager.ExecuteInSequence.value());
             EndPoint ep = destHost != null ? RemoteHostEndPoint.getHypervisorHostEndPoint(destHost) : selector.select(srcForCopy, destData);
             if (ep == null) {
                 String errMsg = "No remote endpoint to send command, check if host or ssvm is down?";
@@ -201,6 +204,23 @@ public class AncientDataMotionStrategy implements DataMotionStrategy {
         }
     }
 
+    /**
+     * Adds {@code 'vmware.create.full.clone'} value for a given primary storage, whose HV is VMware, on datastore's {@code fullCloneFlag} field
+     * @param dataTO Dest data store TO
+     * @return dataTO including fullCloneFlag, if provided
+     */
+    protected DataTO addFullCloneFlagOnVMwareDest(DataTO dataTO) {
+        if (dataTO != null && dataTO.getHypervisorType().equals(Hypervisor.HypervisorType.VMware)){
+            DataStoreTO dataStoreTO = dataTO.getDataStore();
+            if (dataStoreTO != null && dataStoreTO instanceof PrimaryDataStoreTO){
+                PrimaryDataStoreTO primaryDataStoreTO = (PrimaryDataStoreTO) dataStoreTO;
+                Boolean value = CapacityManager.VmwareCreateCloneFull.valueIn(primaryDataStoreTO.getId());
+                primaryDataStoreTO.setFullCloneFlag(value);
+            }
+        }
+        return dataTO;
+    }
+
     protected Answer copyObject(DataObject srcData, DataObject destData) {
         return copyObject(srcData, destData, null);
     }
@@ -257,7 +277,7 @@ public class AncientDataMotionStrategy implements DataMotionStrategy {
                 ep = selector.select(srcData, volObj);
             }
 
-            CopyCommand cmd = new CopyCommand(srcData.getTO(), volObj.getTO(), _createVolumeFromSnapshotWait, VirtualMachineManager.ExecuteInSequence.value());
+            CopyCommand cmd = new CopyCommand(srcData.getTO(), addFullCloneFlagOnVMwareDest(volObj.getTO()), _createVolumeFromSnapshotWait, VirtualMachineManager.ExecuteInSequence.value());
             Answer answer = null;
             if (ep == null) {
                 String errMsg = "No remote endpoint to send command, check if host or ssvm is down?";
@@ -280,7 +300,7 @@ public class AncientDataMotionStrategy implements DataMotionStrategy {
     }
 
     protected Answer cloneVolume(DataObject template, DataObject volume) {
-        CopyCommand cmd = new CopyCommand(template.getTO(), volume.getTO(), 0, VirtualMachineManager.ExecuteInSequence.value());
+        CopyCommand cmd = new CopyCommand(template.getTO(), addFullCloneFlagOnVMwareDest(volume.getTO()), 0, VirtualMachineManager.ExecuteInSequence.value());
         try {
             EndPoint ep = selector.select(volume.getDataStore());
             Answer answer = null;
@@ -330,7 +350,7 @@ public class AncientDataMotionStrategy implements DataMotionStrategy {
 
             objOnImageStore.processEvent(Event.CopyingRequested);
 
-            CopyCommand cmd = new CopyCommand(objOnImageStore.getTO(), destData.getTO(), _copyvolumewait, VirtualMachineManager.ExecuteInSequence.value());
+            CopyCommand cmd = new CopyCommand(objOnImageStore.getTO(), addFullCloneFlagOnVMwareDest(destData.getTO()), _copyvolumewait, VirtualMachineManager.ExecuteInSequence.value());
             EndPoint ep = selector.select(objOnImageStore, destData);
             if (ep == null) {
                 String errMsg = "No remote endpoint to send command, check if host or ssvm is down?";
@@ -477,7 +497,7 @@ public class AncientDataMotionStrategy implements DataMotionStrategy {
             ep = selector.select(srcData, destData);
         }
 
-        CopyCommand cmd = new CopyCommand(srcData.getTO(), destData.getTO(), _createprivatetemplatefromsnapshotwait, VirtualMachineManager.ExecuteInSequence.value());
+        CopyCommand cmd = new CopyCommand(srcData.getTO(), addFullCloneFlagOnVMwareDest(destData.getTO()), _createprivatetemplatefromsnapshotwait, VirtualMachineManager.ExecuteInSequence.value());
         Answer answer = null;
         if (ep == null) {
             String errMsg = "No remote endpoint to send command, check if host or ssvm is down?";
@@ -513,7 +533,7 @@ public class AncientDataMotionStrategy implements DataMotionStrategy {
                 Scope selectedScope = pickCacheScopeForCopy(srcData, destData);
                 cacheData = cacheMgr.getCacheObject(srcData, selectedScope);
 
-                CopyCommand cmd = new CopyCommand(srcData.getTO(), destData.getTO(), _backupsnapshotwait, VirtualMachineManager.ExecuteInSequence.value());
+                CopyCommand cmd = new CopyCommand(srcData.getTO(), addFullCloneFlagOnVMwareDest(destData.getTO()), _backupsnapshotwait, VirtualMachineManager.ExecuteInSequence.value());
                 cmd.setCacheTO(cacheData.getTO());
                 cmd.setOptions(options);
                 EndPoint ep = selector.select(srcData, destData);
@@ -525,6 +545,7 @@ public class AncientDataMotionStrategy implements DataMotionStrategy {
                     answer = ep.sendMessage(cmd);
                 }
             } else {
+                addFullCloneFlagOnVMwareDest(destData.getTO());
                 CopyCommand cmd = new CopyCommand(srcData.getTO(), destData.getTO(), _backupsnapshotwait, VirtualMachineManager.ExecuteInSequence.value());
                 cmd.setOptions(options);
                 EndPoint ep = selector.select(srcData, destData, StorageAction.BACKUPSNAPSHOT);
diff --git a/engine/storage/datamotion/test/org/apache/cloudstack/storage/motion/AncientDataMotionStrategyTest.java b/engine/storage/datamotion/test/org/apache/cloudstack/storage/motion/AncientDataMotionStrategyTest.java
new file mode 100755
index 00000000000..dccb6b445e5
--- /dev/null
+++ b/engine/storage/datamotion/test/org/apache/cloudstack/storage/motion/AncientDataMotionStrategyTest.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.cloudstack.storage.motion;
+
+import static org.mockito.Mockito.when;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
+
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.any;
+
+import org.apache.cloudstack.framework.config.ConfigKey;
+import org.apache.cloudstack.storage.to.PrimaryDataStoreTO;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.mockito.Spy;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import com.cloud.agent.api.to.DataTO;
+import com.cloud.capacity.CapacityManager;
+import com.cloud.hypervisor.Hypervisor.HypervisorType;
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest(CapacityManager.class)
+public class AncientDataMotionStrategyTest {
+
+    @Spy
+    @InjectMocks
+    private AncientDataMotionStrategy strategy = new AncientDataMotionStrategy();
+
+    @Mock
+    DataTO dataTO;
+    @Mock
+    PrimaryDataStoreTO dataStoreTO;
+    @Mock
+    ConfigKey<Boolean> vmwareKey;
+
+    private static final long POOL_ID = 1l;
+    private static final Boolean FULL_CLONE_FLAG = true;
+
+    @Before
+    public void setup() throws Exception {
+        MockitoAnnotations.initMocks(this);
+
+        replaceVmwareCreateCloneFullField();
+
+        when(vmwareKey.valueIn(POOL_ID)).thenReturn(FULL_CLONE_FLAG);
+
+        when(dataTO.getHypervisorType()).thenReturn(HypervisorType.VMware);
+        when(dataTO.getDataStore()).thenReturn(dataStoreTO);
+        when(dataStoreTO.getId()).thenReturn(POOL_ID);
+    }
+
+    private void replaceVmwareCreateCloneFullField() throws Exception {
+        Field field = CapacityManager.class.getDeclaredField("VmwareCreateCloneFull");
+        field.setAccessible(true);
+        // remove final modifier from field
+        Field modifiersField = Field.class.getDeclaredField("modifiers");
+        modifiersField.setAccessible(true);
+        modifiersField.setInt(field, field.getModifiers() & ~Modifier.FINAL);
+        field.set(null, vmwareKey);
+    }
+
+    @Test
+    public void testAddFullCloneFlagOnVMwareDest(){
+        strategy.addFullCloneFlagOnVMwareDest(dataTO);
+        verify(dataStoreTO).setFullCloneFlag(FULL_CLONE_FLAG);
+    }
+
+    @Test
+    public void testAddFullCloneFlagOnNotVmwareDest(){
+        when(dataTO.getHypervisorType()).thenReturn(HypervisorType.Any);
+        verify(dataStoreTO, never()).setFullCloneFlag(any(Boolean.class));
+    }
+
+}
diff --git a/plugins/hypervisors/vmware/src/com/cloud/hypervisor/vmware/resource/VmwareResource.java b/plugins/hypervisors/vmware/src/com/cloud/hypervisor/vmware/resource/VmwareResource.java
index a6db828f22f..522b8ae0440 100644
--- a/plugins/hypervisors/vmware/src/com/cloud/hypervisor/vmware/resource/VmwareResource.java
+++ b/plugins/hypervisors/vmware/src/com/cloud/hypervisor/vmware/resource/VmwareResource.java
@@ -30,6 +30,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.Date;
+import java.util.EnumMap;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -100,6 +101,7 @@ import com.vmware.vim25.VmwareDistributedVirtualSwitchVlanIdSpec;
 import org.apache.cloudstack.storage.command.CopyCommand;
 import org.apache.cloudstack.storage.command.StorageSubSystemCommand;
 import org.apache.cloudstack.storage.resource.NfsSecondaryStorageResource;
+import org.apache.cloudstack.storage.to.PrimaryDataStoreTO;
 import org.apache.cloudstack.storage.to.TemplateObjectTO;
 import org.apache.cloudstack.storage.to.VolumeObjectTO;
 import org.apache.commons.lang.math.NumberUtils;
@@ -266,6 +268,7 @@ import com.cloud.storage.resource.StorageSubsystemCommandHandler;
 import com.cloud.storage.resource.VmwareStorageLayoutHelper;
 import com.cloud.storage.resource.VmwareStorageProcessor;
 import com.cloud.storage.resource.VmwareStorageSubsystemCommandHandler;
+import com.cloud.storage.resource.VmwareStorageProcessor.VmwareStorageProcessorConfigurableFields;
 import com.cloud.storage.template.TemplateProp;
 import com.cloud.utils.DateUtil;
 import com.cloud.utils.ExecutionResult;
@@ -537,15 +540,54 @@ public class VmwareResource implements StoragePoolResource, ServerResource, Vmwa
     protected void checkStorageProcessorAndHandlerNfsVersionAttribute(StorageSubSystemCommand cmd) {
         if (storageNfsVersion != null) return;
         if (cmd instanceof CopyCommand){
-            examineStorageSubSystemCommandNfsVersion((CopyCommand) cmd);
+            EnumMap<VmwareStorageProcessorConfigurableFields,Object> params = new EnumMap<VmwareStorageProcessorConfigurableFields,Object>(VmwareStorageProcessorConfigurableFields.class);
+            examineStorageSubSystemCommandNfsVersion((CopyCommand) cmd, params);
+            params = examineStorageSubSystemCommandFullCloneFlagForVmware((CopyCommand) cmd, params);
+            reconfigureProcessorByHandler(params);
         }
     }
 
+    /**
+     * Reconfigure processor by handler
+     * @param params params
+     */
+    protected void reconfigureProcessorByHandler(EnumMap<VmwareStorageProcessorConfigurableFields,Object> params) {
+        VmwareStorageSubsystemCommandHandler handler = (VmwareStorageSubsystemCommandHandler) storageHandler;
+        boolean success = handler.reconfigureStorageProcessor(params);
+        if (success){
+            s_logger.info("VmwareStorageProcessor and VmwareStorageSubsystemCommandHandler successfully reconfigured");
+        } else {
+            s_logger.error("Error while reconfiguring VmwareStorageProcessor and VmwareStorageSubsystemCommandHandler, params=" + _gson.toJson(params));
+        }
+    }
+
+    /**
+     * Examine StorageSubSystem command to get full clone flag, if provided
+     * @param cmd command to execute
+     * @param params params
+     * @return copy of params including new values, if suitable
+     */
+    protected EnumMap<VmwareStorageProcessorConfigurableFields,Object> examineStorageSubSystemCommandFullCloneFlagForVmware(CopyCommand cmd, EnumMap<VmwareStorageProcessorConfigurableFields,Object> params) {
+        EnumMap<VmwareStorageProcessorConfigurableFields, Object> paramsCopy = new EnumMap<VmwareStorageProcessorConfigurableFields, Object>(params);
+        HypervisorType hypervisor = cmd.getDestTO().getHypervisorType();
+        if (hypervisor != null && hypervisor.equals(HypervisorType.VMware)){
+            DataStoreTO destDataStore = cmd.getDestTO().getDataStore();
+            if (destDataStore instanceof PrimaryDataStoreTO){
+                PrimaryDataStoreTO dest = (PrimaryDataStoreTO) destDataStore;
+                if (dest.isFullCloneFlag() != null){
+                    paramsCopy.put(VmwareStorageProcessorConfigurableFields.FULL_CLONE_FLAG, dest.isFullCloneFlag().booleanValue());
+                }
+            }
+        }
+        return paramsCopy;
+    }
+
     /**
      * Examine StorageSubSystem command to get storage NFS version, if provided
      * @param cmd command to execute
+     * @param params params
      */
-    protected void examineStorageSubSystemCommandNfsVersion(CopyCommand cmd){
+    protected void examineStorageSubSystemCommandNfsVersion(CopyCommand cmd, EnumMap<VmwareStorageProcessorConfigurableFields,Object> params){
         DataStoreTO srcDataStore = cmd.getSrcTO().getDataStore();
         boolean nfsVersionFound = false;
 
@@ -554,7 +596,7 @@ public class VmwareResource implements StoragePoolResource, ServerResource, Vmwa
         }
 
         if (nfsVersionFound){
-            setCurrentNfsVersionInProcessorAndHandler();
+            params.put(VmwareStorageProcessorConfigurableFields.NFS_VERSION, storageNfsVersion);
         }
     }
 
@@ -571,20 +613,6 @@ public class VmwareResource implements StoragePoolResource, ServerResource, Vmwa
         return false;
     }
 
-    /**
-     * Sets _storageNfsVersion into storage processor and storage handler by calling reconfigureNfsVersion on the storage handler,
-     * which will set NFS version into it and the storage processor.
-     */
-    protected void setCurrentNfsVersionInProcessorAndHandler() {
-        VmwareStorageSubsystemCommandHandler handler = (VmwareStorageSubsystemCommandHandler) storageHandler;
-        boolean success = handler.reconfigureNfsVersion(storageNfsVersion);
-        if (success){
-            s_logger.info("NFS version " + storageNfsVersion + " successfully set in VmwareStorageProcessor and VmwareStorageSubsystemCommandHandler");
-        } else {
-            s_logger.error("Error while setting NFS version " + storageNfsVersion);
-        }
-    }
-
     /**
      * Registers the vm to the inventory given the vmx file.
      */
diff --git a/plugins/hypervisors/vmware/src/com/cloud/storage/resource/VmwareStorageProcessor.java b/plugins/hypervisors/vmware/src/com/cloud/storage/resource/VmwareStorageProcessor.java
index eb18e0b9a80..8ce65c42f60 100644
--- a/plugins/hypervisors/vmware/src/com/cloud/storage/resource/VmwareStorageProcessor.java
+++ b/plugins/hypervisors/vmware/src/com/cloud/storage/resource/VmwareStorageProcessor.java
@@ -114,11 +114,25 @@ import com.cloud.vm.VmDetailConstants;
 
 public class VmwareStorageProcessor implements StorageProcessor {
 
+    public enum VmwareStorageProcessorConfigurableFields {
+        NFS_VERSION("nfsVersion"), FULL_CLONE_FLAG("fullCloneFlag");
+
+        private String name;
+
+        VmwareStorageProcessorConfigurableFields(String name){
+            this.name = name;
+        }
+
+        public String getName() {
+            return name;
+        }
+    }
+
     private static final Logger s_logger = Logger.getLogger(VmwareStorageProcessor.class);
     private static final int DEFAULT_NFS_PORT = 2049;
 
     private final VmwareHostService hostService;
-    private final boolean _fullCloneFlag;
+    private boolean _fullCloneFlag;
     private final VmwareStorageMount mountService;
     private final VmwareResource resource;
     private final Integer _timeout;
@@ -2394,4 +2408,9 @@ public class VmwareStorageProcessor implements StorageProcessor {
         this._nfsVersion = nfsVersion;
         s_logger.debug("VmwareProcessor instance now using NFS version: " + nfsVersion);
     }
+
+    public void setFullCloneFlag(boolean value){
+        this._fullCloneFlag = value;
+        s_logger.debug("VmwareProcessor instance - create full clone = " + (value ? "TRUE" : "FALSE"));
+    }
 }
diff --git a/plugins/hypervisors/vmware/src/com/cloud/storage/resource/VmwareStorageSubsystemCommandHandler.java b/plugins/hypervisors/vmware/src/com/cloud/storage/resource/VmwareStorageSubsystemCommandHandler.java
index 7252f51bf39..fc199722b2b 100644
--- a/plugins/hypervisors/vmware/src/com/cloud/storage/resource/VmwareStorageSubsystemCommandHandler.java
+++ b/plugins/hypervisors/vmware/src/com/cloud/storage/resource/VmwareStorageSubsystemCommandHandler.java
@@ -19,6 +19,7 @@
 package com.cloud.storage.resource;
 
 import java.io.File;
+import java.util.EnumMap;
 
 import org.apache.log4j.Logger;
 import org.apache.cloudstack.storage.command.CopyCmdAnswer;
@@ -37,6 +38,7 @@ import com.cloud.agent.api.to.S3TO;
 import com.cloud.agent.api.to.SwiftTO;
 import com.cloud.hypervisor.vmware.manager.VmwareStorageManager;
 import com.cloud.storage.DataStoreRole;
+import com.cloud.storage.resource.VmwareStorageProcessor.VmwareStorageProcessorConfigurableFields;
 
 public class VmwareStorageSubsystemCommandHandler extends StorageSubsystemCommandHandlerBase {
 
@@ -66,21 +68,25 @@ public class VmwareStorageSubsystemCommandHandler extends StorageSubsystemComman
         this._nfsVersion = nfsVersion;
     }
 
-    /**
-     * Reconfigure NFS version for storage operations
-     * @param nfsVersion NFS version to set
-     * @return true if NFS version could be configured, false in other case
-     */
-    public boolean reconfigureNfsVersion(Integer nfsVersion){
-        try {
-            VmwareStorageProcessor processor = (VmwareStorageProcessor) this.processor;
-            processor.setNfsVersion(nfsVersion);
-            this._nfsVersion = nfsVersion;
-            return true;
-        } catch (Exception e){
-            s_logger.error("Error while reconfiguring NFS version " + nfsVersion);
-            return false;
+    public boolean reconfigureStorageProcessor(EnumMap<VmwareStorageProcessorConfigurableFields,Object> params) {
+        VmwareStorageProcessor processor = (VmwareStorageProcessor) this.processor;
+        for (VmwareStorageProcessorConfigurableFields key : params.keySet()){
+            switch (key){
+            case NFS_VERSION:
+                Integer nfsVersion = (Integer) params.get(key);
+                processor.setNfsVersion(nfsVersion);
+                this._nfsVersion = nfsVersion;
+                break;
+            case FULL_CLONE_FLAG:
+                boolean fullClone = (boolean) params.get(key);
+                processor.setFullCloneFlag(fullClone);
+                break;
+            default:
+                s_logger.error("Unknown reconfigurable field " + key.getName() + " for VmwareStorageProcessor");
+                return false;
+            }
         }
+        return true;
     }
 
     @Override
@@ -187,4 +193,5 @@ public class VmwareStorageSubsystemCommandHandler extends StorageSubsystemComman
             return super.execute(cmd);
         }
     }
+
 }
diff --git a/plugins/hypervisors/vmware/test/com/cloud/hypervisor/vmware/resource/VmwareResourceTest.java b/plugins/hypervisors/vmware/test/com/cloud/hypervisor/vmware/resource/VmwareResourceTest.java
index efaf6d28d8b..3aac1323480 100644
--- a/plugins/hypervisors/vmware/test/com/cloud/hypervisor/vmware/resource/VmwareResourceTest.java
+++ b/plugins/hypervisors/vmware/test/com/cloud/hypervisor/vmware/resource/VmwareResourceTest.java
@@ -25,20 +25,24 @@ import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.never;
 import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.times;
 
 import java.util.ArrayList;
 import static org.powermock.api.mockito.PowerMockito.whenNew;
 
 import java.util.Arrays;
+import java.util.EnumMap;
 import java.util.HashMap;
 import java.util.Map;
 
 import org.apache.cloudstack.storage.command.CopyCommand;
+import org.apache.cloudstack.storage.to.PrimaryDataStoreTO;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.InOrder;
 import org.mockito.InjectMocks;
+import org.mockito.Matchers;
 import org.mockito.Mock;
 import org.mockito.Mockito;
 import org.mockito.MockitoAnnotations;
@@ -60,11 +64,13 @@ import com.cloud.agent.api.to.NfsTO;
 import com.cloud.agent.api.to.VirtualMachineTO;
 import com.cloud.agent.api.to.VolumeTO;
 import com.cloud.hypervisor.vmware.mo.DatacenterMO;
+import com.cloud.hypervisor.Hypervisor.HypervisorType;
 import com.cloud.hypervisor.vmware.mo.VirtualMachineMO;
 import com.cloud.hypervisor.vmware.mo.VmwareHypervisorHost;
 import com.cloud.hypervisor.vmware.util.VmwareContext;
 import com.cloud.storage.resource.VmwareStorageProcessor;
 import com.cloud.storage.resource.VmwareStorageSubsystemCommandHandler;
+import com.cloud.storage.resource.VmwareStorageProcessor.VmwareStorageProcessorConfigurableFields;
 
 import com.cloud.utils.exception.CloudRuntimeException;
 
@@ -125,13 +131,19 @@ public class VmwareResourceTest {
     ManagedObjectReference mor;
     @Mock
     DatacenterMO datacenter;
+    @Mock
+    DataTO destDataTO;
+    @Mock
+    PrimaryDataStoreTO destDataStoreTO;
 
     CopyCommand storageCmd;
+    EnumMap<VmwareStorageProcessorConfigurableFields, Object> params = new EnumMap<VmwareStorageProcessorConfigurableFields,Object>(VmwareStorageProcessorConfigurableFields.class);
 
     private static final Integer NFS_VERSION = Integer.valueOf(3);
     private static final Integer NFS_VERSION_NOT_PRESENT = null;
     private static final long VRAM_MEMORY_SIZE = 131072l;
     private static final long VIDEO_CARD_MEMORY_SIZE = 65536l;
+    private static final Boolean FULL_CLONE_FLAG = true;
 
     @Before
     public void setup() throws Exception {
@@ -139,11 +151,17 @@ public class VmwareResourceTest {
         storageCmd = PowerMockito.mock(CopyCommand.class);
         doReturn(context).when(_resource).getServiceContext(null);
         when(cmd.getVirtualMachine()).thenReturn(vmSpec);
+
         when(storageCmd.getSrcTO()).thenReturn(srcDataTO);
         when(srcDataTO.getDataStore()).thenReturn(srcDataNfsTO);
         when(srcDataNfsTO.getNfsVersion()).thenReturn(NFS_VERSION);
         when(videoCard.getVideoRamSizeInKB()).thenReturn(VIDEO_CARD_MEMORY_SIZE);
         when(volume.getPath()).thenReturn(VOLUME_PATH);
+
+        when(storageCmd.getDestTO()).thenReturn(destDataTO);
+        when(destDataTO.getHypervisorType()).thenReturn(HypervisorType.VMware);
+        when(destDataTO.getDataStore()).thenReturn(destDataStoreTO);
+        when(destDataStoreTO.isFullCloneFlag()).thenReturn(FULL_CLONE_FLAG);
     }
 
     //Test successful scaling up the vm
@@ -239,8 +257,9 @@ public class VmwareResourceTest {
 
     @Test
     public void testSetCurrentNfsVersionInProcessorAndHandler(){
-        _resource.setCurrentNfsVersionInProcessorAndHandler();
-        verify(storageHandler).reconfigureNfsVersion(any(Integer.class));
+        params.put(VmwareStorageProcessorConfigurableFields.NFS_VERSION, NFS_VERSION);
+        _resource.reconfigureProcessorByHandler(params);
+        verify(storageHandler).reconfigureStorageProcessor(params);
     }
 
     // ---------------------------------------------------------------------------------------------------
@@ -248,30 +267,69 @@ public class VmwareResourceTest {
     @Test
     public void testExamineStorageSubSystemCommandNfsVersionNotPresent(){
         when(srcDataNfsTO.getNfsVersion()).thenReturn(NFS_VERSION_NOT_PRESENT);
-        _resource.examineStorageSubSystemCommandNfsVersion(storageCmd);
-        verify(_resource, never()).setCurrentNfsVersionInProcessorAndHandler();
+        _resource.examineStorageSubSystemCommandNfsVersion(storageCmd,params);
+        assertTrue(params.isEmpty());
     }
 
     @Test
     public void testExamineStorageSubSystemCommandNfsVersion(){
-        _resource.examineStorageSubSystemCommandNfsVersion(storageCmd);
-        verify(_resource).setCurrentNfsVersionInProcessorAndHandler();
+        _resource.examineStorageSubSystemCommandNfsVersion(storageCmd, params);
+        assertEquals(1, params.size());
+        assertEquals(NFS_VERSION, params.get(VmwareStorageProcessorConfigurableFields.NFS_VERSION));
     }
 
     // ---------------------------------------------------------------------------------------------------
 
+    @Test
+    public void testExamineStorageSubSystemCommandFullCloneFlagForVmwareNullHypervisor(){
+        when(destDataTO.getHypervisorType()).thenReturn(null);
+        _resource.examineStorageSubSystemCommandFullCloneFlagForVmware(storageCmd, params);
+        verify(destDataTO, never()).getDataStore();
+    }
+
+    @Test
+    public void testExamineStorageSubSystemCommandFullCloneFlagForHypervisorNotVmware(){
+        when(destDataTO.getHypervisorType()).thenReturn(HypervisorType.XenServer);
+        _resource.examineStorageSubSystemCommandFullCloneFlagForVmware(storageCmd, params);
+        verify(destDataTO, never()).getDataStore();
+    }
+
+    @Test
+    public void testExamineStorageSubSystemCommandFullCloneFlagForVmware(){
+        EnumMap<VmwareStorageProcessorConfigurableFields, Object> params2 = _resource.examineStorageSubSystemCommandFullCloneFlagForVmware(storageCmd, params);
+        verify(destDataTO).getDataStore();
+        verify(destDataStoreTO, times(2)).isFullCloneFlag();
+        assertEquals(1, params2.size());
+        assertEquals(FULL_CLONE_FLAG, params2.get(VmwareStorageProcessorConfigurableFields.FULL_CLONE_FLAG));
+    }
+
+    @Test
+    public void testExamineStorageSubSystemCommandFullCloneFlagForVmwareNull(){
+        when(destDataStoreTO.isFullCloneFlag()).thenReturn(null);
+        _resource.examineStorageSubSystemCommandFullCloneFlagForVmware(storageCmd, params);
+        verify(destDataTO).getDataStore();
+        verify(destDataStoreTO).isFullCloneFlag();
+        assertTrue(params.isEmpty());
+    }
+
+    // ---------------------------------------------------------------------------------------------------
+
+    @SuppressWarnings("unchecked")
     @Test
     public void checkStorageProcessorAndHandlerNfsVersionAttributeVersionNotSet(){
         _resource.checkStorageProcessorAndHandlerNfsVersionAttribute(storageCmd);
-        verify(_resource).examineStorageSubSystemCommandNfsVersion(storageCmd);
+        verify(_resource).examineStorageSubSystemCommandNfsVersion(Matchers.eq(storageCmd), any(EnumMap.class));
+        verify(_resource).examineStorageSubSystemCommandFullCloneFlagForVmware(Matchers.eq(storageCmd), any(EnumMap.class));
+        verify(_resource).reconfigureProcessorByHandler(any(EnumMap.class));
         assertEquals(NFS_VERSION, _resource.storageNfsVersion);
     }
 
+    @SuppressWarnings("unchecked")
     @Test
     public void checkStorageProcessorAndHandlerNfsVersionAttributeVersionSet(){
         _resource.storageNfsVersion = NFS_VERSION;
         _resource.checkStorageProcessorAndHandlerNfsVersionAttribute(storageCmd);
-        verify(_resource, never()).examineStorageSubSystemCommandNfsVersion(storageCmd);
+        verify(_resource, never()).examineStorageSubSystemCommandNfsVersion(Matchers.eq(storageCmd), any(EnumMap.class));
     }
 
     @Test(expected=CloudRuntimeException.class)
diff --git a/server/src/com/cloud/capacity/CapacityManagerImpl.java b/server/src/com/cloud/capacity/CapacityManagerImpl.java
index d0ae3e99695..a3d2c3f4bab 100644
--- a/server/src/com/cloud/capacity/CapacityManagerImpl.java
+++ b/server/src/com/cloud/capacity/CapacityManagerImpl.java
@@ -1101,6 +1101,6 @@ public class CapacityManagerImpl extends ManagerBase implements CapacityManager,
     @Override
     public ConfigKey<?>[] getConfigKeys() {
         return new ConfigKey<?>[] {CpuOverprovisioningFactor, MemOverprovisioningFactor, StorageCapacityDisableThreshold, StorageOverprovisioningFactor,
-            StorageAllocatedCapacityDisableThreshold, StorageOperationsExcludeCluster};
+            StorageAllocatedCapacityDisableThreshold, StorageOperationsExcludeCluster, VmwareCreateCloneFull};
     }
 }
diff --git a/server/src/com/cloud/vm/UserVmManagerImpl.java b/server/src/com/cloud/vm/UserVmManagerImpl.java
index dd7e817aaea..5c82464c0a6 100644
--- a/server/src/com/cloud/vm/UserVmManagerImpl.java
+++ b/server/src/com/cloud/vm/UserVmManagerImpl.java
@@ -281,7 +281,6 @@ import com.cloud.vm.dao.InstanceGroupDao;
 import com.cloud.vm.dao.InstanceGroupVMMapDao;
 import com.cloud.vm.dao.NicDao;
 import com.cloud.vm.dao.SecondaryStorageVmDao;
-import com.cloud.vm.dao.UserVmCloneSettingDao;
 import com.cloud.vm.dao.UserVmDao;
 import com.cloud.vm.dao.UserVmDetailsDao;
 import com.cloud.vm.dao.VMInstanceDao;
@@ -296,10 +295,6 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
 
     // seconds
 
-    public enum UserVmCloneType {
-        full, linked
-    }
-
     @Inject
     EntityManager _entityMgr;
     @Inject
@@ -319,8 +314,6 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
     @Inject
     protected DomainDao _domainDao = null;
     @Inject
-    protected UserVmCloneSettingDao _vmCloneSettingDao = null;
-    @Inject
     protected UserVmDao _vmDao = null;
     @Inject
     protected UserVmJoinDao _vmJoinDao = null;
@@ -3556,19 +3549,6 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
                     vm.setDisplayVm(true);
                 }
 
-                // If hypervisor is vSphere, check for clone type setting.
-                if (hypervisorType.equals(HypervisorType.VMware)) {
-                    // retrieve clone flag.
-                    UserVmCloneType cloneType = UserVmCloneType.linked;
-                    String value = _configDao.getValue(Config.VmwareCreateFullClone.key());
-                    if (value != null) {
-                        if (Boolean.parseBoolean(value) == true)
-                            cloneType = UserVmCloneType.full;
-                    }
-                    UserVmCloneSettingVO vmCloneSettingVO = new UserVmCloneSettingVO(id, cloneType.toString());
-                    _vmCloneSettingDao.persist(vmCloneSettingVO);
-                }
-
                 long guestOSId = template.getGuestOSId();
                 GuestOSVO guestOS = _guestOSDao.findById(guestOSId);
                 long guestOSCategoryId = guestOS.getCategoryId();