Fix Neptune operator execute_complete missing error handling

Both NeptuneStartDbClusterOperator and NeptuneStopDbClusterOperator
silently succeeded when the trigger reported a failure — no status
validation, so the task was marked success and downstream tasks would
start against a cluster that wasn't actually running.

Also fixes wrong XCom key (cluster_id → db_cluster_id) that caused
empty cluster ID in return value, and removes debug self.log.info(event).

Author: shivaam

providers/amazon/src/airflow/providers/amazon/aws/operators/neptune.py

@@ -187,14 +187,11 @@ def execute(self, context: Context, event: dict[str, Any] | None = None, **kwarg
         return {"db_cluster_id": self.cluster_id}
 
     def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> dict[str, str]:
-        status = ""
-        cluster_id = ""
+        if event and event.get("status") != "success":
+            raise AirflowException(f"Error starting Neptune cluster: {event}")
 
-        if event:
-            status = event.get("status", "")
-            cluster_id = event.get("cluster_id", "")
-
-        self.log.info("Neptune cluster %s available with status: %s", cluster_id, status)
+        cluster_id = event.get("db_cluster_id", "") if event else ""
+        self.log.info("Neptune cluster %s available.", cluster_id)
 
         return {"db_cluster_id": cluster_id}
 
@@ -314,13 +311,10 @@ def execute(self, context: Context, event: dict[str, Any] | None = None, **kwarg
         return {"db_cluster_id": self.cluster_id}
 
     def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> dict[str, str]:
-        status = ""
-        cluster_id = ""
-        self.log.info(event)
-        if event:
-            status = event.get("status", "")
-            cluster_id = event.get("cluster_id", "")
-
-        self.log.info("Neptune cluster %s stopped with status: %s", cluster_id, status)
+        if event and event.get("status") != "success":
+            raise AirflowException(f"Error stopping Neptune cluster: {event}")
+
+        cluster_id = event.get("db_cluster_id", "") if event else ""
+        self.log.info("Neptune cluster %s stopped.", cluster_id)
 
         return {"db_cluster_id": cluster_id}