From fb299c0f93a01c6f7a81b0140d05d92ddf82d167 Mon Sep 17 00:00:00 2001 From: Vaibhav kulkarni Date: Sat, 25 Jun 2016 11:22:26 +0530 Subject: [PATCH] closing InputStream after execution completes --- .../src/main/java/org/apache/nifi/spark/NiFiReceiver.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/nifi-external/nifi-spark-receiver/src/main/java/org/apache/nifi/spark/NiFiReceiver.java b/nifi-external/nifi-spark-receiver/src/main/java/org/apache/nifi/spark/NiFiReceiver.java index 83a7e42ed7a6..c4e419120a8e 100644 --- a/nifi-external/nifi-spark-receiver/src/main/java/org/apache/nifi/spark/NiFiReceiver.java +++ b/nifi-external/nifi-spark-receiver/src/main/java/org/apache/nifi/spark/NiFiReceiver.java @@ -162,10 +162,11 @@ public void run() { } final List dataPackets = new ArrayList<>(); + InputStream inStream =null; do { // Read the data into a byte array and wrap it along with the attributes // into a NiFiDataPacket. - final InputStream inStream = dataPacket.getData(); + inStream = dataPacket.getData(); final byte[] data = new byte[(int) dataPacket.getSize()]; StreamUtils.fillBuffer(inStream, data); @@ -173,6 +174,7 @@ public void run() { final NiFiDataPacket NiFiDataPacket = new StandardNiFiDataPacket(data, attributes); dataPackets.add(NiFiDataPacket); dataPacket = transaction.receive(); + inStream.close(); } while (dataPacket != null); // Confirm transaction to verify the data