@Test public void testAbortIncompleteBatches() throws Exception { long lingerMs = Long.MAX_VALUE; int numRecords = 100; final AtomicInteger numExceptionReceivedInCallback = new AtomicInteger(0); final RecordAccumulator accum = new RecordAccumulator(128 + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 64 * 1024, CompressionType.NONE, lingerMs, 100L, metrics, time, new ApiVersions(), null); class TestCallback implements Callback { @Override public void onCompletion(RecordMetadata metadata, Exception exception) { assertTrue(exception.getMessage().equals("Producer is closed forcefully.")); numExceptionReceivedInCallback.incrementAndGet(); } } for (int i = 0; i < numRecords; i++) accum.append(new TopicPartition(topic, i % 3), 0L, key, value, null, new TestCallback(), maxBlockTimeMs); RecordAccumulator.ReadyCheckResult result = accum.ready(cluster, time.milliseconds()); assertFalse(result.readyNodes.isEmpty()); Map<Integer, List<ProducerBatch>> drained = accum.drain(cluster, result.readyNodes, Integer.MAX_VALUE, time.milliseconds()); assertTrue(accum.hasUndrained()); assertTrue(accum.hasIncomplete()); int numDrainedRecords = 0; for (Map.Entry<Integer, List<ProducerBatch>> drainedEntry : drained.entrySet()) { for (ProducerBatch batch : drainedEntry.getValue()) { assertTrue(batch.isClosed()); assertFalse(batch.produceFuture.completed()); numDrainedRecords += batch.recordCount; } } assertTrue(numDrainedRecords > 0 && numDrainedRecords < numRecords); accum.abortIncompleteBatches(); assertEquals(numRecords, numExceptionReceivedInCallback.get()); assertFalse(accum.hasUndrained()); assertFalse(accum.hasIncomplete()); }
public void abortIncompleteBatches() { do { abortBatches(); } while (appendsInProgress()); abortBatches(); this.batches.clear(); }
RecordAccumulator { public void abortIncompleteBatches() { do { abortBatches(); } while (appendsInProgress()); abortBatches(); this.batches.clear(); } }
RecordAccumulator { public void abortIncompleteBatches() { do { abortBatches(); } while (appendsInProgress()); abortBatches(); this.batches.clear(); } RecordAccumulator(int batchSize,
                             long totalSize,
                             CompressionType compression,
                             long lingerMs,
                             long retryBackoffMs,
                             Metrics metrics,
                             Time time,
                             ApiVersions apiVersions,
                             TransactionManager transactionManager); }
RecordAccumulator { public void abortIncompleteBatches() { do { abortBatches(); } while (appendsInProgress()); abortBatches(); this.batches.clear(); } 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 void abortIncompleteBatches() { do { abortBatches(); } while (appendsInProgress()); abortBatches(); this.batches.clear(); } 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();  }