diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumJsonRecordParserTest.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumJsonRecordParserTest.java
index 747c56d5b55a..4cb64c11f6b6 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumJsonRecordParserTest.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumJsonRecordParserTest.java
@@ -69,6 +69,32 @@ public void testEmptyPrimaryKeysFallBackToKey() throws Exception {
         assertPrimaryKeys(key, value("[]"), "id");
     }
 
+    @Test
+    public void testPrimaryKeysFromValueWithSchemaEnvelope() throws Exception {
+        // Simulates the exact scenario from issue #8961:
+        // Value has schema+payload envelope (standard Debezium JSON),
+        // and primary keys come from the Kafka message key.
+        JsonNode key =
+                OBJECT_MAPPER.readTree(
+                        "{\"schema\":{\"type\":\"struct\",\"fields\":["
+                                + "{\"type\":\"int64\",\"optional\":false,\"field\":\"id\"}]},"
+                                + "\"payload\":{\"id\":1}}");
+
+        JsonNode value =
+                OBJECT_MAPPER.readTree(
+                        "{\"schema\":{\"type\":\"struct\",\"fields\":["
+                                + "{\"type\":\"struct\",\"fields\":["
+                                + "{\"type\":\"int64\",\"optional\":false,\"field\":\"id\"},"
+                                + "{\"type\":\"string\",\"optional\":true,\"field\":\"name\"}],"
+                                + "\"optional\":true,\"field\":\"after\"}],"
+                                + "\"optional\":false},\"payload\":{\"before\":null,"
+                                + "\"after\":{\"id\":1,\"name\":\"Alice\"},"
+                                + "\"source\":{\"db\":\"test\",\"table\":\"users\"},"
+                                + "\"op\":\"c\"}}");
+
+        assertPrimaryKeys(key, value, "id");
+    }
+
     private static JsonNode value(String primaryKeys) throws Exception {
         String primaryKeyField = primaryKeys == null ? "" : "\"pkNames\":" + primaryKeys + ",";
         return OBJECT_MAPPER.readTree(