[streaming] Proper exception propagation for udf exception + collector bugfix

flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamConfig.java

@@ -187,7 +187,7 @@ public class StreamConfig implements Serializable {
 
 	public void setOutputSelectors(List<OutputSelector<?>> outputSelector) {
 		try {
-			if (outputSelector != null) {
+			if (outputSelector != null && !outputSelector.isEmpty()) {
 				setDirectedEmit(true);
 				config.setBytes(OUTPUT_SELECTOR,
 						SerializationUtils.serialize((Serializable) outputSelector));

flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/StreamInvokable.java

@@ -128,6 +128,7 @@ public abstract class StreamInvokable<IN, OUT> implements Serializable {
 				LOG.error("Calling user function failed due to: {}",
 						StringUtils.stringifyException(e));
 			}
+			throw new RuntimeException(e);
 		}
 	}