Hi @varshini_reddy   ,

Could you clarify what you want to achieve:

1. If the job run fails, then you want to stop the job (so it does not run the next time). That what covered my answer
2. If the job run fails, they you want to stop the job run. That is what covered @szymon_dybczak 
3. You have a "For each" task in your job, and you want to stop the job run, if one of the iterations failed? 
   • If that's the case then you can create a Delta table and to keep there Job Run Failures. 
   • It can basically have just 2 columns, Job Root Run Id and Status
   • In your notebook that is being executed by the For each have if statement to check first whether there is an entry in the table where Job Root Run Id = <current job root run id>  and Status = Failure
   • Wrap the remaining code in your notebook in try/except clause. In except insert the data into your Delta table
   • You can get the Job Root Run Id using this code snippet: 

# Retrieve the job root run ID using dbutils
import json
# Get the context object
context_json = dbutils.notebook.entry_point.getDbutils().notebook().getContext().toJson()

# Load the JSON object
context_dict = json.loads(context_json)

# Extract the rootRunId
root_run_id = context_dict.get("rootRunId").get("id")

print(root_run_id)​

The code:

CREATE TABLE IF NOT EXISTS job_run_failures (
job_root_run_id STRING,
status STRING -- e.g., "SUCCESS", "FAILURE"
) USING DELTA;
import json
from datetime import datetime​

# Get the context object
context_json = dbutils.notebook.entry_point.getDbutils().notebook().getContext().toJson()

# Load the JSON object
context_dict = json.loads(context_json)

# Retrieve the current job root run ID
job_root_run_id = context_dict.get("rootRunId").get("id")

# Check if a failure is already logged for this job run
existing_failure = spark.sql(f"""
SELECT COUNT(*)
FROM job_run_failures
WHERE job_root_run_id = '{job_root_run_id}' AND status = 'FAILURE'
""").collect()[0][0]

# If a failure is detected, stop execution
if existing_failure > 0:
dbutils.notebook.exit("Exiting due to a previously logged failure.")

try:
# Your main code logic here
print("Executing main task...")
# Simulate a task failure
# raise ValueError("Simulated task failure") # Uncomment to test failure handling

except Exception as e:
# Log the failure to the Delta table using MERGE to handle concurrency
spark.sql(f"""
MERGE INTO job_run_failures AS target
USING (SELECT '{job_root_run_id}' AS job_root_run_id, 'FAILURE' AS status) AS source
ON target.job_root_run_id = source.job_root_run_id
WHEN NOT MATCHED THEN
INSERT (job_root_run_id, status)
VALUES (source.job_root_run_id, source.status)
""")
# Exit the notebook with an error message
dbutils.notebook.exit(f"FAILED: {str(e)}")