diff --git a/singer_sdk/sinks/core.py b/singer_sdk/sinks/core.py
index ec43c4060..9928aa6f2 100644
--- a/singer_sdk/sinks/core.py
+++ b/singer_sdk/sinks/core.py
@@ -215,6 +215,9 @@ def datetime_error_treatment(self) -> DatetimeErrorTreatmentEnum:
     def key_properties(self) -> list[str]:
         """Return key properties.
 
+        Override this method to return a list of key properties in a format that is
+        compatible with the target.
+
         Returns:
             A list of stream key properties.
         """
@@ -331,10 +334,10 @@ def _singer_validate_message(self, record: dict) -> None:
         Raises:
             MissingKeyPropertiesError: If record is missing one or more key properties.
         """
-        if not all(key_property in record for key_property in self.key_properties):
+        if any(key_property not in record for key_property in self._key_properties):
             msg = (
                 f"Record is missing one or more key_properties. \n"
-                f"Key Properties: {self.key_properties}, "
+                f"Key Properties: {self._key_properties}, "
                 f"Record Keys: {list(record.keys())}"
             )
             raise MissingKeyPropertiesError(
diff --git a/tests/conftest.py b/tests/conftest.py
index 142e76fe1..cb201c9a1 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -86,6 +86,10 @@ def process_batch(self, context: dict) -> None:
         self.target.records_written.extend(context["records"])
         self.target.num_batches_processed += 1
 
+    @property
+    def key_properties(self) -> list[str]:
+        return [key.upper() for key in super().key_properties]
+
 
 class TargetMock(Target):
     """A mock Target class."""
diff --git a/tests/core/test_target_base.py b/tests/core/test_target_base.py
index 778fab722..1fd6b9a93 100644
--- a/tests/core/test_target_base.py
+++ b/tests/core/test_target_base.py
@@ -2,6 +2,9 @@
 
 import copy
 
+import pytest
+
+from singer_sdk.exceptions import MissingKeyPropertiesError
 from tests.conftest import BatchSinkMock, TargetMock
 
 
@@ -28,3 +31,25 @@ def test_get_sink():
         key_properties=key_properties,
     )
     assert sink_returned == sink
+
+
+def test_validate_record():
+    target = TargetMock()
+    sink = BatchSinkMock(
+        target=target,
+        stream_name="test",
+        schema={
+            "properties": {
+                "id": {"type": ["integer"]},
+                "name": {"type": ["string"]},
+            },
+        },
+        key_properties=["id"],
+    )
+
+    # Test valid record
+    sink._singer_validate_message({"id": 1, "name": "test"})
+
+    # Test invalid record
+    with pytest.raises(MissingKeyPropertiesError):
+        sink._singer_validate_message({"name": "test"})