diff --git a/heat/engine/resource.py b/heat/engine/resource.py
index d6eef638a2..34ad36b2be 100644
--- a/heat/engine/resource.py
+++ b/heat/engine/resource.py
@@ -931,7 +931,6 @@ class Resource(object):
             if not self._needs_update(after, before, after_props, before_props,
                                       prev_resource):
                 return
-
             if not cfg.CONF.convergence_engine:
                 if (self.action, self.status) in (
                         (self.CREATE, self.IN_PROGRESS),
@@ -958,14 +957,21 @@ class Resource(object):
                 self._update_stored_properties()
         except exception.UpdateReplace as ex:
             # catch all UpdateReplace expections
-            if (self.stack.action == 'ROLLBACK' and
-                    self.stack.status == 'IN_PROGRESS' and
-                    not cfg.CONF.convergence_engine):
-                # handle case, when it's rollback and we should restore
-                # old resource
-                self.restore_prev_rsrc()
-            else:
-                self.prepare_for_replace()
+            try:
+                if (self.stack.action == 'ROLLBACK' and
+                        self.stack.status == 'IN_PROGRESS' and
+                        not cfg.CONF.convergence_engine):
+                    # handle case, when it's rollback and we should restore
+                    # old resource
+                    self.restore_prev_rsrc()
+                else:
+                    self.prepare_for_replace()
+            except Exception as e:
+                # if any exception happen, we should set the resource to
+                # FAILED, then raise ResourceFailure
+                failure = exception.ResourceFailure(e, self, action)
+                self.state_set(action, self.FAILED, six.text_type(failure))
+                raise failure
             raise ex
 
     def prepare_for_replace(self):
diff --git a/heat/tests/test_resource.py b/heat/tests/test_resource.py
index 960ca12c5f..d083701bc1 100644
--- a/heat/tests/test_resource.py
+++ b/heat/tests/test_resource.py
@@ -363,63 +363,76 @@ class ResourceTest(common.HeatTestCase):
         self.assertIsNotNone(res.updated_time)
         self.assertNotEqual(res.updated_time, stored_time)
 
-    def test_update_replace(self):
+    def _setup_resource_for_update(self, res_name):
         class TestResource(resource.Resource):
             properties_schema = {'a_string': {'Type': 'String'}}
             update_allowed_properties = ('a_string',)
 
         resource._register_class('TestResource', TestResource)
 
-        tmpl = rsrc_defn.ResourceDefinition('test_resource',
+        tmpl = rsrc_defn.ResourceDefinition(res_name,
                                             'TestResource')
         res = TestResource('test_resource', tmpl, self.stack)
+
+        utmpl = rsrc_defn.ResourceDefinition(res_name, 'TestResource',
+                                             {'a_string': 'foo'})
+
+        return res, utmpl
+
+    def test_update_replace(self):
+        res, utmpl = self._setup_resource_for_update(
+            res_name='test_update_replace')
         res.prepare_for_replace = mock.Mock()
 
-        utmpl = rsrc_defn.ResourceDefinition('test_resource', 'TestResource',
-                                             {'a_string': 'foo'})
         self.assertRaises(
             exception.UpdateReplace, scheduler.TaskRunner(res.update, utmpl))
         self.assertTrue(res.prepare_for_replace.called)
 
+    def test_update_replace_prepare_replace_error(self):
+        # test if any error happened when prepare_for_replace,
+        # whether the resource will go to FAILED
+        res, utmpl = self._setup_resource_for_update(
+            res_name='test_update_replace_prepare_replace_error')
+        res.prepare_for_replace = mock.Mock(side_effect=Exception)
+
+        self.assertRaises(
+            exception.ResourceFailure,
+            scheduler.TaskRunner(res.update, utmpl))
+        self.assertTrue(res.prepare_for_replace.called)
+        self.assertEqual((res.UPDATE, res.FAILED), res.state)
+
     def test_update_rsrc_in_progress_raises_exception(self):
-        class TestResource(resource.Resource):
-            properties_schema = {'a_string': {'Type': 'String'}}
-            update_allowed_properties = ('a_string',)
+        res, utmpl = self._setup_resource_for_update(
+            res_name='test_update_rsrc_in_progress_raises_exception')
 
         cfg.CONF.set_override('convergence_engine', False)
-        resource._register_class('TestResource', TestResource)
 
-        tmpl = rsrc_defn.ResourceDefinition('test_resource',
-                                            'TestResource')
-        res = TestResource('test_resource', tmpl, self.stack)
-
-        utmpl = rsrc_defn.ResourceDefinition('test_resource', 'TestResource',
-                                             {'a_string': 'foo'})
         res.action = res.UPDATE
         res.status = res.IN_PROGRESS
         self.assertRaises(
             exception.ResourceFailure, scheduler.TaskRunner(res.update, utmpl))
 
     def test_update_replace_rollback(self):
-        class TestResource(resource.Resource):
-            properties_schema = {'a_string': {'Type': 'String'}}
-            update_allowed_properties = ('a_string',)
-
-        resource._register_class('TestResource', TestResource)
-
-        tmpl = rsrc_defn.ResourceDefinition('test_resource',
-                                            'TestResource')
-        self.stack.state_set('ROLLBACK', 'IN_PROGRESS', 'Simulate rollback')
-        res = TestResource('test_resource', tmpl, self.stack)
-
+        res, utmpl = self._setup_resource_for_update(
+            res_name='test_update_replace_rollback')
         res.restore_prev_rsrc = mock.Mock()
+        self.stack.state_set('ROLLBACK', 'IN_PROGRESS', 'Simulate rollback')
 
-        utmpl = rsrc_defn.ResourceDefinition('test_resource', 'TestResource',
-                                             {'a_string': 'foo'})
         self.assertRaises(
             exception.UpdateReplace, scheduler.TaskRunner(res.update, utmpl))
         self.assertTrue(res.restore_prev_rsrc.called)
 
+    def test_update_replace_rollback_restore_prev_rsrc_error(self):
+        res, utmpl = self._setup_resource_for_update(
+            res_name='restore_prev_rsrc_error')
+        res.restore_prev_rsrc = mock.Mock(side_effect=Exception)
+        self.stack.state_set('ROLLBACK', 'IN_PROGRESS', 'Simulate rollback')
+
+        self.assertRaises(
+            exception.ResourceFailure, scheduler.TaskRunner(res.update, utmpl))
+        self.assertTrue(res.restore_prev_rsrc.called)
+        self.assertEqual((res.UPDATE, res.FAILED), res.state)
+
     def test_update_replace_in_failed_without_nested(self):
         tmpl = rsrc_defn.ResourceDefinition('test_resource',
                                             'GenericResourceType',