@Test public void testExpiredBatches() throws InterruptedException { long retryBackoffMs = 100L; long lingerMs = 3000L; int requestTimeout = 60; int batchSize = 1025; RecordAccumulator accum = new RecordAccumulator(batchSize + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 10 * batchSize, CompressionType.NONE, lingerMs, retryBackoffMs, metrics, time, new ApiVersions(), null); int appends = expectedNumAppends(batchSize); for (int i = 0; i < appends; i++) { accum.append(tp1, 0L, key, value, Record.EMPTY_HEADERS, null, maxBlockTimeMs); assertEquals("No partitions should be ready.", 0, accum.ready(cluster, time.milliseconds()).readyNodes.size()); } accum.append(tp1, 0L, key, value, Record.EMPTY_HEADERS, null, 0); Set<Node> readyNodes = accum.ready(cluster, time.milliseconds()).readyNodes; assertEquals("Our partition's leader should be ready", Collections.singleton(node1), readyNodes); time.sleep(requestTimeout + 1); accum.mutePartition(tp1); List<ProducerBatch> expiredBatches = accum.expiredBatches(requestTimeout, time.milliseconds()); assertEquals("The batch should not be expired when the partition is muted", 0, expiredBatches.size()); accum.unmutePartition(tp1); expiredBatches = accum.expiredBatches(requestTimeout, time.milliseconds()); assertEquals("The batch should be expired", 1, expiredBatches.size()); assertEquals("No partitions should be ready.", 0, accum.ready(cluster, time.milliseconds()).readyNodes.size()); time.sleep(lingerMs); assertEquals("Our partition's leader should be ready", Collections.singleton(node1), readyNodes); time.sleep(requestTimeout + 1); accum.mutePartition(tp1); expiredBatches = accum.expiredBatches(requestTimeout, time.milliseconds()); assertEquals("The batch should not be expired when metadata is still available and partition is muted", 0, expiredBatches.size()); accum.unmutePartition(tp1); expiredBatches = accum.expiredBatches(requestTimeout, time.milliseconds()); assertEquals("The batch should be expired when the partition is not muted", 1, expiredBatches.size()); assertEquals("No partitions should be ready.", 0, accum.ready(cluster, time.milliseconds()).readyNodes.size()); accum.append(tp1, 0L, key, value, Record.EMPTY_HEADERS, null, 0); time.sleep(lingerMs); readyNodes = accum.ready(cluster, time.milliseconds()).readyNodes; assertEquals("Our partition's leader should be ready", Collections.singleton(node1), readyNodes); Map<Integer, List<ProducerBatch>> drained = accum.drain(cluster, readyNodes, Integer.MAX_VALUE, time.milliseconds()); assertEquals("There should be only one batch.", drained.get(node1.id()).size(), 1); time.sleep(1000L); accum.reenqueue(drained.get(node1.id()).get(0), time.milliseconds()); time.sleep(requestTimeout + retryBackoffMs); expiredBatches = accum.expiredBatches(requestTimeout, time.milliseconds()); assertEquals("The batch should not be expired.", 0, expiredBatches.size()); time.sleep(1L); accum.mutePartition(tp1); expiredBatches = accum.expiredBatches(requestTimeout, time.milliseconds()); assertEquals("The batch should not be expired when the partition is muted", 0, expiredBatches.size()); accum.unmutePartition(tp1); expiredBatches = accum.expiredBatches(requestTimeout, time.milliseconds()); assertEquals("The batch should be expired when the partition is not muted.", 1, expiredBatches.size()); }
public List<ProducerBatch> expiredBatches(int requestTimeout, long now) { List<ProducerBatch> expiredBatches = new ArrayList<>(); for (Map.Entry<TopicPartition, Deque<ProducerBatch>> entry : this.batches.entrySet()) { Deque<ProducerBatch> dq = entry.getValue(); TopicPartition tp = entry.getKey(); if (!muted.contains(tp)) { synchronized (dq) { ProducerBatch lastBatch = dq.peekLast(); Iterator<ProducerBatch> batchIterator = dq.iterator(); while (batchIterator.hasNext()) { ProducerBatch batch = batchIterator.next(); boolean isFull = batch != lastBatch || batch.isFull(); if (batch.maybeExpire(requestTimeout, retryBackoffMs, now, this.lingerMs, isFull)) { expiredBatches.add(batch); batchIterator.remove(); } else { break; } } } } } return expiredBatches; }
RecordAccumulator { public List<ProducerBatch> expiredBatches(int requestTimeout, long now) { List<ProducerBatch> expiredBatches = new ArrayList<>(); for (Map.Entry<TopicPartition, Deque<ProducerBatch>> entry : this.batches.entrySet()) { Deque<ProducerBatch> dq = entry.getValue(); TopicPartition tp = entry.getKey(); if (!muted.contains(tp)) { synchronized (dq) { ProducerBatch lastBatch = dq.peekLast(); Iterator<ProducerBatch> batchIterator = dq.iterator(); while (batchIterator.hasNext()) { ProducerBatch batch = batchIterator.next(); boolean isFull = batch != lastBatch || batch.isFull(); if (batch.maybeExpire(requestTimeout, retryBackoffMs, now, this.lingerMs, isFull)) { expiredBatches.add(batch); batchIterator.remove(); } else { break; } } } } } return expiredBatches; } }
RecordAccumulator { public List<ProducerBatch> expiredBatches(int requestTimeout, long now) { List<ProducerBatch> expiredBatches = new ArrayList<>(); for (Map.Entry<TopicPartition, Deque<ProducerBatch>> entry : this.batches.entrySet()) { Deque<ProducerBatch> dq = entry.getValue(); TopicPartition tp = entry.getKey(); if (!muted.contains(tp)) { synchronized (dq) { ProducerBatch lastBatch = dq.peekLast(); Iterator<ProducerBatch> batchIterator = dq.iterator(); while (batchIterator.hasNext()) { ProducerBatch batch = batchIterator.next(); boolean isFull = batch != lastBatch || batch.isFull(); if (batch.maybeExpire(requestTimeout, retryBackoffMs, now, this.lingerMs, isFull)) { expiredBatches.add(batch); batchIterator.remove(); } else { break; } } } } } return expiredBatches; } RecordAccumulator(int batchSize,
                             long totalSize,
                             CompressionType compression,
                             long lingerMs,
                             long retryBackoffMs,
                             Metrics metrics,
                             Time time,
                             ApiVersions apiVersions,
                             TransactionManager transactionManager); }
RecordAccumulator { public List<ProducerBatch> expiredBatches(int requestTimeout, long now) { List<ProducerBatch> expiredBatches = new ArrayList<>(); for (Map.Entry<TopicPartition, Deque<ProducerBatch>> entry : this.batches.entrySet()) { Deque<ProducerBatch> dq = entry.getValue(); TopicPartition tp = entry.getKey(); if (!muted.contains(tp)) { synchronized (dq) { ProducerBatch lastBatch = dq.peekLast(); Iterator<ProducerBatch> batchIterator = dq.iterator(); while (batchIterator.hasNext()) { ProducerBatch batch = batchIterator.next(); boolean isFull = batch != lastBatch || batch.isFull(); if (batch.maybeExpire(requestTimeout, retryBackoffMs, now, this.lingerMs, isFull)) { expiredBatches.add(batch); batchIterator.remove(); } else { break; } } } } } return expiredBatches; } RecordAccumulator(int batchSize,
                             long totalSize,
                             CompressionType compression,
                             long lingerMs,
                             long retryBackoffMs,
                             Metrics metrics,
                             Time time,
                             ApiVersions apiVersions,
                             TransactionManager transactionManager); RecordAppendResult append(TopicPartition tp,
                                     long timestamp,
                                     byte[] key,
                                     byte[] value,
                                     Header[] headers,
                                     Callback callback,
                                     long maxTimeToBlock); List<ProducerBatch> expiredBatches(int requestTimeout, long now); void reenqueue(ProducerBatch batch, long now); int splitAndReenqueue(ProducerBatch bigBatch); ReadyCheckResult ready(Cluster cluster, long nowMs); boolean hasUndrained(); Map<Integer, List<ProducerBatch>> drain(Cluster cluster,
                                                   Set<Node> nodes,
                                                   int maxSize,
                                                   long now); void deallocate(ProducerBatch batch); void beginFlush(); void awaitFlushCompletion(); boolean hasIncomplete(); void abortIncompleteBatches(); void mutePartition(TopicPartition tp); void unmutePartition(TopicPartition tp); void close(); }
RecordAccumulator { public List<ProducerBatch> expiredBatches(int requestTimeout, long now) { List<ProducerBatch> expiredBatches = new ArrayList<>(); for (Map.Entry<TopicPartition, Deque<ProducerBatch>> entry : this.batches.entrySet()) { Deque<ProducerBatch> dq = entry.getValue(); TopicPartition tp = entry.getKey(); if (!muted.contains(tp)) { synchronized (dq) { ProducerBatch lastBatch = dq.peekLast(); Iterator<ProducerBatch> batchIterator = dq.iterator(); while (batchIterator.hasNext()) { ProducerBatch batch = batchIterator.next(); boolean isFull = batch != lastBatch || batch.isFull(); if (batch.maybeExpire(requestTimeout, retryBackoffMs, now, this.lingerMs, isFull)) { expiredBatches.add(batch); batchIterator.remove(); } else { break; } } } } } return expiredBatches; } RecordAccumulator(int batchSize,
                             long totalSize,
                             CompressionType compression,
                             long lingerMs,
                             long retryBackoffMs,
                             Metrics metrics,
                             Time time,
                             ApiVersions apiVersions,
                             TransactionManager transactionManager); RecordAppendResult append(TopicPartition tp,
                                     long timestamp,
                                     byte[] key,
                                     byte[] value,
                                     Header[] headers,
                                     Callback callback,
                                     long maxTimeToBlock); List<ProducerBatch> expiredBatches(int requestTimeout, long now); void reenqueue(ProducerBatch batch, long now); int splitAndReenqueue(ProducerBatch bigBatch); ReadyCheckResult ready(Cluster cluster, long nowMs); boolean hasUndrained(); Map<Integer, List<ProducerBatch>> drain(Cluster cluster,
                                                   Set<Node> nodes,
                                                   int maxSize,
                                                   long now); void deallocate(ProducerBatch batch); void beginFlush(); void awaitFlushCompletion(); boolean hasIncomplete(); void abortIncompleteBatches(); void mutePartition(TopicPartition tp); void unmutePartition(TopicPartition tp); void close();  }