test_class.declarator: MAX_REQUEST_SIZE = 1024 * 1024
test_class.modifier: private static final
test_class.original_string: private static final int MAX_REQUEST_SIZE = 1024 * 1024;
test_class.type: int
test_class.var_name: MAX_REQUEST_SIZE
test_class.declarator: ACKS_ALL = -1
test_class.modifier: private static final
test_class.original_string: private static final short ACKS_ALL = -1;
test_class.type: short
test_class.var_name: ACKS_ALL
test_class.declarator: MAX_RETRIES = 0
test_class.modifier: private static final
test_class.original_string: private static final int MAX_RETRIES = 0;
test_class.type: int
test_class.var_name: MAX_RETRIES
test_class.declarator: CLIENT_ID = "clientId"
test_class.modifier: private static final
test_class.original_string: private static final String CLIENT_ID = "clientId";
test_class.type: String
test_class.var_name: CLIENT_ID
test_class.declarator: METRIC_GROUP = "producer-metrics"
test_class.modifier: private static final
test_class.original_string: private static final String METRIC_GROUP = "producer-metrics";
test_class.type: String
test_class.var_name: METRIC_GROUP
test_class.declarator: EPS = 0.0001
test_class.modifier: private static final
test_class.original_string: private static final double EPS = 0.0001;
test_class.type: double
test_class.var_name: EPS
test_class.declarator: MAX_BLOCK_TIMEOUT = 1000
test_class.modifier: private static final
test_class.original_string: private static final int MAX_BLOCK_TIMEOUT = 1000;
test_class.type: int
test_class.var_name: MAX_BLOCK_TIMEOUT
test_class.declarator: REQUEST_TIMEOUT = 1000
test_class.modifier: private static final
test_class.original_string: private static final int REQUEST_TIMEOUT = 1000;
test_class.type: int
test_class.var_name: REQUEST_TIMEOUT
test_class.declarator: tp = new TopicPartition("test", 0)
test_class.modifier: private
test_class.original_string: private TopicPartition tp = new TopicPartition("test", 0);
test_class.type: TopicPartition
test_class.var_name: tp
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: client = new MockClient(time)
test_class.modifier: private
test_class.original_string: private MockClient client = new MockClient(time);
test_class.type: MockClient
test_class.var_name: client
test_class.declarator: batchSize = 16 * 1024
test_class.modifier: private
test_class.original_string: private int batchSize = 16 * 1024;
test_class.type: int
test_class.var_name: batchSize
test_class.declarator: metadata = new Metadata(0, Long.MAX_VALUE)
test_class.modifier: private
test_class.original_string: private Metadata metadata = new Metadata(0, Long.MAX_VALUE);
test_class.type: Metadata
test_class.var_name: metadata
test_class.declarator: cluster = TestUtils.singletonCluster("test", 1)
test_class.modifier: private
test_class.original_string: private Cluster cluster = TestUtils.singletonCluster("test", 1);
test_class.type: Cluster
test_class.var_name: cluster
test_class.declarator: metrics = null
test_class.modifier: private
test_class.original_string: private Metrics metrics = null;
test_class.type: Metrics
test_class.var_name: metrics
test_class.declarator: accumulator = null
test_class.modifier: private
test_class.original_string: private RecordAccumulator accumulator = null;
test_class.type: RecordAccumulator
test_class.var_name: accumulator
test_class.declarator: sender = null
test_class.modifier: private
test_class.original_string: private Sender sender = null;
test_class.type: Sender
test_class.var_name: sender
test_class.file: clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
test_class.identifier: SenderTest
test_case.body: @Test
      public void testSendInOrder() throws Exception {
          int maxRetries = 1;
          Metrics m = new Metrics();
          try {
              Sender sender = new Sender(client,
                  metadata,
                  this.accumulator,
                  true,
                  MAX_REQUEST_SIZE,
                  ACKS_ALL,
                  maxRetries,
                  m,
                  time,
                  "clientId",
                  REQUEST_TIMEOUT);
  
              // Create a two broker cluster, with partition 0 on broker 0 and partition 1 on broker 1
              Cluster cluster1 = TestUtils.clusterWith(2, "test", 2);
              metadata.update(cluster1, time.milliseconds());
  
              // Send the first message.
              TopicPartition tp2 = new TopicPartition("test", 1);
              accumulator.append(tp2, 0L, "key1".getBytes(), "value1".getBytes(), null, MAX_BLOCK_TIMEOUT);
              sender.run(time.milliseconds()); // connect
              sender.run(time.milliseconds()); // send produce request
              String id = client.requests().peek().request().destination();
              assertEquals(ApiKeys.PRODUCE.id, client.requests().peek().request().header().apiKey());
              Node node = new Node(Integer.valueOf(id), "localhost", 0);
              assertEquals(1, client.inFlightRequestCount());
              assertTrue("Client ready status should be true", client.isReady(node, 0L));
  
              time.sleep(900);
              // Now send another message to tp2
              accumulator.append(tp2, 0L, "key2".getBytes(), "value2".getBytes(), null, MAX_BLOCK_TIMEOUT);
  
              // Update metadata before sender receives response from broker 0. Now partition 2 moves to broker 0
              Cluster cluster2 = TestUtils.singletonCluster("test", 2);
              metadata.update(cluster2, time.milliseconds());
              // Sender should not send the second message to node 0.
              sender.run(time.milliseconds());
              assertEquals(1, client.inFlightRequestCount());
          } finally {
              m.close();
          }
  
      }
test_case.class_method_signature: SenderTest.testSendInOrder()
test_case.full_signature: @Test public void testSendInOrder()
test_case.identifier: testSendInOrder
test_case.invocations: clusterWith update milliseconds append getBytes getBytes run milliseconds run milliseconds destination request peek requests assertEquals apiKey header request peek requests valueOf assertEquals inFlightRequestCount assertTrue isReady sleep append getBytes getBytes singletonCluster update milliseconds run milliseconds assertEquals inFlightRequestCount close
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testSendInOrder()
focal_class.declarator: log = LoggerFactory.getLogger(Sender.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger log = LoggerFactory.getLogger(Sender.class);
focal_class.type: Logger
focal_class.var_name: log
focal_class.declarator: client
focal_class.modifier: private final
focal_class.original_string: private final KafkaClient client;
focal_class.type: KafkaClient
focal_class.var_name: client
focal_class.declarator: accumulator
focal_class.modifier: private final
focal_class.original_string: private final RecordAccumulator accumulator;
focal_class.type: RecordAccumulator
focal_class.var_name: accumulator
focal_class.declarator: metadata
focal_class.modifier: private final
focal_class.original_string: private final Metadata metadata;
focal_class.type: Metadata
focal_class.var_name: metadata
focal_class.declarator: guaranteeMessageOrder
focal_class.modifier: private final
focal_class.original_string: private final boolean guaranteeMessageOrder;
focal_class.type: boolean
focal_class.var_name: guaranteeMessageOrder
focal_class.declarator: maxRequestSize
focal_class.modifier: private final
focal_class.original_string: private final int maxRequestSize;
focal_class.type: int
focal_class.var_name: maxRequestSize
focal_class.declarator: acks
focal_class.modifier: private final
focal_class.original_string: private final short acks;
focal_class.type: short
focal_class.var_name: acks
focal_class.declarator: retries
focal_class.modifier: private final
focal_class.original_string: private final int retries;
focal_class.type: int
focal_class.var_name: retries
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: running
focal_class.modifier: private volatile
focal_class.original_string: private volatile boolean running;
focal_class.type: boolean
focal_class.var_name: running
focal_class.declarator: forceClose
focal_class.modifier: private volatile
focal_class.original_string: private volatile boolean forceClose;
focal_class.type: boolean
focal_class.var_name: forceClose
focal_class.declarator: sensors
focal_class.modifier: private final
focal_class.original_string: private final SenderMetrics sensors;
focal_class.type: SenderMetrics
focal_class.var_name: sensors
focal_class.declarator: clientId
focal_class.modifier: private
focal_class.original_string: private String clientId;
focal_class.type: String
focal_class.var_name: clientId
focal_class.declarator: requestTimeout
focal_class.modifier: private final
focal_class.original_string: private final int requestTimeout;
focal_class.type: int
focal_class.var_name: requestTimeout
focal_class.file: clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
focal_class.identifier: Sender
focal_class.interfaces: implements Runnable
focal_class.class_method_signature: Sender.Sender(KafkaClient client,
          Metadata metadata,
          RecordAccumulator accumulator,
          boolean guaranteeMessageOrder,
          int maxRequestSize,
          short acks,
          int retries,
          Metrics metrics,
          Time time,
          String clientId,
          int requestTimeout)
focal_class.full_signature: public  Sender(KafkaClient client,
          Metadata metadata,
          RecordAccumulator accumulator,
          boolean guaranteeMessageOrder,
          int maxRequestSize,
          short acks,
          int retries,
          Metrics metrics,
          Time time,
          String clientId,
          int requestTimeout)
focal_class.identifier: Sender
focal_class.modifiers: public
focal_class.parameters: (KafkaClient client,
          Metadata metadata,
          RecordAccumulator accumulator,
          boolean guaranteeMessageOrder,
          int maxRequestSize,
          short acks,
          int retries,
          Metrics metrics,
          Time time,
          String clientId,
          int requestTimeout)
focal_class.signature:  Sender(KafkaClient client,
          Metadata metadata,
          RecordAccumulator accumulator,
          boolean guaranteeMessageOrder,
          int maxRequestSize,
          short acks,
          int retries,
          Metrics metrics,
          Time time,
          String clientId,
          int requestTimeout)
focal_class.class_method_signature: Sender.run()
focal_class.full_signature: public void run()
focal_class.identifier: run
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void run()
focal_class.class_method_signature: Sender.run(long now)
focal_class.full_signature:  void run(long now)
focal_class.identifier: run
focal_class.parameters: (long now)
focal_class.return: void
focal_class.signature: void run(long now)
focal_class.class_method_signature: Sender.initiateClose()
focal_class.full_signature: public void initiateClose()
focal_class.identifier: initiateClose
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void initiateClose()
focal_class.class_method_signature: Sender.forceClose()
focal_class.full_signature: public void forceClose()
focal_class.identifier: forceClose
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void forceClose()
focal_class.class_method_signature: Sender.handleProduceResponse(ClientResponse response, Map<TopicPartition, RecordBatch> batches, long now)
focal_class.full_signature: private void handleProduceResponse(ClientResponse response, Map<TopicPartition, RecordBatch> batches, long now)
focal_class.identifier: handleProduceResponse
focal_class.modifiers: private
focal_class.parameters: (ClientResponse response, Map<TopicPartition, RecordBatch> batches, long now)
focal_class.return: void
focal_class.signature: void handleProduceResponse(ClientResponse response, Map<TopicPartition, RecordBatch> batches, long now)
focal_class.class_method_signature: Sender.completeBatch(RecordBatch batch, Errors error, long baseOffset, long timestamp, long correlationId, long now)
focal_class.full_signature: private void completeBatch(RecordBatch batch, Errors error, long baseOffset, long timestamp, long correlationId, long now)
focal_class.identifier: completeBatch
focal_class.modifiers: private
focal_class.parameters: (RecordBatch batch, Errors error, long baseOffset, long timestamp, long correlationId, long now)
focal_class.return: void
focal_class.signature: void completeBatch(RecordBatch batch, Errors error, long baseOffset, long timestamp, long correlationId, long now)
focal_class.class_method_signature: Sender.canRetry(RecordBatch batch, Errors error)
focal_class.full_signature: private boolean canRetry(RecordBatch batch, Errors error)
focal_class.identifier: canRetry
focal_class.modifiers: private
focal_class.parameters: (RecordBatch batch, Errors error)
focal_class.return: boolean
focal_class.signature: boolean canRetry(RecordBatch batch, Errors error)
focal_class.class_method_signature: Sender.createProduceRequests(Map<Integer/* nodeId */, List<RecordBatch>> collated, long now)
focal_class.full_signature: private List<ClientRequest> createProduceRequests(Map<Integer/* nodeId */, List<RecordBatch>> collated, long now)
focal_class.identifier: createProduceRequests
focal_class.modifiers: private
focal_class.parameters: (Map<Integer/* nodeId */, List<RecordBatch>> collated, long now)
focal_class.return: List<ClientRequest>
focal_class.signature: List<ClientRequest> createProduceRequests(Map<Integer/* nodeId */, List<RecordBatch>> collated, long now)
focal_class.class_method_signature: Sender.produceRequest(long now, int destination, short acks, int timeout, List<RecordBatch> batches)
focal_class.full_signature: private ClientRequest produceRequest(long now, int destination, short acks, int timeout, List<RecordBatch> batches)
focal_class.identifier: produceRequest
focal_class.modifiers: private
focal_class.parameters: (long now, int destination, short acks, int timeout, List<RecordBatch> batches)
focal_class.return: ClientRequest
focal_class.signature: ClientRequest produceRequest(long now, int destination, short acks, int timeout, List<RecordBatch> batches)
focal_class.class_method_signature: Sender.wakeup()
focal_class.full_signature: public void wakeup()
focal_class.identifier: wakeup
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void wakeup()
focal_method.body: public void run() {
          log.debug("Starting Kafka producer I/O thread.");
  
          // main loop, runs until close is called
          while (running) {
              try {
                  run(time.milliseconds());
              } catch (Exception e) {
                  log.error("Uncaught error in kafka producer I/O thread: ", e);
              }
          }
  
          log.debug("Beginning shutdown of Kafka producer I/O thread, sending remaining records.");
  
          // okay we stopped accepting requests but there may still be
          // requests in the accumulator or waiting for acknowledgment,
          // wait until these are completed.
          while (!forceClose && (this.accumulator.hasUnsent() || this.client.inFlightRequestCount() > 0)) {
              try {
                  run(time.milliseconds());
              } catch (Exception e) {
                  log.error("Uncaught error in kafka producer I/O thread: ", e);
              }
          }
          if (forceClose) {
              // We need to fail all the incomplete batches and wake up the threads waiting on
              // the futures.
              this.accumulator.abortIncompleteBatches();
          }
          try {
              this.client.close();
          } catch (Exception e) {
              log.error("Failed to close network client", e);
          }
  
          log.debug("Shutdown of Kafka producer I/O thread has completed.");
      }
focal_method.class_method_signature: Sender.run()
focal_method.full_signature: public void run()
focal_method.identifier: run
focal_method.invocations: debug run milliseconds error debug hasUnsent inFlightRequestCount run milliseconds error abortIncompleteBatches close error debug
focal_method.modifiers: public
focal_method.parameters: ()
focal_method.return: void
focal_method.signature: void run()