diff --git a/sources/pg_replication/helpers.py b/sources/pg_replication/helpers.py
index 0c300ed00..690c37f80 100644
--- a/sources/pg_replication/helpers.py
+++ b/sources/pg_replication/helpers.py
@@ -705,6 +705,8 @@ def process_msg(self, msg: ReplicationMessage) -> None:
                 "The truncate operation is currently not supported. "
                 "Truncate replication messages are ignored."
             )
+        elif op == 89:  # ASCII for 'Y'
+            pass  # Type messages describe custom types and do not carry row changes.
         else:
             raise ValueError(f"Unknown replication op {op}")
 
diff --git a/tests/pg_replication/test_pg_replication.py b/tests/pg_replication/test_pg_replication.py
index 0f419bd96..14da73c75 100644
--- a/tests/pg_replication/test_pg_replication.py
+++ b/tests/pg_replication/test_pg_replication.py
@@ -1,5 +1,6 @@
 import pytest
 
+from types import SimpleNamespace
 from typing import Set, Tuple
 from copy import deepcopy
 from psycopg2.errors import InsufficientPrivilege
@@ -15,6 +16,7 @@
 )
 from sources.pg_replication import replication_resource
 from sources.pg_replication.helpers import (
+    MessageConsumer,
     init_replication,
     get_pg_version,
 )
@@ -24,6 +26,15 @@
 from .utils import add_pk, assert_loaded_data, is_super_user
 
 
+def test_message_consumer_ignores_type_message() -> None:
+    consumer = MessageConsumer(upto_lsn=0, pub_ops={})
+
+    consumer.process_msg(SimpleNamespace(payload=b"Y"))  # type: ignore[arg-type]
+
+    assert consumer.data_items == {}
+    assert consumer.consumed_all is False
+
+
 @pytest.mark.parametrize("destination_name", ALL_DESTINATIONS)
 def test_schema_replication(
     src_config: Tuple[dlt.Pipeline, str, str], destination_name: str