diff --git a/airflow/api_connexion/schemas/task_instance_schema.py b/airflow/api_connexion/schemas/task_instance_schema.py
index 02dc1fb3f6fa6..c9084151d88ef 100644
--- a/airflow/api_connexion/schemas/task_instance_schema.py
+++ b/airflow/api_connexion/schemas/task_instance_schema.py
@@ -155,7 +155,7 @@ def validate_form(self, data, **kwargs):
 class SetTaskInstanceStateFormSchema(Schema):
     """Schema for handling the request of setting state of task instance of a DAG."""
 
-    dry_run = fields.Boolean(dump_default=True)
+    dry_run = fields.Boolean(load_default=True)
     task_id = fields.Str(required=True)
     execution_date = fields.DateTime(validate=validate_istimezone)
     dag_run_id = fields.Str()
diff --git a/tests/api_connexion/schemas/test_task_instance_schema.py b/tests/api_connexion/schemas/test_task_instance_schema.py
index 76047498d27a3..55acbb8b958a0 100644
--- a/tests/api_connexion/schemas/test_task_instance_schema.py
+++ b/tests/api_connexion/schemas/test_task_instance_schema.py
@@ -237,6 +237,22 @@ def test_success(self):
         }
         assert expected_result == result
 
+    def test_dry_run_is_optional(self):
+        data = self.current_input.copy()
+        data.pop("dry_run")
+        result = set_task_instance_state_form.load(self.current_input)
+        expected_result = {
+            "dry_run": True,
+            "execution_date": dt.datetime(2020, 1, 1, 0, 0, tzinfo=dt.timezone(dt.timedelta(0), "+0000")),
+            "include_downstream": True,
+            "include_future": True,
+            "include_past": True,
+            "include_upstream": True,
+            "new_state": "failed",
+            "task_id": "print_the_context",
+        }
+        assert expected_result == result
+
     @pytest.mark.parametrize(
         "override_data",
         [