test_class.declarator: topic = "test"
test_class.modifier: private
test_class.original_string: private String topic = "test";
test_class.type: String
test_class.var_name: topic
test_class.declarator: partition1 = 0
test_class.modifier: private
test_class.original_string: private int partition1 = 0;
test_class.type: int
test_class.var_name: partition1
test_class.declarator: partition2 = 1
test_class.modifier: private
test_class.original_string: private int partition2 = 1;
test_class.type: int
test_class.var_name: partition2
test_class.declarator: partition3 = 2
test_class.modifier: private
test_class.original_string: private int partition3 = 2;
test_class.type: int
test_class.var_name: partition3
test_class.declarator: node1 = new Node(0, "localhost", 1111)
test_class.modifier: private
test_class.original_string: private Node node1 = new Node(0, "localhost", 1111);
test_class.type: Node
test_class.var_name: node1
test_class.declarator: node2 = new Node(1, "localhost", 1112)
test_class.modifier: private
test_class.original_string: private Node node2 = new Node(1, "localhost", 1112);
test_class.type: Node
test_class.var_name: node2
test_class.declarator: tp1 = new TopicPartition(topic, partition1)
test_class.modifier: private
test_class.original_string: private TopicPartition tp1 = new TopicPartition(topic, partition1);
test_class.type: TopicPartition
test_class.var_name: tp1
test_class.declarator: tp2 = new TopicPartition(topic, partition2)
test_class.modifier: private
test_class.original_string: private TopicPartition tp2 = new TopicPartition(topic, partition2);
test_class.type: TopicPartition
test_class.var_name: tp2
test_class.declarator: tp3 = new TopicPartition(topic, partition3)
test_class.modifier: private
test_class.original_string: private TopicPartition tp3 = new TopicPartition(topic, partition3);
test_class.type: TopicPartition
test_class.var_name: tp3
test_class.declarator: part1 = new PartitionInfo(topic, partition1, node1, null, null)
test_class.modifier: private
test_class.original_string: private PartitionInfo part1 = new PartitionInfo(topic, partition1, node1, null, null);
test_class.type: PartitionInfo
test_class.var_name: part1
test_class.declarator: part2 = new PartitionInfo(topic, partition2, node1, null, null)
test_class.modifier: private
test_class.original_string: private PartitionInfo part2 = new PartitionInfo(topic, partition2, node1, null, null);
test_class.type: PartitionInfo
test_class.var_name: part2
test_class.declarator: part3 = new PartitionInfo(topic, partition3, node2, null, null)
test_class.modifier: private
test_class.original_string: private PartitionInfo part3 = new PartitionInfo(topic, partition3, node2, null, null);
test_class.type: PartitionInfo
test_class.var_name: part3
test_class.declarator: time = new MockTime()
test_class.modifier: private
test_class.original_string: private MockTime time = new MockTime();
test_class.type: MockTime
test_class.var_name: time
test_class.declarator: systemTime = new SystemTime()
test_class.modifier: private
test_class.original_string: private SystemTime systemTime = new SystemTime();
test_class.type: SystemTime
test_class.var_name: systemTime
test_class.declarator: key = "key".getBytes()
test_class.modifier: private
test_class.original_string: private byte[] key = "key".getBytes();
test_class.type: byte[]
test_class.var_name: key
test_class.declarator: value = "value".getBytes()
test_class.modifier: private
test_class.original_string: private byte[] value = "value".getBytes();
test_class.type: byte[]
test_class.var_name: value
test_class.declarator: msgSize = Records.LOG_OVERHEAD + Record.recordSize(key, value)
test_class.modifier: private
test_class.original_string: private int msgSize = Records.LOG_OVERHEAD + Record.recordSize(key, value);
test_class.type: int
test_class.var_name: msgSize
test_class.declarator: cluster = new Cluster(Arrays.asList(node1, node2), Arrays.asList(part1, part2, part3), Collections.<String>emptySet())
test_class.modifier: private
test_class.original_string: private Cluster cluster = new Cluster(Arrays.asList(node1, node2), Arrays.asList(part1, part2, part3), Collections.<String>emptySet());
test_class.type: Cluster
test_class.var_name: cluster
test_class.declarator: metrics = new Metrics(time)
test_class.modifier: private
test_class.original_string: private Metrics metrics = new Metrics(time);
test_class.type: Metrics
test_class.var_name: metrics
test_class.declarator: maxBlockTimeMs = 1000
test_class.modifier: private final
test_class.original_string: private final long maxBlockTimeMs = 1000;
test_class.type: long
test_class.var_name: maxBlockTimeMs
test_class.file: clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java
test_class.identifier: RecordAccumulatorTest
test_case.body: @Test
      public void testAbortIncompleteBatches() throws Exception {
          long lingerMs = Long.MAX_VALUE;
          final AtomicInteger numExceptionReceivedInCallback = new AtomicInteger(0);
          final RecordAccumulator accum = new RecordAccumulator(4 * 1024, 64 * 1024, CompressionType.NONE, lingerMs, 100L, metrics, time);
          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 < 100; i++)
              accum.append(new TopicPartition(topic, i % 3), 0L, key, value, new TestCallback(), maxBlockTimeMs);
          RecordAccumulator.ReadyCheckResult result = accum.ready(cluster, time.milliseconds());
          assertEquals("No nodes should be ready.", 0, result.readyNodes.size());
  
          accum.abortIncompleteBatches();
          assertEquals(numExceptionReceivedInCallback.get(), 100);
          assertFalse(accum.hasUnsent());
  
      }
test_case.class_method_signature: RecordAccumulatorTest.testAbortIncompleteBatches()
test_case.full_signature: @Test public void testAbortIncompleteBatches()
test_case.identifier: testAbortIncompleteBatches
test_case.invocations: assertTrue equals getMessage incrementAndGet append ready milliseconds assertEquals size abortIncompleteBatches assertEquals get assertFalse hasUnsent
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testAbortIncompleteBatches()
focal_class.declarator: log = LoggerFactory.getLogger(RecordAccumulator.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger log = LoggerFactory.getLogger(RecordAccumulator.class);
focal_class.type: Logger
focal_class.var_name: log
focal_class.declarator: closed
focal_class.modifier: private volatile
focal_class.original_string: private volatile boolean closed;
focal_class.type: boolean
focal_class.var_name: closed
focal_class.declarator: flushesInProgress
focal_class.modifier: private final
focal_class.original_string: private final AtomicInteger flushesInProgress;
focal_class.type: AtomicInteger
focal_class.var_name: flushesInProgress
focal_class.declarator: appendsInProgress
focal_class.modifier: private final
focal_class.original_string: private final AtomicInteger appendsInProgress;
focal_class.type: AtomicInteger
focal_class.var_name: appendsInProgress
focal_class.declarator: batchSize
focal_class.modifier: private final
focal_class.original_string: private final int batchSize;
focal_class.type: int
focal_class.var_name: batchSize
focal_class.declarator: compression
focal_class.modifier: private final
focal_class.original_string: private final CompressionType compression;
focal_class.type: CompressionType
focal_class.var_name: compression
focal_class.declarator: lingerMs
focal_class.modifier: private final
focal_class.original_string: private final long lingerMs;
focal_class.type: long
focal_class.var_name: lingerMs
focal_class.declarator: retryBackoffMs
focal_class.modifier: private final
focal_class.original_string: private final long retryBackoffMs;
focal_class.type: long
focal_class.var_name: retryBackoffMs
focal_class.declarator: free
focal_class.modifier: private final
focal_class.original_string: private final BufferPool free;
focal_class.type: BufferPool
focal_class.var_name: free
focal_class.declarator: time
focal_class.modifier: private final
focal_class.original_string: private final Time time;
focal_class.type: Time
focal_class.var_name: time
focal_class.declarator: batches
focal_class.modifier: private final
focal_class.original_string: private final ConcurrentMap<TopicPartition, Deque<RecordBatch>> batches;
focal_class.type: ConcurrentMap<TopicPartition, Deque<RecordBatch>>
focal_class.var_name: batches
focal_class.declarator: incomplete
focal_class.modifier: private final
focal_class.original_string: private final IncompleteRecordBatches incomplete;
focal_class.type: IncompleteRecordBatches
focal_class.var_name: incomplete
focal_class.declarator: muted
focal_class.modifier: private final
focal_class.original_string: private final Set<TopicPartition> muted;
focal_class.type: Set<TopicPartition>
focal_class.var_name: muted
focal_class.declarator: drainIndex
focal_class.modifier: private
focal_class.original_string: private int drainIndex;
focal_class.type: int
focal_class.var_name: drainIndex
focal_class.file: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
focal_class.identifier: RecordAccumulator
focal_class.class_method_signature: RecordAccumulator.RecordAccumulator(int batchSize,
          long totalSize,
          CompressionType compression,
          long lingerMs,
          long retryBackoffMs,
          Metrics metrics,
          Time time)
focal_class.full_signature: public  RecordAccumulator(int batchSize,
          long totalSize,
          CompressionType compression,
          long lingerMs,
          long retryBackoffMs,
          Metrics metrics,
          Time time)
focal_class.identifier: RecordAccumulator
focal_class.modifiers: public
focal_class.parameters: (int batchSize,
          long totalSize,
          CompressionType compression,
          long lingerMs,
          long retryBackoffMs,
          Metrics metrics,
          Time time)
focal_class.signature:  RecordAccumulator(int batchSize,
          long totalSize,
          CompressionType compression,
          long lingerMs,
          long retryBackoffMs,
          Metrics metrics,
          Time time)
focal_class.class_method_signature: RecordAccumulator.registerMetrics(Metrics metrics, String metricGrpName)
focal_class.full_signature: private void registerMetrics(Metrics metrics, String metricGrpName)
focal_class.identifier: registerMetrics
focal_class.modifiers: private
focal_class.parameters: (Metrics metrics, String metricGrpName)
focal_class.return: void
focal_class.signature: void registerMetrics(Metrics metrics, String metricGrpName)
focal_class.class_method_signature: RecordAccumulator.append(TopicPartition tp,
          long timestamp,
          byte[] key,
          byte[] value,
          Callback callback,
          long maxTimeToBlock)
focal_class.full_signature: public RecordAppendResult append(TopicPartition tp,
          long timestamp,
          byte[] key,
          byte[] value,
          Callback callback,
          long maxTimeToBlock)
focal_class.identifier: append
focal_class.modifiers: public
focal_class.parameters: (TopicPartition tp,
          long timestamp,
          byte[] key,
          byte[] value,
          Callback callback,
          long maxTimeToBlock)
focal_class.return: RecordAppendResult
focal_class.signature: RecordAppendResult append(TopicPartition tp,
          long timestamp,
          byte[] key,
          byte[] value,
          Callback callback,
          long maxTimeToBlock)
focal_class.class_method_signature: RecordAccumulator.tryAppend(long timestamp, byte[] key, byte[] value, Callback callback, Deque<RecordBatch> deque)
focal_class.full_signature: private RecordAppendResult tryAppend(long timestamp, byte[] key, byte[] value, Callback callback, Deque<RecordBatch> deque)
focal_class.identifier: tryAppend
focal_class.modifiers: private
focal_class.parameters: (long timestamp, byte[] key, byte[] value, Callback callback, Deque<RecordBatch> deque)
focal_class.return: RecordAppendResult
focal_class.signature: RecordAppendResult tryAppend(long timestamp, byte[] key, byte[] value, Callback callback, Deque<RecordBatch> deque)
focal_class.class_method_signature: RecordAccumulator.abortExpiredBatches(int requestTimeout, long now)
focal_class.full_signature: public List<RecordBatch> abortExpiredBatches(int requestTimeout, long now)
focal_class.identifier: abortExpiredBatches
focal_class.modifiers: public
focal_class.parameters: (int requestTimeout, long now)
focal_class.return: List<RecordBatch>
focal_class.signature: List<RecordBatch> abortExpiredBatches(int requestTimeout, long now)
focal_class.class_method_signature: RecordAccumulator.reenqueue(RecordBatch batch, long now)
focal_class.full_signature: public void reenqueue(RecordBatch batch, long now)
focal_class.identifier: reenqueue
focal_class.modifiers: public
focal_class.parameters: (RecordBatch batch, long now)
focal_class.return: void
focal_class.signature: void reenqueue(RecordBatch batch, long now)
focal_class.class_method_signature: RecordAccumulator.ready(Cluster cluster, long nowMs)
focal_class.full_signature: public ReadyCheckResult ready(Cluster cluster, long nowMs)
focal_class.identifier: ready
focal_class.modifiers: public
focal_class.parameters: (Cluster cluster, long nowMs)
focal_class.return: ReadyCheckResult
focal_class.signature: ReadyCheckResult ready(Cluster cluster, long nowMs)
focal_class.class_method_signature: RecordAccumulator.hasUnsent()
focal_class.full_signature: public boolean hasUnsent()
focal_class.identifier: hasUnsent
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: boolean
focal_class.signature: boolean hasUnsent()
focal_class.class_method_signature: RecordAccumulator.drain(Cluster cluster,
          Set<Node> nodes,
          int maxSize,
          long now)
focal_class.full_signature: public Map<Integer, List<RecordBatch>> drain(Cluster cluster,
          Set<Node> nodes,
          int maxSize,
          long now)
focal_class.identifier: drain
focal_class.modifiers: public
focal_class.parameters: (Cluster cluster,
          Set<Node> nodes,
          int maxSize,
          long now)
focal_class.return: Map<Integer, List<RecordBatch>>
focal_class.signature: Map<Integer, List<RecordBatch>> drain(Cluster cluster,
          Set<Node> nodes,
          int maxSize,
          long now)
focal_class.class_method_signature: RecordAccumulator.getDeque(TopicPartition tp)
focal_class.full_signature: private Deque<RecordBatch> getDeque(TopicPartition tp)
focal_class.identifier: getDeque
focal_class.modifiers: private
focal_class.parameters: (TopicPartition tp)
focal_class.return: Deque<RecordBatch>
focal_class.signature: Deque<RecordBatch> getDeque(TopicPartition tp)
focal_class.class_method_signature: RecordAccumulator.getOrCreateDeque(TopicPartition tp)
focal_class.full_signature: private Deque<RecordBatch> getOrCreateDeque(TopicPartition tp)
focal_class.identifier: getOrCreateDeque
focal_class.modifiers: private
focal_class.parameters: (TopicPartition tp)
focal_class.return: Deque<RecordBatch>
focal_class.signature: Deque<RecordBatch> getOrCreateDeque(TopicPartition tp)
focal_class.class_method_signature: RecordAccumulator.deallocate(RecordBatch batch)
focal_class.full_signature: public void deallocate(RecordBatch batch)
focal_class.identifier: deallocate
focal_class.modifiers: public
focal_class.parameters: (RecordBatch batch)
focal_class.return: void
focal_class.signature: void deallocate(RecordBatch batch)
focal_class.class_method_signature: RecordAccumulator.flushInProgress()
focal_class.full_signature:  boolean flushInProgress()
focal_class.identifier: flushInProgress
focal_class.parameters: ()
focal_class.return: boolean
focal_class.signature: boolean flushInProgress()
focal_class.class_method_signature: RecordAccumulator.batches()
focal_class.full_signature:  Map<TopicPartition, Deque<RecordBatch>> batches()
focal_class.identifier: batches
focal_class.parameters: ()
focal_class.return: Map<TopicPartition, Deque<RecordBatch>>
focal_class.signature: Map<TopicPartition, Deque<RecordBatch>> batches()
focal_class.class_method_signature: RecordAccumulator.beginFlush()
focal_class.full_signature: public void beginFlush()
focal_class.identifier: beginFlush
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void beginFlush()
focal_class.class_method_signature: RecordAccumulator.appendsInProgress()
focal_class.full_signature: private boolean appendsInProgress()
focal_class.identifier: appendsInProgress
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: boolean
focal_class.signature: boolean appendsInProgress()
focal_class.class_method_signature: RecordAccumulator.awaitFlushCompletion()
focal_class.full_signature: public void awaitFlushCompletion()
focal_class.identifier: awaitFlushCompletion
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void awaitFlushCompletion()
focal_class.class_method_signature: RecordAccumulator.abortIncompleteBatches()
focal_class.full_signature: public void abortIncompleteBatches()
focal_class.identifier: abortIncompleteBatches
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void abortIncompleteBatches()
focal_class.class_method_signature: RecordAccumulator.abortBatches()
focal_class.full_signature: private void abortBatches()
focal_class.identifier: abortBatches
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void abortBatches()
focal_class.class_method_signature: RecordAccumulator.mutePartition(TopicPartition tp)
focal_class.full_signature: public void mutePartition(TopicPartition tp)
focal_class.identifier: mutePartition
focal_class.modifiers: public
focal_class.parameters: (TopicPartition tp)
focal_class.return: void
focal_class.signature: void mutePartition(TopicPartition tp)
focal_class.class_method_signature: RecordAccumulator.unmutePartition(TopicPartition tp)
focal_class.full_signature: public void unmutePartition(TopicPartition tp)
focal_class.identifier: unmutePartition
focal_class.modifiers: public
focal_class.parameters: (TopicPartition tp)
focal_class.return: void
focal_class.signature: void unmutePartition(TopicPartition tp)
focal_class.class_method_signature: RecordAccumulator.close()
focal_class.full_signature: public void close()
focal_class.identifier: close
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void close()
focal_method.body: public void abortIncompleteBatches() {
          // We need to keep aborting the incomplete batch until no thread is trying to append to
          // 1. Avoid losing batches.
          // 2. Free up memory in case appending threads are blocked on buffer full.
          // This is a tight loop but should be able to get through very quickly.
          do {
              abortBatches();
          } while (appendsInProgress());
          // After this point, no thread will append any messages because they will see the close
          // flag set. We need to do the last abort after no thread was appending in case there was a new
          // batch appended by the last appending thread.
          abortBatches();
          this.batches.clear();
      }
focal_method.class_method_signature: RecordAccumulator.abortIncompleteBatches()
focal_method.full_signature: public void abortIncompleteBatches()
focal_method.identifier: abortIncompleteBatches
focal_method.invocations: abortBatches appendsInProgress abortBatches clear
focal_method.modifiers: public
focal_method.parameters: ()
focal_method.return: void
focal_method.signature: void abortIncompleteBatches()