kafka git commit: KAFKA-4725; Stop leaking messages in produce request body when requests are delayed

[ju...@apache.org]

Repository: kafka
Updated Branches:
  refs/heads/trunk 6cef4f5e4 -> 7f6bf5d44


KAFKA-4725; Stop leaking messages in produce request body when requests are delayed

This change is in response to [KAFKA-4725](https://issues.apache.org/jira/browse/KAFKA-4725).

When a produce request is received, if the user/client is exceeding their produce quota, the response will be delayed until the quota is refilled appropriately.

Unfortunately, the request body is still referenced in the callback which in turn leaks the messages contained within the request.

This change allows the `KafkaApis` method to take ownership of the request body from the `RequestChannel.Request` object.

I am not sure whether this breaks other invariants which are assumed within other parts of Kafka.

Author: Tim Carey-Smith <ti...@spork.in>

Reviewers: Ismael Juma <is...@juma.me.uk>, Jun Rao <ju...@gmail.com>

Closes #2496 from halorgium/fix-throttled-response-leak


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/7f6bf5d4
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/7f6bf5d4
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/7f6bf5d4

Branch: refs/heads/trunk
Commit: 7f6bf5d44796ac14665d5ba8c2243f5d7f5d30bd
Parents: 6cef4f5
Author: Tim Carey-Smith <ti...@spork.in>
Authored: Tue Feb 7 15:12:47 2017 -0800
Committer: Jun Rao <ju...@gmail.com>
Committed: Tue Feb 7 15:12:47 2017 -0800

----------------------------------------------------------------------
 .../main/java/org/apache/kafka/common/requests/ProduceRequest.java  | 1 +
 1 file changed, 1 insertion(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/7f6bf5d4/clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java
index e4a2aae..bd3ae8f 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java
@@ -160,6 +160,7 @@ public class ProduceRequest extends AbstractRequest {
     }
 
     public void clearPartitionRecords() {
+        struct.clear();
         partitionRecords.clear();
     }