diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3f4a79fad7..d9a06f7c84 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,7 @@
 ## HEAD (Unreleased)
+
 - Allow opting out of CRD rendering for Helm v3 by specifying `SkipCRDRendering` argument to Helm charts. (https://github.com/pulumi/pulumi-kubernetes/pull/1572)
+- Add replaceUnready annotation for Jobs. (https://github.com/pulumi/pulumi-kubernetes/pull/1575)
 
 ## 3.1.2 (May 12, 2021)
 
diff --git a/provider/cmd/pulumi-resource-kubernetes/schema.json b/provider/cmd/pulumi-resource-kubernetes/schema.json
index 36aacd13a2..d6956a2ade 100644
--- a/provider/cmd/pulumi-resource-kubernetes/schema.json
+++ b/provider/cmd/pulumi-resource-kubernetes/schema.json
@@ -9680,7 +9680,7 @@
             }
         },
         "kubernetes:batch/v1:Job": {
-            "description": "Job represents the configuration of a single job.\n\nThis resource waits until its status is ready before registering success\nfor create/update, and populating output properties from the current state of the resource.\nThe following conditions are used to determine whether the resource creation has\nsucceeded or failed:\n\n1. The Job's '.status.startTime' is set, which indicates that the Job has started running.\n2. The Job's '.status.conditions' has a status of type 'Complete', and a 'status' set\n   to 'True'.\n3. The Job's '.status.conditions' do not have a status of type 'Failed', with a\n\t'status' set to 'True'. If this condition is set, we should fail the Job immediately.\n\nIf the Job has not reached a Ready state after 10 minutes, it will\ntime out and mark the resource update as Failed. You can override the default timeout value\nby setting the 'customTimeouts' option on the resource.",
+            "description": "Job represents the configuration of a single job.\n\nThis resource waits until its status is ready before registering success\nfor create/update, and populating output properties from the current state of the resource.\nThe following conditions are used to determine whether the resource creation has\nsucceeded or failed:\n\n1. The Job's '.status.startTime' is set, which indicates that the Job has started running.\n2. The Job's '.status.conditions' has a status of type 'Complete', and a 'status' set\n   to 'True'.\n3. The Job's '.status.conditions' do not have a status of type 'Failed', with a\n\t'status' set to 'True'. If this condition is set, we should fail the Job immediately.\n\nIf the Job has not reached a Ready state after 10 minutes, it will\ntime out and mark the resource update as Failed. You can override the default timeout value\nby setting the 'customTimeouts' option on the resource.\n\nBy default, if a resource failed to become ready in a previous update, \nPulumi will continue to wait for readiness on the next update. If you would prefer\nto schedule a replacement for an unready resource on the next update, you can add the\n\"pulumi.com/replaceUnready\": \"true\" annotation to the resource definition.",
             "properties": {
                 "apiVersion": {
                     "type": "string",
@@ -33131,7 +33131,7 @@
             ]
         },
         "kubernetes:batch/v1:Job": {
-            "description": "Job represents the configuration of a single job.\n\nThis resource waits until its status is ready before registering success\nfor create/update, and populating output properties from the current state of the resource.\nThe following conditions are used to determine whether the resource creation has\nsucceeded or failed:\n\n1. The Job's '.status.startTime' is set, which indicates that the Job has started running.\n2. The Job's '.status.conditions' has a status of type 'Complete', and a 'status' set\n   to 'True'.\n3. The Job's '.status.conditions' do not have a status of type 'Failed', with a\n\t'status' set to 'True'. If this condition is set, we should fail the Job immediately.\n\nIf the Job has not reached a Ready state after 10 minutes, it will\ntime out and mark the resource update as Failed. You can override the default timeout value\nby setting the 'customTimeouts' option on the resource.",
+            "description": "Job represents the configuration of a single job.\n\nThis resource waits until its status is ready before registering success\nfor create/update, and populating output properties from the current state of the resource.\nThe following conditions are used to determine whether the resource creation has\nsucceeded or failed:\n\n1. The Job's '.status.startTime' is set, which indicates that the Job has started running.\n2. The Job's '.status.conditions' has a status of type 'Complete', and a 'status' set\n   to 'True'.\n3. The Job's '.status.conditions' do not have a status of type 'Failed', with a\n\t'status' set to 'True'. If this condition is set, we should fail the Job immediately.\n\nIf the Job has not reached a Ready state after 10 minutes, it will\ntime out and mark the resource update as Failed. You can override the default timeout value\nby setting the 'customTimeouts' option on the resource.\n\nBy default, if a resource failed to become ready in a previous update, \nPulumi will continue to wait for readiness on the next update. If you would prefer\nto schedule a replacement for an unready resource on the next update, you can add the\n\"pulumi.com/replaceUnready\": \"true\" annotation to the resource definition.",
             "properties": {
                 "apiVersion": {
                     "type": "string",
diff --git a/provider/pkg/gen/additionalComments.go b/provider/pkg/gen/additionalComments.go
index 5e1b81d26d..70dbf578cd 100644
--- a/provider/pkg/gen/additionalComments.go
+++ b/provider/pkg/gen/additionalComments.go
@@ -147,6 +147,13 @@ https://kubernetes.io/docs/concepts/configuration/secret/#risks`
 	}
 }
 
+func replaceUnreadyComment() string {
+	return `By default, if a resource failed to become ready in a previous update, 
+Pulumi will continue to wait for readiness on the next update. If you would prefer
+to schedule a replacement for an unready resource on the next update, you can add the
+"pulumi.com/replaceUnready": "true" annotation to the resource definition.`
+}
+
 // PulumiComment adds additional information to the docs generated automatically from the OpenAPI specs.
 // This includes information about Pulumi's await behavior, deprecation information, etc.
 func PulumiComment(kind string) string {
@@ -154,8 +161,10 @@ func PulumiComment(kind string) string {
 
 	k := kinds.Kind(kind)
 	switch k {
-	case kinds.Deployment, kinds.Ingress, kinds.Job, kinds.Pod, kinds.Service, kinds.StatefulSet:
+	case kinds.Deployment, kinds.Ingress, kinds.Pod, kinds.Service, kinds.StatefulSet:
 		return prefix + awaitComments(k)
+	case kinds.Job:
+		return prefix + awaitComments(k) + prefix + replaceUnreadyComment()
 	case kinds.Secret:
 		return prefix + helpfulLinkComments(k)
 	default:
diff --git a/provider/pkg/metadata/annotations.go b/provider/pkg/metadata/annotations.go
index 72f4ccb613..72f370684c 100644
--- a/provider/pkg/metadata/annotations.go
+++ b/provider/pkg/metadata/annotations.go
@@ -31,6 +31,7 @@ const (
 	AnnotationSkipAwait         = AnnotationPrefix + "skipAwait"
 	AnnotationTimeoutSeconds    = AnnotationPrefix + "timeoutSeconds"
 	AnnotationInitialAPIVersion = AnnotationPrefix + "initialApiVersion"
+	AnnotationReplaceUnready    = AnnotationPrefix + "replaceUnready"
 
 	AnnotationHelmHook = "helm.sh/hook"
 )
diff --git a/provider/pkg/metadata/overrides.go b/provider/pkg/metadata/overrides.go
index 786baf6fb6..4a139ad589 100644
--- a/provider/pkg/metadata/overrides.go
+++ b/provider/pkg/metadata/overrides.go
@@ -26,6 +26,11 @@ func SkipAwaitLogic(obj *unstructured.Unstructured) bool {
 	return IsAnnotationTrue(obj, AnnotationSkipAwait)
 }
 
+// ReplaceUnready returns true if the `pulumi.com/replaceUnready` annotation is "true", false otherwise.
+func ReplaceUnready(obj *unstructured.Unstructured) bool {
+	return IsAnnotationTrue(obj, AnnotationReplaceUnready)
+}
+
 // TimeoutDuration returns the resource timeout duration. There are a number of things it can do here in this order
 // 1. Return the timeout as specified in the customResource options
 // 2. Return the timeout as specified in `pulumi.com/timeoutSeconds` annotation,
diff --git a/provider/pkg/provider/provider.go b/provider/pkg/provider/provider.go
index 1deb545555..485b950396 100644
--- a/provider/pkg/provider/provider.go
+++ b/provider/pkg/provider/provider.go
@@ -1,4 +1,4 @@
-// Copyright 2016-2018, Pulumi Corporation.
+// Copyright 2016-2021, Pulumi Corporation.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -35,6 +35,7 @@ import (
 	structpb "github.com/golang/protobuf/ptypes/struct"
 	pkgerrors "github.com/pkg/errors"
 	"github.com/pulumi/pulumi-kubernetes/provider/v3/pkg/await"
+	"github.com/pulumi/pulumi-kubernetes/provider/v3/pkg/await/states"
 	"github.com/pulumi/pulumi-kubernetes/provider/v3/pkg/clients"
 	"github.com/pulumi/pulumi-kubernetes/provider/v3/pkg/cluster"
 	"github.com/pulumi/pulumi-kubernetes/provider/v3/pkg/gen"
@@ -1368,6 +1369,23 @@ func (k *kubeProvider) Diff(
 		}
 	}
 
+	if metadata.ReplaceUnready(newInputs) {
+		switch newInputs.GetKind() {
+		case "Job":
+			jobChecker := states.NewJobChecker()
+			job, err := clients.FromUnstructured(newInputs)
+			if err == nil {
+				jobChecker.Update(job)
+				if !jobChecker.Ready() {
+					hasChanges = pulumirpc.DiffResponse_DIFF_SOME
+					replaces = append(replaces, `.metadata.annotations["pulumi.com/replaceUnready"]`)
+				}
+			}
+		default:
+			_ = k.host.Log(ctx, diag.Warning, urn, "replaceUnready annotation is not supported for this resource")
+		}
+	}
+
 	// Delete before replacement if we are forced to replace the old object, and the new version of
 	// that object MUST have the same name.
 	deleteBeforeReplace :=
diff --git a/sdk/dotnet/Batch/V1/Job.cs b/sdk/dotnet/Batch/V1/Job.cs
index cb18183899..402759c965 100755
--- a/sdk/dotnet/Batch/V1/Job.cs
+++ b/sdk/dotnet/Batch/V1/Job.cs
@@ -26,6 +26,11 @@ namespace Pulumi.Kubernetes.Batch.V1
     /// If the Job has not reached a Ready state after 10 minutes, it will
     /// time out and mark the resource update as Failed. You can override the default timeout value
     /// by setting the 'customTimeouts' option on the resource.
+    /// 
+    /// By default, if a resource failed to become ready in a previous update,
+    /// Pulumi will continue to wait for readiness on the next update. If you would prefer
+    /// to schedule a replacement for an unready resource on the next update, you can add the
+    /// "pulumi.com/replaceUnready": "true" annotation to the resource definition.
     /// </summary>
     [KubernetesResourceType("kubernetes:batch/v1:Job")]
     public partial class Job : KubernetesResource
diff --git a/sdk/go/kubernetes/batch/v1/job.go b/sdk/go/kubernetes/batch/v1/job.go
index 48c739e747..925093feb3 100644
--- a/sdk/go/kubernetes/batch/v1/job.go
+++ b/sdk/go/kubernetes/batch/v1/job.go
@@ -27,6 +27,11 @@ import (
 // If the Job has not reached a Ready state after 10 minutes, it will
 // time out and mark the resource update as Failed. You can override the default timeout value
 // by setting the 'customTimeouts' option on the resource.
+//
+// By default, if a resource failed to become ready in a previous update,
+// Pulumi will continue to wait for readiness on the next update. If you would prefer
+// to schedule a replacement for an unready resource on the next update, you can add the
+// "pulumi.com/replaceUnready": "true" annotation to the resource definition.
 type Job struct {
 	pulumi.CustomResourceState
 
diff --git a/sdk/go/kubernetes/batch/v1/pulumiTypes.go b/sdk/go/kubernetes/batch/v1/pulumiTypes.go
index a4333fbf26..34cef70374 100644
--- a/sdk/go/kubernetes/batch/v1/pulumiTypes.go
+++ b/sdk/go/kubernetes/batch/v1/pulumiTypes.go
@@ -666,6 +666,11 @@ func (o CronJobStatusPtrOutput) LastSuccessfulTime() pulumi.StringPtrOutput {
 // If the Job has not reached a Ready state after 10 minutes, it will
 // time out and mark the resource update as Failed. You can override the default timeout value
 // by setting the 'customTimeouts' option on the resource.
+//
+// By default, if a resource failed to become ready in a previous update,
+// Pulumi will continue to wait for readiness on the next update. If you would prefer
+// to schedule a replacement for an unready resource on the next update, you can add the
+// "pulumi.com/replaceUnready": "true" annotation to the resource definition.
 type JobType struct {
 	// APIVersion defines the versioned schema of this representation of an object. Servers should convert recognized schemas to the latest internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources
 	ApiVersion *string `pulumi:"apiVersion"`
@@ -706,6 +711,11 @@ type JobTypeInput interface {
 // If the Job has not reached a Ready state after 10 minutes, it will
 // time out and mark the resource update as Failed. You can override the default timeout value
 // by setting the 'customTimeouts' option on the resource.
+//
+// By default, if a resource failed to become ready in a previous update,
+// Pulumi will continue to wait for readiness on the next update. If you would prefer
+// to schedule a replacement for an unready resource on the next update, you can add the
+// "pulumi.com/replaceUnready": "true" annotation to the resource definition.
 type JobTypeArgs struct {
 	// APIVersion defines the versioned schema of this representation of an object. Servers should convert recognized schemas to the latest internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources
 	ApiVersion pulumi.StringPtrInput `pulumi:"apiVersion"`
@@ -772,6 +782,11 @@ func (i JobTypeArray) ToJobTypeArrayOutputWithContext(ctx context.Context) JobTy
 // If the Job has not reached a Ready state after 10 minutes, it will
 // time out and mark the resource update as Failed. You can override the default timeout value
 // by setting the 'customTimeouts' option on the resource.
+//
+// By default, if a resource failed to become ready in a previous update,
+// Pulumi will continue to wait for readiness on the next update. If you would prefer
+// to schedule a replacement for an unready resource on the next update, you can add the
+// "pulumi.com/replaceUnready": "true" annotation to the resource definition.
 type JobTypeOutput struct{ *pulumi.OutputState }
 
 func (JobTypeOutput) ElementType() reflect.Type {
diff --git a/sdk/nodejs/batch/v1/job.ts b/sdk/nodejs/batch/v1/job.ts
index b10f9e5865..92f1da4765 100644
--- a/sdk/nodejs/batch/v1/job.ts
+++ b/sdk/nodejs/batch/v1/job.ts
@@ -22,6 +22,11 @@ import * as utilities from "../../utilities";
  * If the Job has not reached a Ready state after 10 minutes, it will
  * time out and mark the resource update as Failed. You can override the default timeout value
  * by setting the 'customTimeouts' option on the resource.
+ *
+ * By default, if a resource failed to become ready in a previous update,
+ * Pulumi will continue to wait for readiness on the next update. If you would prefer
+ * to schedule a replacement for an unready resource on the next update, you can add the
+ * "pulumi.com/replaceUnready": "true" annotation to the resource definition.
  */
 export class Job extends pulumi.CustomResource {
     /**
diff --git a/sdk/nodejs/types/input.ts b/sdk/nodejs/types/input.ts
index a3c82f1b9d..6af01d8af9 100644
--- a/sdk/nodejs/types/input.ts
+++ b/sdk/nodejs/types/input.ts
@@ -4831,6 +4831,11 @@ export namespace batch {
          * If the Job has not reached a Ready state after 10 minutes, it will
          * time out and mark the resource update as Failed. You can override the default timeout value
          * by setting the 'customTimeouts' option on the resource.
+         *
+         * By default, if a resource failed to become ready in a previous update, 
+         * Pulumi will continue to wait for readiness on the next update. If you would prefer
+         * to schedule a replacement for an unready resource on the next update, you can add the
+         * "pulumi.com/replaceUnready": "true" annotation to the resource definition.
          */
         export interface Job {
             /**
diff --git a/sdk/nodejs/types/output.ts b/sdk/nodejs/types/output.ts
index 52967a2abb..dedaa88fac 100644
--- a/sdk/nodejs/types/output.ts
+++ b/sdk/nodejs/types/output.ts
@@ -5091,6 +5091,11 @@ export namespace batch {
          * If the Job has not reached a Ready state after 10 minutes, it will
          * time out and mark the resource update as Failed. You can override the default timeout value
          * by setting the 'customTimeouts' option on the resource.
+         *
+         * By default, if a resource failed to become ready in a previous update, 
+         * Pulumi will continue to wait for readiness on the next update. If you would prefer
+         * to schedule a replacement for an unready resource on the next update, you can add the
+         * "pulumi.com/replaceUnready": "true" annotation to the resource definition.
          */
         export interface Job {
             /**
diff --git a/sdk/python/pulumi_kubernetes/batch/v1/Job.py b/sdk/python/pulumi_kubernetes/batch/v1/Job.py
index c792fe6086..dd9af28409 100644
--- a/sdk/python/pulumi_kubernetes/batch/v1/Job.py
+++ b/sdk/python/pulumi_kubernetes/batch/v1/Job.py
@@ -114,6 +114,11 @@ def __init__(__self__,
         time out and mark the resource update as Failed. You can override the default timeout value
         by setting the 'customTimeouts' option on the resource.
 
+        By default, if a resource failed to become ready in a previous update,
+        Pulumi will continue to wait for readiness on the next update. If you would prefer
+        to schedule a replacement for an unready resource on the next update, you can add the
+        "pulumi.com/replaceUnready": "true" annotation to the resource definition.
+
         :param str resource_name: The name of the resource.
         :param pulumi.ResourceOptions opts: Options for the resource.
         :param pulumi.Input[str] api_version: APIVersion defines the versioned schema of this representation of an object. Servers should convert recognized schemas to the latest internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources
@@ -145,6 +150,11 @@ def __init__(__self__,
         time out and mark the resource update as Failed. You can override the default timeout value
         by setting the 'customTimeouts' option on the resource.
 
+        By default, if a resource failed to become ready in a previous update,
+        Pulumi will continue to wait for readiness on the next update. If you would prefer
+        to schedule a replacement for an unready resource on the next update, you can add the
+        "pulumi.com/replaceUnready": "true" annotation to the resource definition.
+
         :param str resource_name: The name of the resource.
         :param JobArgs args: The arguments to use to populate this resource's properties.
         :param pulumi.ResourceOptions opts: Options for the resource.
diff --git a/sdk/python/pulumi_kubernetes/batch/v1/_inputs.py b/sdk/python/pulumi_kubernetes/batch/v1/_inputs.py
index bbf2c336e1..a985e4e056 100644
--- a/sdk/python/pulumi_kubernetes/batch/v1/_inputs.py
+++ b/sdk/python/pulumi_kubernetes/batch/v1/_inputs.py
@@ -308,6 +308,11 @@ def __init__(__self__, *,
         If the Job has not reached a Ready state after 10 minutes, it will
         time out and mark the resource update as Failed. You can override the default timeout value
         by setting the 'customTimeouts' option on the resource.
+
+        By default, if a resource failed to become ready in a previous update, 
+        Pulumi will continue to wait for readiness on the next update. If you would prefer
+        to schedule a replacement for an unready resource on the next update, you can add the
+        "pulumi.com/replaceUnready": "true" annotation to the resource definition.
         :param pulumi.Input[str] api_version: APIVersion defines the versioned schema of this representation of an object. Servers should convert recognized schemas to the latest internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources
         :param pulumi.Input[str] kind: Kind is a string value representing the REST resource this object represents. Servers may infer this from the endpoint the client submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds
         :param pulumi.Input['_meta.v1.ObjectMetaArgs'] metadata: Standard object's metadata. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#metadata
diff --git a/sdk/python/pulumi_kubernetes/batch/v1/outputs.py b/sdk/python/pulumi_kubernetes/batch/v1/outputs.py
index b8f6026b46..8a0800cab4 100644
--- a/sdk/python/pulumi_kubernetes/batch/v1/outputs.py
+++ b/sdk/python/pulumi_kubernetes/batch/v1/outputs.py
@@ -313,6 +313,11 @@ class Job(dict):
     If the Job has not reached a Ready state after 10 minutes, it will
     time out and mark the resource update as Failed. You can override the default timeout value
     by setting the 'customTimeouts' option on the resource.
+
+    By default, if a resource failed to become ready in a previous update, 
+    Pulumi will continue to wait for readiness on the next update. If you would prefer
+    to schedule a replacement for an unready resource on the next update, you can add the
+    "pulumi.com/replaceUnready": "true" annotation to the resource definition.
     """
     @staticmethod
     def __key_warning(key: str):
@@ -354,6 +359,11 @@ def __init__(__self__, *,
         If the Job has not reached a Ready state after 10 minutes, it will
         time out and mark the resource update as Failed. You can override the default timeout value
         by setting the 'customTimeouts' option on the resource.
+
+        By default, if a resource failed to become ready in a previous update, 
+        Pulumi will continue to wait for readiness on the next update. If you would prefer
+        to schedule a replacement for an unready resource on the next update, you can add the
+        "pulumi.com/replaceUnready": "true" annotation to the resource definition.
         :param str api_version: APIVersion defines the versioned schema of this representation of an object. Servers should convert recognized schemas to the latest internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources
         :param str kind: Kind is a string value representing the REST resource this object represents. Servers may infer this from the endpoint the client submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds
         :param '_meta.v1.ObjectMetaArgs' metadata: Standard object's metadata. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#metadata
diff --git a/tests/sdk/nodejs/nodejs_test.go b/tests/sdk/nodejs/nodejs_test.go
index 8d41c468a9..47e8361739 100644
--- a/tests/sdk/nodejs/nodejs_test.go
+++ b/tests/sdk/nodejs/nodejs_test.go
@@ -30,6 +30,7 @@ import (
 	"github.com/pulumi/pulumi-kubernetes/tests/v3"
 	"github.com/pulumi/pulumi/pkg/v3/resource/deploy/providers"
 	"github.com/pulumi/pulumi/pkg/v3/testing/integration"
+	"github.com/pulumi/pulumi/sdk/v3/go/common/apitype"
 	"github.com/pulumi/pulumi/sdk/v3/go/common/resource"
 	"github.com/pulumi/pulumi/sdk/v3/go/common/tokens"
 	"github.com/stretchr/testify/assert"
@@ -849,6 +850,90 @@ func TestRenderYAML(t *testing.T) {
 	integration.ProgramTest(t, &test)
 }
 
+func TestReplaceUnready(t *testing.T) {
+	test := baseOptions.With(integration.ProgramTestOptions{
+		Dir:                  filepath.Join("replace-unready", "step1"),
+		Quick:                true,
+		ExpectFailure:        true, // The Job is intended to fail.
+		ExpectRefreshChanges: true,
+		ExtraRuntimeValidation: func(t *testing.T, stackInfo integration.RuntimeValidationStackInfo) {
+			assert.NotNil(t, stackInfo.Deployment)
+			assert.Equal(t, 3, len(stackInfo.Deployment.Resources))
+
+			tests.SortResourcesByURN(stackInfo)
+
+			job := stackInfo.Deployment.Resources[0]
+			provRes := stackInfo.Deployment.Resources[1]
+			stackRes := stackInfo.Deployment.Resources[2]
+
+			assert.Equal(t, resource.RootStackType, stackRes.URN.Type())
+			assert.True(t, providers.IsProviderType(provRes.URN.Type()))
+
+			assert.Equal(t, tokens.Type("kubernetes:batch/v1:Job"), job.URN.Type())
+		},
+		EditDirs: []integration.EditDir{
+			{
+				Dir:           filepath.Join("replace-unready", "step2"),
+				Additive:      true,
+				ExpectFailure: true,
+				ExtraRuntimeValidation: func(t *testing.T, stackInfo integration.RuntimeValidationStackInfo) {
+					assert.NotNil(t, stackInfo.Deployment)
+					assert.Equal(t, 3, len(stackInfo.Deployment.Resources))
+
+					tests.SortResourcesByURN(stackInfo)
+
+					job := stackInfo.Deployment.Resources[0]
+					provRes := stackInfo.Deployment.Resources[1]
+					stackRes := stackInfo.Deployment.Resources[2]
+
+					assert.Equal(t, resource.RootStackType, stackRes.URN.Type())
+					assert.True(t, providers.IsProviderType(provRes.URN.Type()))
+
+					assert.Equal(t, tokens.Type("kubernetes:batch/v1:Job"), job.URN.Type())
+
+					// Check the event stream for a preview showing that the Job will be updated.
+					for _, e := range stackInfo.Events {
+						if e.ResourcePreEvent != nil && e.ResourcePreEvent.Metadata.Type == "kubernetes:batch/v1:Job" {
+							assert.Equal(t, e.ResourcePreEvent.Metadata.Op, apitype.OpUpdate)
+						}
+					}
+				},
+			},
+			{
+				Dir:           filepath.Join("replace-unready", "step3"),
+				Additive:      true,
+				ExpectFailure: true,
+				ExtraRuntimeValidation: func(t *testing.T, stackInfo integration.RuntimeValidationStackInfo) {
+					// The stack has an extra Job now from the failed update in step2.
+					assert.NotNil(t, stackInfo.Deployment)
+					assert.Equal(t, 4, len(stackInfo.Deployment.Resources))
+
+					tests.SortResourcesByURN(stackInfo)
+
+					job := stackInfo.Deployment.Resources[0]
+					jobOld := stackInfo.Deployment.Resources[1]
+					provRes := stackInfo.Deployment.Resources[2]
+					stackRes := stackInfo.Deployment.Resources[3]
+
+					assert.Equal(t, resource.RootStackType, stackRes.URN.Type())
+					assert.True(t, providers.IsProviderType(provRes.URN.Type()))
+
+					assert.Equal(t, tokens.Type("kubernetes:batch/v1:Job"), job.URN.Type())
+					assert.Equal(t, tokens.Type("kubernetes:batch/v1:Job"), jobOld.URN.Type())
+
+					// Check the event stream for a preview showing that the Job will be replaced.
+					for _, e := range stackInfo.Events {
+						if e.ResourcePreEvent != nil && e.ResourcePreEvent.Metadata.Type == "kubernetes:batch/v1:Job" {
+							assert.Equal(t, e.ResourcePreEvent.Metadata.Op, apitype.OpCreateReplacement)
+						}
+					}
+				},
+			},
+		},
+	})
+	integration.ProgramTest(t, &test)
+}
+
 func TestRetry(t *testing.T) {
 	test := baseOptions.With(integration.ProgramTestOptions{
 		Dir:   filepath.Join("retry", "step1"),
diff --git a/tests/sdk/nodejs/replace-unready/step1/Pulumi.yaml b/tests/sdk/nodejs/replace-unready/step1/Pulumi.yaml
new file mode 100644
index 0000000000..5f9aea4c69
--- /dev/null
+++ b/tests/sdk/nodejs/replace-unready/step1/Pulumi.yaml
@@ -0,0 +1,3 @@
+name: replace-unready
+description: A program that tests the replaceUnready annotation.
+runtime: nodejs
diff --git a/tests/sdk/nodejs/replace-unready/step1/index.ts b/tests/sdk/nodejs/replace-unready/step1/index.ts
new file mode 100644
index 0000000000..0f51d15f9b
--- /dev/null
+++ b/tests/sdk/nodejs/replace-unready/step1/index.ts
@@ -0,0 +1,40 @@
+// Copyright 2016-2021, Pulumi Corporation.
+//
+// Licensed 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.
+
+import * as k8s from "@pulumi/kubernetes";
+
+// Create a Job that will fail.
+// The replaceUnready annotation is set to false, so the replace behavior is disabled.
+new k8s.batch.v1.Job("test", {
+  metadata: {
+    annotations: {
+      "pulumi.com/replaceUnready": "false"
+    }
+  },
+  spec: {
+    template: {
+      spec: {
+        containers: [
+          {
+            name: "boom",
+            image: "alpine",
+            command: ["fail"],
+          }
+        ],
+        restartPolicy: "Never"
+      }
+    },
+    backoffLimit: 0
+  }
+});
diff --git a/tests/sdk/nodejs/replace-unready/step1/package.json b/tests/sdk/nodejs/replace-unready/step1/package.json
new file mode 100644
index 0000000000..40ec3e08ee
--- /dev/null
+++ b/tests/sdk/nodejs/replace-unready/step1/package.json
@@ -0,0 +1,12 @@
+{
+    "name": "replace-unready",
+	"version": "0.1.0",
+    "dependencies": {
+        "@pulumi/pulumi": "latest"
+    },
+    "devDependencies": {
+    },
+    "peerDependencies": {
+        "@pulumi/kubernetes": "latest"
+    }
+}
diff --git a/tests/sdk/nodejs/replace-unready/step1/tsconfig.json b/tests/sdk/nodejs/replace-unready/step1/tsconfig.json
new file mode 100644
index 0000000000..5dacccbd42
--- /dev/null
+++ b/tests/sdk/nodejs/replace-unready/step1/tsconfig.json
@@ -0,0 +1,22 @@
+{
+    "compilerOptions": {
+        "outDir": "bin",
+        "target": "es6",
+        "module": "commonjs",
+        "moduleResolution": "node",
+        "declaration": true,
+        "sourceMap": true,
+        "stripInternal": true,
+        "experimentalDecorators": true,
+        "pretty": true,
+        "noFallthroughCasesInSwitch": true,
+        "noImplicitAny": true,
+        "noImplicitReturns": true,
+        "forceConsistentCasingInFileNames": true,
+        "strictNullChecks": true
+    },
+    "files": [
+        "index.ts"
+    ]
+}
+
diff --git a/tests/sdk/nodejs/replace-unready/step2/index.ts b/tests/sdk/nodejs/replace-unready/step2/index.ts
new file mode 100644
index 0000000000..0f51d15f9b
--- /dev/null
+++ b/tests/sdk/nodejs/replace-unready/step2/index.ts
@@ -0,0 +1,40 @@
+// Copyright 2016-2021, Pulumi Corporation.
+//
+// Licensed 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.
+
+import * as k8s from "@pulumi/kubernetes";
+
+// Create a Job that will fail.
+// The replaceUnready annotation is set to false, so the replace behavior is disabled.
+new k8s.batch.v1.Job("test", {
+  metadata: {
+    annotations: {
+      "pulumi.com/replaceUnready": "false"
+    }
+  },
+  spec: {
+    template: {
+      spec: {
+        containers: [
+          {
+            name: "boom",
+            image: "alpine",
+            command: ["fail"],
+          }
+        ],
+        restartPolicy: "Never"
+      }
+    },
+    backoffLimit: 0
+  }
+});
diff --git a/tests/sdk/nodejs/replace-unready/step3/index.ts b/tests/sdk/nodejs/replace-unready/step3/index.ts
new file mode 100644
index 0000000000..7376d2b881
--- /dev/null
+++ b/tests/sdk/nodejs/replace-unready/step3/index.ts
@@ -0,0 +1,40 @@
+// Copyright 2016-2021, Pulumi Corporation.
+//
+// Licensed 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.
+
+import * as k8s from "@pulumi/kubernetes";
+
+// Create a Job that will fail.
+// The replaceUnready annotation is set to true, so the replace behavior is enabled.
+new k8s.batch.v1.Job("test", {
+  metadata: {
+    annotations: {
+      "pulumi.com/replaceUnready": "true"
+    }
+  },
+  spec: {
+    template: {
+      spec: {
+        containers: [
+          {
+            name: "boom",
+            image: "alpine",
+            command: ["fail"],
+          }
+        ],
+        restartPolicy: "Never"
+      }
+    },
+    backoffLimit: 0
+  }
+});