@Test
public void testUpdateMutationWithDifferentPK() throws Exception {

  Serializable[] old = new Serializable[4];
  old[0] = 12131L;
  old[1] = "test_user";
  old[2] = 25;
  old[3] = 0;
  Serializable[] current = new Serializable[4];
  current[0] = 12334L;
  current[1] = old[1];
  current[2] = 26;
  current[3] = old[3];
  Map.Entry<Serializable[], Serializable[]> change = new AbstractMap.SimpleEntry<>(old, current);
  BinlogEvent event =
      new UpdateEvent(TABLE_ID, SERVER_ID, TIMESTAMP, BINLOG_FILE_POS, ImmutableList.of(change));
  List<? extends Mutation> mutations = eventMapper.map(event);
  assertEquals(2, mutations.size());
  assertTrue(mutations.get(0) instanceof MysqlDeleteMutation);
  MysqlDeleteMutation deleteMutation = (MysqlDeleteMutation) mutations.get(0);
  validateMetadata(deleteMutation, 0);
  Row row = deleteMutation.getRow();
  assertEquals(12131L, row.getColumns().get("id").getValue());
  assertEquals("test_user", row.getColumns().get("name").getValue());
  assertEquals(25, row.getColumns().get("age").getValue());
  assertEquals(0, row.getColumns().get("sex").getValue());
  assertTrue(mutations.get(1) instanceof MysqlInsertMutation);
  MysqlInsertMutation insertMutation = (MysqlInsertMutation) mutations.get(1);
  validateMetadata(insertMutation, 0);
  row = insertMutation.getRow();
  assertEquals(12334L, row.getColumns().get("id").getValue());
  assertEquals("test_user", row.getColumns().get("name").getValue());
  assertEquals(26, row.getColumns().get("age").getValue());
  assertEquals(0, row.getColumns().get("sex").getValue());
}