public static TableSchema toSchema(RowType rowType) {
    TableSchema.Builder builder = TableSchema.builder();
    for (RowType.RowField field : rowType.getFields()) {
      builder.field(field.getName(), TypeConversions.fromLogicalToDataType(field.getType()));
    }
    return builder.build();
  }
@Test
  public void testConvertFlinkSchemaWithPrimaryKeys() {
    Schema icebergSchema =
        new Schema(
            Lists.newArrayList(
                Types.NestedField.required(1, "int", Types.IntegerType.get()),
                Types.NestedField.required(2, "string", Types.StringType.get())),
            Sets.newHashSet(1, 2));

    TableSchema tableSchema = FlinkSchemaUtil.toSchema(icebergSchema);
    assertThat(tableSchema.getPrimaryKey())
        .isPresent()
        .get()
        .satisfies(k -> assertThat(k.getColumns()).containsExactly("int", "string"));
  }