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 testQuotaMetrics() throws Exception {
          final long offset = 0;
          for (int i = 1; i <= 3; i++) {
              Future<RecordMetadata> future = accumulator.append(tp, 0L, "key".getBytes(), "value".getBytes(), null, MAX_BLOCK_TIMEOUT).future;
              sender.run(time.milliseconds()); // send produce request
              client.respond(produceResponse(tp, offset, Errors.NONE.code(), 100 * i));
              sender.run(time.milliseconds());
          }
          Map<MetricName, KafkaMetric> allMetrics = metrics.metrics();
          KafkaMetric avgMetric = allMetrics.get(metrics.metricName("produce-throttle-time-avg", METRIC_GROUP, ""));
          KafkaMetric maxMetric = allMetrics.get(metrics.metricName("produce-throttle-time-max", METRIC_GROUP, ""));
          assertEquals(200, avgMetric.value(), EPS);
          assertEquals(300, maxMetric.value(), EPS);
      }
test_case.class_method_signature: SenderTest.testQuotaMetrics()
test_case.full_signature: @Test public void testQuotaMetrics()
test_case.identifier: testQuotaMetrics
test_case.invocations: append getBytes getBytes run milliseconds respond produceResponse code run milliseconds metrics get metricName get metricName assertEquals value assertEquals value
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testQuotaMetrics()
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()