[FLINK-7430] ContinuousFileReaderOperator swallows exceptions

• Type: Bug
• Status: Resolved
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 1.3.2, 1.4.0
• Fix Version/s: 1.4.0
• Component/s: API / DataStream, Connectors / FileSystem

Environment

• macOS 10.12.6
• Oracle JDK 1.8.0_144
• Flink 1.3.2

• Assignee: Till Rohrmann
• Reporter: Peter Ertl
• Created: 11/Aug/17 11:04
• Updated: 06/Sep/17 18:51
• Resolved: 06/Sep/17 14:12

Description

The class ContinuousFileReaderOperator is swallowing exceptions as the following example demonstrates:

package org.apache.flink.streaming.examples;

import java.io.File;
import java.io.IOException;

import org.apache.flink.api.common.io.OutputFormat;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class FormatExceptionSwallowed {
	public static void main(String[] args) throws Exception {
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		File bla = File.createTempFile("foo", "baz");
		try(PrintWriter w = new PrintWriter(bla)) {
			w.println("one");
			w.println("two");
			w.println("three");
		}
		env.readTextFile(bla.getCanonicalPath())
			.writeUsingOutputFormat(new OutputFormat<String>() {
				@Override
				public void configure(final Configuration parameters) {
				}
				@Override
				public void open(final int taskNumber, final int numTasks) throws IOException {
				}
				@Override
				public void writeRecord(final String record) throws IOException {
					throw new IllegalArgumentException("bla");
				}
				@Override
				public void close() throws IOException {
				}
			});
		env.execute("go");
		
		// JOB TERMINATES WITH NO EXCEPTION / ERROR whatsoever ... 
	}
}

Issue Links

• is related to: FLINK-6833 Race condition: Asynchronous checkpointing task can fail completed StreamTask
• links to: GitHub Pull Request #4650