diff --git a/singer_sdk/_singerlib/catalog.py b/singer_sdk/_singerlib/catalog.py
index 986272e3c7..c1281e091f 100644
--- a/singer_sdk/_singerlib/catalog.py
+++ b/singer_sdk/_singerlib/catalog.py
@@ -87,6 +87,7 @@ class StreamMetadata(Metadata):
     """Stream metadata."""
 
     table_key_properties: t.Sequence[str] | None = None
+    replication_key: str | None = None
     forced_replication_method: str | None = None
     valid_replication_keys: list[str] | None = None
     schema_name: str | None = None
diff --git a/singer_sdk/streams/core.py b/singer_sdk/streams/core.py
index 009726b7fe..423de6994e 100644
--- a/singer_sdk/streams/core.py
+++ b/singer_sdk/streams/core.py
@@ -1277,10 +1277,21 @@ def apply_catalog(self, catalog: singer.Catalog) -> None:
 
         catalog_entry = catalog.get_stream(self.name)
         if catalog_entry:
-            self.primary_keys = catalog_entry.key_properties
-            self.replication_key = catalog_entry.replication_key
-            if catalog_entry.replication_method:
-                self.forced_replication_method = catalog_entry.replication_method
+            stream_metadata = catalog_entry.metadata.root
+
+            self.primary_keys = (
+                catalog_entry.key_properties or stream_metadata.table_key_properties
+            )
+            self.replication_key = (
+                catalog_entry.replication_key or stream_metadata.replication_key
+            )
+
+            replication_method = (
+                catalog_entry.replication_method
+                or stream_metadata.forced_replication_method
+            )
+            if replication_method:
+                self.forced_replication_method = replication_method
 
     def _get_state_partition_context(
         self,