From d906096820c4463c05d953a7830e7b4b40c58a49 Mon Sep 17 00:00:00 2001
From: Chris Cranford <chris@hibernate.org>
Date: Mon, 30 Sep 2019 13:39:03 -0400
Subject: [PATCH] DBZ-1507 Ensure message keys have correct field order

---
 .../main/java/io/debezium/relational/Key.java |  7 +----
 .../java/io/debezium/relational/Table.java    |  2 +-
 .../relational/TableSchemaBuilderTest.java    | 27 +++++++++++++++++++
 3 files changed, 29 insertions(+), 7 deletions(-)

diff --git a/debezium-core/src/main/java/io/debezium/relational/Key.java b/debezium-core/src/main/java/io/debezium/relational/Key.java
index af2aa6df4..c92b57ad5 100644
--- a/debezium-core/src/main/java/io/debezium/relational/Key.java
+++ b/debezium-core/src/main/java/io/debezium/relational/Key.java
@@ -80,12 +80,7 @@ public static interface KeyMapper {
     private static class IdentityKeyMapper {
 
         public static KeyMapper getInstance() {
-            return (table) -> {
-                return table.columns()
-                        .stream()
-                        .filter(c -> table.isPrimaryKeyColumn(c.name()))
-                        .collect(Collectors.toList());
-            };
+            return (table) -> table.primaryKeyColumns();
         }
     }
 
diff --git a/debezium-core/src/main/java/io/debezium/relational/Table.java b/debezium-core/src/main/java/io/debezium/relational/Table.java
index 56ebc6168..4624f60c5 100644
--- a/debezium-core/src/main/java/io/debezium/relational/Table.java
+++ b/debezium-core/src/main/java/io/debezium/relational/Table.java
@@ -43,7 +43,7 @@ public static TableEditor editor() {
      * @return the list of columns that make up the primary key; never null but possibly empty
      */
     default List<Column> primaryKeyColumns() {
-        return filterColumns(col->isPrimaryKeyColumn(col.name()));
+        return primaryKeyColumnNames().stream().map(this::columnWithName).collect(Collectors.toList());
     }
 
     /**
diff --git a/debezium-core/src/test/java/io/debezium/relational/TableSchemaBuilderTest.java b/debezium-core/src/test/java/io/debezium/relational/TableSchemaBuilderTest.java
index 78fa8dbb8..2d3add7cc 100644
--- a/debezium-core/src/test/java/io/debezium/relational/TableSchemaBuilderTest.java
+++ b/debezium-core/src/test/java/io/debezium/relational/TableSchemaBuilderTest.java
@@ -395,4 +395,31 @@ public void customKeyMapperShouldMapMultipleTables() {
         assertThat(key2.fields()).hasSize(1);
         assertThat(key2.field("C1").name()).isEqualTo("C1");
     }
+
+    @Test
+    @FixFor("DBZ-1507")
+    public void defaultKeyMapperShouldOrderKeyColumnsBasedOnPrimaryKeyColumnNamesOrder() {
+        TableId id2 = new TableId("catalog", "schema", "info");
+        Table table2 = Table.editor()
+                .tableId(id2)
+                .addColumns(Column.editor().name("t1ID")
+                                    .type("INT").jdbcType(Types.INTEGER)
+                                    .optional(false)
+                                    .create(),
+                            Column.editor().name("t2ID")
+                                    .type("INT").jdbcType(Types.INTEGER)
+                                    .optional(false)
+                                    .create())
+                .setPrimaryKeyNames("t2ID", "t1ID")
+                .create();
+
+        TableSchema schema2 = new TableSchemaBuilder(new JdbcValueConverters(), adjuster, SchemaBuilder.struct().build(), false)
+                .create(prefix, "sometopic", table2, null, null, null);
+
+        Schema key2 = schema2.keySchema();
+        assertThat(key2).isNotNull();
+        assertThat(key2.fields()).hasSize(2);
+        assertThat(key2.fields().get(0).name()).isEqualTo("t2ID");
+        assertThat(key2.fields().get(1).name()).isEqualTo("t1ID");
+    }
 }