diff --git a/parsl/executors/high_throughput/executor.py b/parsl/executors/high_throughput/executor.py
index 7e4ae2ad2e..5407346ad8 100644
--- a/parsl/executors/high_throughput/executor.py
+++ b/parsl/executors/high_throughput/executor.py
@@ -645,7 +645,8 @@ def submit(self, func, resource_specification, *args, **kwargs):
         Returns:
               Future
         """
-        validate_resource_spec(resource_specification)
+
+        validate_resource_spec(resource_specification, self.enable_mpi_mode)
 
         if self.bad_state_is_set:
             raise self.executor_exception
diff --git a/parsl/executors/high_throughput/mpi_prefix_composer.py b/parsl/executors/high_throughput/mpi_prefix_composer.py
index 8b38be8549..07abf8f9d5 100644
--- a/parsl/executors/high_throughput/mpi_prefix_composer.py
+++ b/parsl/executors/high_throughput/mpi_prefix_composer.py
@@ -8,8 +8,18 @@
                    'mpiexec')
 
 
+class MissingResourceSpecification(Exception):
+    """Exception raised when input is  not supplied a resource specification"""
+
+    def __init__(self, reason: str):
+        self.reason = reason
+
+    def __str__(self):
+        return f"Missing resource specification: {self.reason}"
+
+
 class InvalidResourceSpecification(Exception):
-    """Exception raised when Invalid keys are supplied via resource specification"""
+    """Exception raised when Invalid input is supplied via resource specification"""
 
     def __init__(self, invalid_keys: Set[str]):
         self.invalid_keys = invalid_keys
@@ -18,13 +28,19 @@ def __str__(self):
         return f"Invalid resource specification options supplied: {self.invalid_keys}"
 
 
-def validate_resource_spec(resource_spec: Dict[str, str]):
+def validate_resource_spec(resource_spec: Dict[str, str], is_mpi_enabled: bool):
     """Basic validation of keys in the resource_spec
 
     Raises: InvalidResourceSpecification if the resource_spec
         is invalid (e.g, contains invalid keys)
     """
     user_keys = set(resource_spec.keys())
+
+    # empty resource_spec when mpi_mode is set causes parsl to hang
+    # ref issue #3427
+    if is_mpi_enabled and len(user_keys) == 0:
+        raise MissingResourceSpecification('MPI mode requires optional parsl_resource_specification keyword argument to be configured')
+
     legal_keys = set(("ranks_per_node",
                       "num_nodes",
                       "num_ranks",
diff --git a/parsl/tests/test_mpi_apps/test_mpi_mode_enabled.py b/parsl/tests/test_mpi_apps/test_mpi_mode_enabled.py
index 86a7fdb9ec..c3d1bcfb26 100644
--- a/parsl/tests/test_mpi_apps/test_mpi_mode_enabled.py
+++ b/parsl/tests/test_mpi_apps/test_mpi_mode_enabled.py
@@ -6,6 +6,8 @@
 from parsl import python_app, bash_app
 from parsl.tests.configs.htex_local import fresh_config
 
+from parsl.executors.high_throughput.mpi_prefix_composer import MissingResourceSpecification
+
 import os
 
 EXECUTOR_LABEL = "MPI_TEST"
@@ -168,3 +170,11 @@ def test_simulated_load(rounds: int = 100):
         total_ranks, nodes = future.result(timeout=10)
         assert len(nodes) == futures[future]["num_nodes"]
         assert total_ranks == futures[future]["num_nodes"] * futures[future]["ranks_per_node"]
+
+
+@pytest.mark.local
+def test_missing_resource_spec():
+
+    with pytest.raises(MissingResourceSpecification):
+        future = mock_app(sleep_dur=0.4)
+        future.result(timeout=10)
diff --git a/parsl/tests/test_mpi_apps/test_resource_spec.py b/parsl/tests/test_mpi_apps/test_resource_spec.py
index 4bef22cfff..7e533c128f 100644
--- a/parsl/tests/test_mpi_apps/test_resource_spec.py
+++ b/parsl/tests/test_mpi_apps/test_resource_spec.py
@@ -19,7 +19,8 @@
 )
 from parsl.executors.high_throughput.mpi_prefix_composer import (
     validate_resource_spec,
-    InvalidResourceSpecification
+    InvalidResourceSpecification,
+    MissingResourceSpecification
 )
 
 EXECUTOR_LABEL = "MPI_TEST"
@@ -122,18 +123,22 @@ def test_top_level():
 
 @pytest.mark.local
 @pytest.mark.parametrize(
-    "resource_spec, exception",
+    "resource_spec, is_mpi_enabled, exception",
     (
-        ({"num_nodes": 2, "ranks_per_node": 1}, None),
-        ({"launcher_options": "--debug_foo"}, None),
-        ({"num_nodes": 2, "BAD_OPT": 1}, InvalidResourceSpecification),
-        ({}, None),
+        ({"num_nodes": 2, "ranks_per_node": 1}, False, None),
+        ({"launcher_options": "--debug_foo"}, False, None),
+        ({"num_nodes": 2, "BAD_OPT": 1}, False, InvalidResourceSpecification),
+        ({}, False, None),
+        ({"num_nodes": 2, "ranks_per_node": 1}, True, None),
+        ({"launcher_options": "--debug_foo"}, True, None),
+        ({"num_nodes": 2, "BAD_OPT": 1}, True, InvalidResourceSpecification),
+        ({}, True, MissingResourceSpecification),
     )
 )
-def test_resource_spec(resource_spec: Dict, exception):
+def test_resource_spec(resource_spec: Dict, is_mpi_enabled: bool, exception):
     if exception:
         with pytest.raises(exception):
-            validate_resource_spec(resource_spec)
+            validate_resource_spec(resource_spec, is_mpi_enabled)
     else:
-        result = validate_resource_spec(resource_spec)
+        result = validate_resource_spec(resource_spec, is_mpi_enabled)
         assert result is None