Add AsyncOperationInProgress to ExternalObservation

When AsyncOperationInProgress is true, the managed reconciler sets
Synced=False with reason ReconcilePending instead of ReconcileSuccess.
This prevents a false Synced=True signal during long-running async
operations (e.g. MSK cluster instance type changes).

Note: go.mod contains a temporary replace directive pointing at
ajnye/crossplane for the ReconcilePending condition constructor.
This will be updated to the upstream commit once
crossplane/crossplane#7352 merges.

Fixes #941

Signed-off-by: Aj Nye <aj.nye@appian.com>

Author: ajnye

go.mod

@@ -200,3 +200,5 @@ require (
 	sigs.k8s.io/randfill v1.0.0 // indirect
 	sigs.k8s.io/structured-merge-diff/v6 v6.3.2-0.20260122202528-d9cc6641c482 // indirect
 )
+
+replace github.com/crossplane/crossplane/apis/v2 => github.com/ajnye/crossplane/apis/v2 v2.0.0-20260429153315-46c3b9a56f05

go.sum

@@ -66,6 +66,8 @@ github.com/AzureAD/microsoft-authentication-library-for-go v1.6.0 h1:XRzhVemXdgv
 github.com/AzureAD/microsoft-authentication-library-for-go v1.6.0/go.mod h1:HKpQxkWaGLJ+D/5H8QRpyQXA1eKjxkFlOMwck5+33Jk=
 github.com/Masterminds/semver/v3 v3.4.0 h1:Zog+i5UMtVoCU8oKka5P7i9q9HgrJeGzI9SA1Xbatp0=
 github.com/Masterminds/semver/v3 v3.4.0/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM=
+github.com/ajnye/crossplane/apis/v2 v2.0.0-20260429153315-46c3b9a56f05 h1:oLTNu18VEUJ4MIwUDu7R6Mogw63TiwCgLiAYrcQLctM=
+github.com/ajnye/crossplane/apis/v2 v2.0.0-20260429153315-46c3b9a56f05/go.mod h1:h7KE74Z4TFs1L/FFv3RdsiG9Uax7L56oHpcggSZnONg=
 github.com/alessio/shellescape v1.4.1 h1:V7yhSDDn8LP4lc4jS8pFkt0zCnzVJlG5JXy9BVKJUX0=
 github.com/alessio/shellescape v1.4.1/go.mod h1:PZAiSCk0LJaZkiCSkPv8qIobYglO3FPpyFjDCtHLS30=
 github.com/antlr4-go/antlr/v4 v4.13.1 h1:SqQKkuVZ+zWkMMNkjy5FZe5mr5WURWnlpmOuzYWrPrQ=
@@ -136,8 +138,6 @@ github.com/coreos/go-oidc/v3 v3.17.0/go.mod h1:wqPbKFrVnE90vty060SB40FCJ8fTHTxSw
 github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g=
 github.com/creack/pty v1.1.18 h1:n56/Zwd5o6whRC5PMGretI4IdRLlmBXYNjScPaBgsbY=
 github.com/creack/pty v1.1.18/go.mod h1:MOBLtS5ELjhRRrroQr9kyvTxUAFNvYEK993ew/Vr4O4=
-github.com/crossplane/crossplane/apis/v2 v2.0.0-20260424160951-8f231230ebb6 h1:9ki6AJQgBJIcLNjK+scUZp2ZDenuAo18d0JSNOlkY2Y=
-github.com/crossplane/crossplane/apis/v2 v2.0.0-20260424160951-8f231230ebb6/go.mod h1:h7KE74Z4TFs1L/FFv3RdsiG9Uax7L56oHpcggSZnONg=
 github.com/cyberphone/json-canonicalization v0.0.0-20241213102144-19d51d7fe467 h1:uX1JmpONuD549D73r6cgnxyUu18Zb7yHAy5AYU0Pm4Q=
 github.com/cyberphone/json-canonicalization v0.0.0-20241213102144-19d51d7fe467/go.mod h1:uzvlm1mxhHkdfqitSA92i7Se+S9ksOn3a3qmv/kyOCw=
 github.com/danieljoos/wincred v1.2.3 h1:v7dZC2x32Ut3nEfRH+vhoZGvN72+dQ/snVXo/vMFLdQ=

pkg/reconciler/managed/reconciler.go

@@ -536,6 +536,13 @@ type ExternalObservation struct {
 	// finding where the observed diverges from the desired state.
 	// The string should be a cmp.Diff that details the difference.
 	Diff string
+
+	// AsyncOperationInProgress indicates that an asynchronous operation
+	// (e.g. a long-running cloud API call) is currently in progress for
+	// this resource. When true, the managed reconciler will set
+	// Synced=False with reason ReconcilePending instead of
+	// ReconcileSuccess, and will not call Update().
+	AsyncOperationInProgress bool
 }
 
 // An ExternalCreation is the result of the creation of an external resource.
@@ -1496,8 +1503,14 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (resu
 		// https://github.com/crossplane/crossplane/issues/289
 		reconcileAfter := r.pollIntervalHook(managed, r.effectivePollInterval(managed))
 		log.Debug("External resource is up to date", "requeue-after", time.Now().Add(reconcileAfter))
-		status.MarkConditions(xpv2.ReconcileSuccess())
-		r.metricRecorder.recordFirstTimeReady(managed)
+
+		if observation.AsyncOperationInProgress {
+			log.Debug("Async operation in progress, setting ReconcilePending")
+			status.MarkConditions(xpv2.ReconcilePending("Async operation in progress"))
+		} else {
+			status.MarkConditions(xpv2.ReconcileSuccess())
+			r.metricRecorder.recordFirstTimeReady(managed)
+		}
 
 		// record that we intentionally did not update the managed resource
 		// because no drift was detected. We call this so late in the reconcile

pkg/reconciler/managed/reconciler_legacy_test.go

@@ -1121,6 +1121,47 @@ func TestReconciler(t *testing.T) {
 			},
 			want: want{result: reconcile.Result{RequeueAfter: defaultPollInterval}},
 		},
+		"ExternalResourceUpToDateAsyncOperationInProgress": {
+			reason: "When an async operation is in progress, Synced should be set to ReconcilePending instead of ReconcileSuccess.",
+			args: args{
+				m: &fake.Manager{
+					Client: &test.MockClient{
+						MockGet: legacyManagedMockGetFn(nil, 42),
+						MockStatusUpdate: test.MockSubResourceUpdateFn(func(_ context.Context, obj client.Object, _ ...client.SubResourceUpdateOption) error {
+							want := newLegacyManaged(42)
+							want.SetConditions(xpv2.ReconcilePending("Async operation in progress").WithObservedGeneration(42))
+
+							if diff := cmp.Diff(want, obj, test.EquateConditions()); diff != "" {
+								reason := "An async-in-progress reconcile should set ReconcilePending, not ReconcileSuccess."
+								t.Errorf("\nReason: %s\n-want, +got:\n%s", reason, diff)
+							}
+
+							return nil
+						}),
+					},
+					Scheme: fake.SchemeWith(&fake.LegacyManaged{}),
+				},
+				mg: resource.ManagedKind(fake.GVK(&fake.LegacyManaged{})),
+				o: []ReconcilerOption{
+					WithInitializers(),
+					WithReferenceResolver(ReferenceResolverFn(func(_ context.Context, _ resource.Managed) error { return nil })),
+					WithExternalConnector(ExternalConnectorFn(func(_ context.Context, _ resource.Managed) (ExternalClient, error) {
+						return &ExternalClientFns{
+							ObserveFn: func(_ context.Context, _ resource.Managed) (ExternalObservation, error) {
+								return ExternalObservation{
+									ResourceExists:           true,
+									ResourceUpToDate:         true,
+									AsyncOperationInProgress: true,
+								}, nil
+							},
+							DisconnectFn: func(_ context.Context) error { return nil },
+						}, nil
+					})),
+					WithFinalizer(resource.FinalizerFns{AddFinalizerFn: func(_ context.Context, _ resource.Object) error { return nil }}),
+				},
+			},
+			want: want{result: reconcile.Result{RequeueAfter: defaultPollInterval}},
+		},
 		"ExternalResourceUpToDateWithJitter": {
 			reason: "When the external resource exists and is up to date a requeue should be triggered after a long wait with jitter added.",
 			args: args{

pkg/reconciler/managed/reconciler_modern_test.go

@@ -1127,6 +1127,47 @@ func TestModernReconciler(t *testing.T) {
 			},
 			want: want{result: reconcile.Result{RequeueAfter: defaultPollInterval}},
 		},
+		"ExternalResourceUpToDateAsyncOperationInProgress": {
+			reason: "When an async operation is in progress, Synced should be set to ReconcilePending instead of ReconcileSuccess.",
+			args: args{
+				m: &fake.Manager{
+					Client: &test.MockClient{
+						MockGet: modernManagedMockGetFn(nil, 42),
+						MockStatusUpdate: test.MockSubResourceUpdateFn(func(_ context.Context, obj client.Object, _ ...client.SubResourceUpdateOption) error {
+							want := newModernManaged(42)
+							want.SetConditions(xpv2.ReconcilePending("Async operation in progress").WithObservedGeneration(42))
+
+							if diff := cmp.Diff(want, obj, test.EquateConditions()); diff != "" {
+								reason := "An async-in-progress reconcile should set ReconcilePending, not ReconcileSuccess."
+								t.Errorf("\nReason: %s\n-want, +got:\n%s", reason, diff)
+							}
+
+							return nil
+						}),
+					},
+					Scheme: fake.SchemeWith(&fake.ModernManaged{}),
+				},
+				mg: resource.ManagedKind(fake.GVK(&fake.ModernManaged{})),
+				o: []ReconcilerOption{
+					WithInitializers(),
+					WithReferenceResolver(ReferenceResolverFn(func(_ context.Context, _ resource.Managed) error { return nil })),
+					WithExternalConnector(ExternalConnectorFn(func(_ context.Context, _ resource.Managed) (ExternalClient, error) {
+						return &ExternalClientFns{
+							ObserveFn: func(_ context.Context, _ resource.Managed) (ExternalObservation, error) {
+								return ExternalObservation{
+									ResourceExists:           true,
+									ResourceUpToDate:         true,
+									AsyncOperationInProgress: true,
+								}, nil
+							},
+							DisconnectFn: func(_ context.Context) error { return nil },
+						}, nil
+					})),
+					WithFinalizer(resource.FinalizerFns{AddFinalizerFn: func(_ context.Context, _ resource.Object) error { return nil }}),
+				},
+			},
+			want: want{result: reconcile.Result{RequeueAfter: defaultPollInterval}},
+		},
 		"ExternalResourceUpToDateWithJitter": {
 			reason: "When the external resource exists and is up to date a requeue should be triggered after a long wait with jitter added.",
 			args: args{