From d3aa8126a86eb1a603e57bdbcbc63648bfb86ead Mon Sep 17 00:00:00 2001
From: Gwen Shapira <cshapi@gmail.com>
Date: Tue, 7 Jul 2015 19:45:53 -0700
Subject: [PATCH] prevent double closing of record batches in producer

---
 .../main/java/org/apache/kafka/common/record/MemoryRecords.java   | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java
index b2db240..5f1b45c 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java
@@ -115,9 +115,11 @@ public class MemoryRecords implements Records {
      * Close this batch for no more appends
      */
     public void close() {
-        compressor.close();
-        writable = false;
-        buffer = compressor.buffer();
+        if (writable) {
+            compressor.close();
+            writable = false;
+            buffer = compressor.buffer();
+        }
     }
 
     /** Write the records in this set to the given channel */
-- 
2.3.2 (Apple Git-55)