static Map<Integer, Schema.Field> mapFieldPositions(CSVFormat format, Schema schema) {
    List<String> header = Arrays.asList(format.getHeader());
    Map<Integer, Schema.Field> indexToFieldMap = new HashMap<>();
    for (Schema.Field field : schema.getFields()) {
      int index = getIndex(header, field);
      if (index >= 0) {
        indexToFieldMap.put(index, field);
      }
    }
    return indexToFieldMap;
  }
@Test
  public void givenNonNullableHeaderAndSchemaFieldMismatch_throws() {
    Schema schema =
        Schema.builder()
            .addStringField("another_string")
            .addInt32Field("an_integer")
            .addStringField("a_string")
            .build();
    IllegalArgumentException e =
        assertThrows(
            IllegalArgumentException.class,
            () ->
                CsvIOParseHelpers.mapFieldPositions(
                    csvFormat().withHeader("an_integer", "a_string"), schema));
    assertEquals(
        "header does not contain required class org.apache.beam.sdk.schemas.Schema field: "
            + schema.getField("another_string").getName(),
        e.getMessage());
  }