Repository: kafka Updated Branches: refs/heads/trunk 124f73b17 -> 1cd22ed33
KAFKA2805; RecordAccumulator request timeout not enforced when all brokers are gone Removed the check for expiring only those batches whose metadata is unavailable. Now the batches will be expired irrespective of whether the leader is available or not, as soon as it reaches the requestimeout threshold. Author: Mayuresh Gharat <[email protected]> Reviewers: Jun Rao <[email protected]> Closes #503 from MayureshGharat/kafka-2805 Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/1cd22ed3 Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/1cd22ed3 Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/1cd22ed3 Branch: refs/heads/trunk Commit: 1cd22ed33f1e1f63d8cb63f68309e5d8f43be1e1 Parents: 124f73b Author: Mayuresh Gharat <[email protected]> Authored: Wed Nov 11 18:41:45 2015 -0800 Committer: Jun Rao <[email protected]> Committed: Wed Nov 11 18:41:45 2015 -0800 ---------------------------------------------------------------------- .../producer/internals/RecordAccumulator.java | 21 +++++++++----------- 1 file changed, 9 insertions(+), 12 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/1cd22ed3/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java index db61121..d4a8a23 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java @@ -216,18 +216,15 @@ public final class RecordAccumulator { Iterator<RecordBatch> batchIterator = dq.iterator(); while (batchIterator.hasNext()) { RecordBatch batch = batchIterator.next(); - Node leader = cluster.leaderFor(topicAndPartition); - if (leader == null) { - // check if the batch is expired - if (batch.maybeExpire(requestTimeout, now, this.lingerMs)) { - expiredBatches.add(batch); - count++; - batchIterator.remove(); - deallocate(batch); - } else { - if (!batch.inRetry()) { - break; - } + // check if the batch is expired + if (batch.maybeExpire(requestTimeout, now, this.lingerMs)) { + expiredBatches.add(batch); + count++; + batchIterator.remove(); + deallocate(batch); + } else { + if (!batch.inRetry()) { + break; } } }
