diff --git a/singer_sdk/target_base.py b/singer_sdk/target_base.py
index d8320394a..facd69c87 100644
--- a/singer_sdk/target_base.py
+++ b/singer_sdk/target_base.py
@@ -159,7 +159,6 @@ def get_sink(
         """
         _ = record  # Custom implementations may use record in sink selection.
         if schema is None:
-            self._assert_sink_exists(stream_name)
             return self._sinks_active[stream_name]
 
         existing_sink = self._sinks_active.get(stream_name, None)
@@ -262,7 +261,8 @@ def _assert_sink_exists(self, stream_name: str) -> None:
         if not self.sink_exists(stream_name):
             msg = (
                 f"A record for stream '{stream_name}' was encountered before a "
-                "corresponding schema."
+                "corresponding schema. Check that the Tap correctly implements "
+                "the Singer spec."
             )
             raise RecordsWithoutSchemaException(msg)
 
@@ -317,6 +317,8 @@ def _process_record_message(self, message_dict: dict) -> None:
         self._assert_line_requires(message_dict, requires={"stream", "record"})
 
         stream_name = message_dict["stream"]
+        self._assert_sink_exists(stream_name)
+
         for stream_map in self.mapper.stream_maps[stream_name]:
             raw_record = copy.copy(message_dict["record"])
             transformed_record = stream_map.transform(raw_record)
diff --git a/singer_sdk/testing/target_tests.py b/singer_sdk/testing/target_tests.py
index 8b582ed12..e1bd6662b 100644
--- a/singer_sdk/testing/target_tests.py
+++ b/singer_sdk/testing/target_tests.py
@@ -4,6 +4,8 @@
 
 import pytest
 
+from singer_sdk.exceptions import RecordsWithoutSchemaException
+
 from .templates import TargetFileTestTemplate, TargetTestTemplate
 
 
@@ -97,9 +99,7 @@ class TargetRecordBeforeSchemaTest(TargetFileTestTemplate):
 
     def test(self) -> None:
         """Run test."""
-        # TODO: the SDK should raise a better error than KeyError in this case
-        # https://github.com/meltano/sdk/issues/1755
-        with pytest.raises(KeyError):
+        with pytest.raises(RecordsWithoutSchemaException):
             super().test()