[streaming] ack update

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/FaultTolerancyBuffer.java

@@ -16,6 +16,7 @@ public class FaultTolerancyBuffer {
 	private int numberOfOutputs;
 
 	public FaultTolerancyBuffer(List<RecordWriter<Record>> outputs) {
+		this.outputs=outputs;
 		this.recordBuffer = new HashMap<String, StreamRecord>();
 		this.ackCounter = new HashMap<String, Integer>();
 		this.numberOfOutputs = outputs.size();

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/invokable/StreamInvokable.java

@@ -39,6 +39,7 @@ public abstract class StreamInvokable {
 
 			} catch (Exception e) {
 				System.out.println("Emit error");
+				emittedRecords.failRecord(streamRecord.getId());
 			}
 		}
 	}