afedulov commented on a change in pull request #17598:
URL: https://github.com/apache/flink/pull/17598#discussion_r792676507
##########
File path:
flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvReaderFormat.java
##########
@@ -155,16 +169,34 @@
private static final class Reader<R, T> implements StreamFormat.Reader<T> {
private final MappingIterator<R> iterator;
private final Converter<R, T, Void> converter;
+ private final boolean ignoreParseErrors;
- public Reader(MappingIterator<R> iterator, Converter<R, T, Void>
converter) {
+ public Reader(
+ MappingIterator<R> iterator,
+ Converter<R, T, Void> converter,
+ boolean ignoreParseErrors) {
this.iterator = checkNotNull(iterator);
this.converter = checkNotNull(converter);
+ this.ignoreParseErrors = ignoreParseErrors;
}
@Nullable
@Override
public T read() throws IOException {
- return iterator.hasNext() ? converter.convert(iterator.next(),
null) : null;
+ while (true) {
+ try {
+ if (iterator.hasNext()) {
+ R nextElement = iterator.next();
+ return converter.convert(nextElement, null);
+ } else {
+ return null;
+ }
+ } catch (Throwable t) {
Review comment:
I decided not to reinvent the wheel and handled it exactly as this
option is currently handled in the existing implementation to avoid surprises
for the users that already rely on this functionality:
https://github.com/apache/flink/blob/16452fbb57c11418796912251152947601572106/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvRowDataDeserializationSchema.java#L157
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
