diff --git a/heat/engine/resource.py b/heat/engine/resource.py
index 777929c08..7ca5f1e7d 100644
--- a/heat/engine/resource.py
+++ b/heat/engine/resource.py
@@ -657,7 +657,10 @@ class Resource(object):
     def _needs_update(self, after, before, after_props, before_props,
                       prev_resource):
         if self.status == self.FAILED:
-            raise UpdateReplace(self)
+            if hasattr(self, 'nested'):
+                return True
+            else:
+                raise UpdateReplace(self)
 
         if prev_resource is not None:
             cur_class_def, cur_ver = self.implementation_signature()
diff --git a/heat/tests/test_resource.py b/heat/tests/test_resource.py
index 7a7c24225..3abf44e75 100644
--- a/heat/tests/test_resource.py
+++ b/heat/tests/test_resource.py
@@ -284,6 +284,31 @@ class ResourceTest(common.HeatTestCase):
         self.assertRaises(
             resource.UpdateReplace, scheduler.TaskRunner(res.update, utmpl))
 
+    def test_update_replace_in_failed_without_nested(self):
+        tmpl = rsrc_defn.ResourceDefinition('test_resource',
+                                            'GenericResourceType',
+                                            {'Foo': 'abc'})
+        res = generic_rsrc.ResourceWithProps('test_resource', tmpl, self.stack)
+        res.update_allowed_properties = ('Foo',)
+        self.m.StubOutWithMock(generic_rsrc.ResourceWithProps, 'handle_create')
+        generic_rsrc.ResourceWithProps.handle_create().AndRaise(
+            exception.ResourceFailure)
+        self.m.ReplayAll()
+
+        self.assertRaises(exception.ResourceFailure,
+                          scheduler.TaskRunner(res.create))
+        self.assertEqual((res.CREATE, res.FAILED), res.state)
+
+        utmpl = rsrc_defn.ResourceDefinition('test_resource',
+                                             'GenericResourceType',
+                                             {'Foo': 'xyz'})
+        # resource in failed status and hasn't nested will enter
+        # UpdateReplace flow
+        self.assertRaises(
+            resource.UpdateReplace, scheduler.TaskRunner(res.update, utmpl))
+
+        self.m.VerifyAll()
+
     def test_updated_time_changes_only_on_update_calls(self):
         tmpl = rsrc_defn.ResourceDefinition('test_resource',
                                             'GenericResourceType')
diff --git a/heat/tests/test_stack_resource.py b/heat/tests/test_stack_resource.py
index 0fac606cc..5f48805c3 100644
--- a/heat/tests/test_stack_resource.py
+++ b/heat/tests/test_stack_resource.py
@@ -658,6 +658,24 @@ class StackResourceTest(common.HeatTestCase):
 
         self.m.VerifyAll()
 
+    def test_need_update_in_failed_state_for_nested_resource(self):
+        """
+        The resource in FAILED state and has nested stack,
+        should need update.
+        """
+        self.parent_resource.set_template(self.templ, {"KeyName": "test"})
+        scheduler.TaskRunner(self.parent_resource.create)()
+        self.parent_resource.state_set('CREATE', 'FAILED')
+
+        need_update = self.parent_resource._needs_update(
+            self.parent_resource.t,
+            self.parent_resource.t,
+            self.parent_resource.properties,
+            self.parent_resource.properties,
+            self.parent_resource)
+
+        self.assertEqual(True, need_update)
+
     def test_create_complete_state_err(self):
         """
         check_create_complete should raise error when create task is