diff --git a/sdk/python/kfp/compiler/compiler.py b/sdk/python/kfp/compiler/compiler.py
index 879868700c5..3df52b2b92e 100644
--- a/sdk/python/kfp/compiler/compiler.py
+++ b/sdk/python/kfp/compiler/compiler.py
@@ -673,6 +673,10 @@ def _create_dag_templates(self, pipeline, op_transformers=None, op_to_templates_
       templates.append(template)
 
     for op in pipeline.ops.values():
+      if hasattr(op, 'importer_spec'):
+        raise NotImplementedError(
+            'dsl.importer is not supported for Kubeflow Pipelines open source yet.')
+
       if self._mode == dsl.PipelineExecutionMode.V2_COMPATIBLE:
         v2_compat.update_op(op,
                             pipeline_name=self._pipeline_name_param,
diff --git a/sdk/python/kfp/compiler/v2_compatible_compiler_test.py b/sdk/python/kfp/compiler/v2_compatible_compiler_test.py
index f3f14ae8ac9..e30d2682e73 100644
--- a/sdk/python/kfp/compiler/v2_compatible_compiler_test.py
+++ b/sdk/python/kfp/compiler/v2_compatible_compiler_test.py
@@ -136,5 +136,19 @@ def my_pipeline():
       compiler.Compiler(mode=dsl.PipelineExecutionMode.V2_COMPATIBLE).compile(
           pipeline_func=my_pipeline, package_path='result.json')
 
+  def test_use_importer_should_error(self):
+
+    @dsl.pipeline(name='test-pipeline')
+    def my_pipeline():
+      dsl.importer(artifact_uri='dummy', artifact_class=dsl.io_types.Artifact)
+
+    with self.assertRaisesRegex(
+        NotImplementedError,
+        'dsl.importer is not supported for Kubeflow Pipelines open source yet.',
+    ):
+      compiler.Compiler(mode=dsl.PipelineExecutionMode.V2_COMPATIBLE).compile(
+          pipeline_func=my_pipeline, package_path='result.json')
+
+
 if __name__ == '__main__':
   unittest.main()
diff --git a/sdk/python/tests/compiler/compiler_tests.py b/sdk/python/tests/compiler/compiler_tests.py
index 750ae747bc7..db9bf0c9b14 100644
--- a/sdk/python/tests/compiler/compiler_tests.py
+++ b/sdk/python/tests/compiler/compiler_tests.py
@@ -1275,5 +1275,18 @@ def remove_metadata(yaml) -> None:
     # compare
     self.assertEqual(pipeline_yaml_arg, pipeline_yaml_kwarg)
 
+  def test_use_importer_should_error(self):
+
+    @dsl.pipeline(name='test-pipeline')
+    def my_pipeline():
+      dsl.importer(artifact_uri='dummy', artifact_class=dsl.io_types.Artifact)
+
+    with self.assertRaisesRegex(
+        NotImplementedError,
+        'dsl.importer is not supported for Kubeflow Pipelines open source yet.',
+    ):
+      kfp.compiler.Compiler().compile(
+          pipeline_func=my_pipeline, package_path='result.json')
+
 if __name__ == '__main__':
   unittest.main()