diff --git a/.changelog/2163.txt b/.changelog/2163.txt
new file mode 100644
index 0000000000..7cfed89cd5
--- /dev/null
+++ b/.changelog/2163.txt
@@ -0,0 +1,3 @@
+```release-note:bug
+`manifest/provider/apply.go`: update flow in `wait` block to fix timeout bug within tf apply where the resource is created and appears in Kubernetes but does not appear in TF state file after deadline. The fix would ensure that the resource has been created in the state file while also tainting the resource requiring the user to make the necessary changes in order for their to not be another timeout error.
+```
\ No newline at end of file
diff --git a/manifest/provider/apply.go b/manifest/provider/apply.go
index 68495d8949..22451dc5ed 100644
--- a/manifest/provider/apply.go
+++ b/manifest/provider/apply.go
@@ -425,12 +425,12 @@ func (s *RawProviderServer) ApplyResourceChange(ctx context.Context, req *tfprot
 		if !waitConfig.IsNull() {
 			err = s.waitForCompletion(ctxDeadline, waitConfig, rs, rname, wt, th)
 			if err != nil {
-				if err == context.DeadlineExceeded {
+				if reason, ok := err.(WaiterError); ok {
 					resp.Diagnostics = append(resp.Diagnostics,
 						&tfprotov5.Diagnostic{
 							Severity: tfprotov5.DiagnosticSeverityError,
 							Summary:  "Operation timed out",
-							Detail:   "Terraform timed out waiting on the operation to complete",
+							Detail:   reason.Error(),
 						})
 				} else {
 					resp.Diagnostics = append(resp.Diagnostics,
@@ -439,11 +439,10 @@ func (s *RawProviderServer) ApplyResourceChange(ctx context.Context, req *tfprot
 							Summary:  "Error waiting for operation to complete",
 							Detail:   err.Error(),
 						})
+					return resp, nil
 				}
-				return resp, nil
 			}
-
-			r, err := rs.Get(ctxDeadline, rname, metav1.GetOptions{})
+			r, err := rs.Get(ctx, rname, metav1.GetOptions{})
 			if err != nil {
 				s.logger.Error("[ApplyResourceChange][ReadAfterWait]", "API error", dump(err), "API response", dump(result))
 				resp.Diagnostics = append(resp.Diagnostics,
@@ -452,7 +451,6 @@ func (s *RawProviderServer) ApplyResourceChange(ctx context.Context, req *tfprot
 						Summary:  fmt.Sprintf(`Failed to read resource %q after wait conditions`, rname),
 						Detail:   err.Error(),
 					})
-
 				return resp, nil
 			}
 			result = r
diff --git a/manifest/provider/waiter.go b/manifest/provider/waiter.go
index e4e15729bd..43a6e122d9 100644
--- a/manifest/provider/waiter.go
+++ b/manifest/provider/waiter.go
@@ -42,6 +42,14 @@ type Waiter interface {
 	Wait(context.Context) error
 }
 
+type WaiterError struct {
+	Reason string
+}
+
+func (e WaiterError) Error() string {
+	return fmt.Sprintf("timed out waiting on %v", e.Reason)
+}
+
 // NewResourceWaiter constructs an appropriate Waiter using the supplied waitForBlock configuration
 func NewResourceWaiter(resource dynamic.ResourceInterface, resourceName string, resourceType tftypes.Type, th map[string]string, waitForBlock tftypes.Value, hl hclog.Logger) (Waiter, error) {
 	var waitForBlockVal map[string]tftypes.Value
@@ -145,7 +153,7 @@ func (w *FieldWaiter) Wait(ctx context.Context) error {
 	for {
 		if deadline, ok := ctx.Deadline(); ok {
 			if time.Now().After(deadline) {
-				return context.DeadlineExceeded
+				return WaiterError{Reason: "field matchers"}
 			}
 		}
 
@@ -283,7 +291,7 @@ func (w *RolloutWaiter) Wait(ctx context.Context) error {
 	for {
 		if deadline, ok := ctx.Deadline(); ok {
 			if time.Now().After(deadline) {
-				return context.DeadlineExceeded
+				return WaiterError{Reason: "rollout to complete"}
 			}
 		}
 
@@ -333,7 +341,7 @@ func (w *ConditionsWaiter) Wait(ctx context.Context) error {
 	for {
 		if deadline, ok := ctx.Deadline(); ok {
 			if time.Now().After(deadline) {
-				return context.DeadlineExceeded
+				return WaiterError{Reason: "conditions"}
 			}
 		}
 
diff --git a/manifest/test/acceptance/wait_test.go b/manifest/test/acceptance/wait_test.go
index 7c652aefdb..6eb7cd916d 100644
--- a/manifest/test/acceptance/wait_test.go
+++ b/manifest/test/acceptance/wait_test.go
@@ -219,9 +219,18 @@ func TestKubernetesManifest_Wait_InvalidCondition(t *testing.T) {
 	tf.Init(ctx)
 
 	err = tf.Apply(ctx)
-	if err == nil || !strings.Contains(err.Error(), "Terraform timed out waiting on the operation to complete") {
+	if err == nil || !strings.Contains(err.Error(), "timed out waiting on") {
 		t.Fatalf("Waiter should have timed out")
 	}
+
+	st, err := tf.State(ctx)
+	if err != nil {
+		t.Fatalf("Failed to get state: %q", err)
+	}
+	tfstate := tfstatehelper.NewHelper(st)
+	if !tfstate.ResourceExists(t, "kubernetes_manifest.test") {
+		t.Fatalf("Expected resource to exist in state")
+	}
 }
 
 func TestKubernetesManifest_WaitFields_Annotations(t *testing.T) {
diff --git a/manifest/test/helper/state/state_helper.go b/manifest/test/helper/state/state_helper.go
index 2d38d1c64d..7a2d6b9e1d 100644
--- a/manifest/test/helper/state/state_helper.go
+++ b/manifest/test/helper/state/state_helper.go
@@ -26,6 +26,11 @@ type Helper struct {
 func NewHelper(tfstate *tfjson.State) *Helper {
 	return &Helper{tfstate}
 }
+func (s *Helper) ResourceExists(t *testing.T, resourceAddress string) bool {
+	t.Helper()
+	_, err := getAttributesValuesFromResource(s, resourceAddress)
+	return err == nil
+}
 
 // getAttributesValuesFromResource pulls out the AttributeValues field from the resource at the given address
 func getAttributesValuesFromResource(state *Helper, address string) (interface{}, error) {