You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Peter Ertl (JIRA)" <ji...@apache.org> on 2017/08/11 11:07:00 UTC
[jira] [Updated] (FLINK-7430) ContinuousFileReaderOperator swallows exceptions

     [ https://issues.apache.org/jira/browse/FLINK-7430?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Peter Ertl updated FLINK-7430:
------------------------------
    Description: 
The class ContinuousFileReaderOperator is swallowing exception as the following example demonstrates:

{code:java}
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");
		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 ... 
	}
}

{code}


  was:
The class ContinuousFileReaderOperator is swallowing exception as the following example demonstrated:

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");
		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 ... 
	}
}



> ContinuousFileReaderOperator swallows exceptions
> ------------------------------------------------
>
>                 Key: FLINK-7430
>                 URL: https://issues.apache.org/jira/browse/FLINK-7430
>             Project: Flink
>          Issue Type: Bug
>          Components: DataStream API
>    Affects Versions: 1.3.2
>            Reporter: Peter Ertl
>
> The class ContinuousFileReaderOperator is swallowing exception as the following example demonstrates:
> {code:java}
> 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");
> 		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 ... 
> 	}
> }
> {code}



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)