diff --git a/airflow/www/views.py b/airflow/www/views.py
index f9cebf90b7612..ae1520e50a6b7 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -1419,12 +1419,21 @@ def get_logs_with_metadata(self, session=None):
         map_index = request.args.get('map_index', -1, type=int)
         try_number = request.args.get('try_number', type=int)
         metadata = request.args.get('metadata')
-        metadata = json.loads(metadata)
         response_format = request.args.get('format', 'json')
 
         # metadata may be null
         if not metadata:
             metadata = {}
+        else:
+            # Validate JSON metadata
+            try:
+                metadata = json.loads(metadata)
+            except json.decoder.JSONDecodeError:
+                error_message = "Invalid JSON metadata"
+                response = jsonify({"error": error_message})
+                response.status_code = 400
+
+                return response
 
         # Convert string datetime into actual datetime
         try:
diff --git a/tests/www/views/test_views_log.py b/tests/www/views/test_views_log.py
index 3460d74304567..f697cd3772c28 100644
--- a/tests/www/views/test_views_log.py
+++ b/tests/www/views/test_views_log.py
@@ -337,6 +337,26 @@ def test_get_logs_with_metadata(log_admin_client, metadata):
     assert 'Log for testing.' in data
 
 
+def test_get_logs_with_invalid_metadata(log_admin_client):
+    """Test invalid metadata JSON returns error message"""
+    metadata = "invalid"
+    url_template = "get_logs_with_metadata?dag_id={}&task_id={}&execution_date={}&try_number={}&metadata={}"
+    response = log_admin_client.get(
+        url_template.format(
+            DAG_ID,
+            TASK_ID,
+            urllib.parse.quote_plus(DEFAULT_DATE.isoformat()),
+            1,
+            metadata,
+        ),
+        data={"username": "test", "password": "test"},
+        follow_redirects=True,
+    )
+
+    assert response.status_code == 400
+    assert response.json == {"error": "Invalid JSON metadata"}
+
+
 @unittest.mock.patch(
     "airflow.utils.log.file_task_handler.FileTaskHandler.read",
     return_value=(['airflow log line'], [{'end_of_log': True}]),