public <OutputT extends @NonNull Object> CsvIOParse<T> withCustomRecordParsing(
      String fieldName, SerializableFunction<String, OutputT> customRecordParsingFn) {

    Map<String, SerializableFunction<String, Object>> customProcessingMap =
        getConfigBuilder().getOrCreateCustomProcessingMap();

    customProcessingMap.put(fieldName, customRecordParsingFn::apply);
    getConfigBuilder().setCustomProcessingMap(customProcessingMap);
    return this;
  }
@Test
  public void givenCustomParsingError_emits() {
    PCollection<String> records =
        csvRecords(pipeline, "instant,instantList", "2024-01-23T10:00:05.000Z,BAD CELL");
    CsvIOParse<TimeContaining> underTest =
        underTest(
                TIME_CONTAINING_SCHEMA,
                CSVFormat.DEFAULT
                    .withHeader("instant", "instantList")
                    .withAllowDuplicateHeaderNames(false),
                new HashMap<>(),
                timeContainingFromRowFn(),
                TIME_CONTAINING_CODER)
            .withCustomRecordParsing("instantList", instantListParsingLambda());

    CsvIOParseResult<TimeContaining> result = records.apply(underTest);
    PAssert.that(result.getOutput()).empty();
    PAssert.thatSingleton(result.getErrors().apply(Count.globally())).isEqualTo(1L);

    pipeline.run();
  }