test_class.declarator: listener = new NoOpConsumerRebalanceListener()
test_class.modifier: private
test_class.original_string: private ConsumerRebalanceListener listener = new NoOpConsumerRebalanceListener();
test_class.type: ConsumerRebalanceListener
test_class.var_name: listener
test_class.declarator: topicName = "test"
test_class.modifier: private
test_class.original_string: private String topicName = "test";
test_class.type: String
test_class.var_name: topicName
test_class.declarator: groupId = "test-group"
test_class.modifier: private
test_class.original_string: private String groupId = "test-group";
test_class.type: String
test_class.var_name: groupId
test_class.declarator: metricGroup = "consumer" + groupId + "-fetch-manager-metrics"
test_class.modifier: private final
test_class.original_string: private final String metricGroup = "consumer" + groupId + "-fetch-manager-metrics";
test_class.type: String
test_class.var_name: metricGroup
test_class.declarator: tp = new TopicPartition(topicName, 0)
test_class.modifier: private
test_class.original_string: private TopicPartition tp = new TopicPartition(topicName, 0);
test_class.type: TopicPartition
test_class.var_name: tp
test_class.declarator: minBytes = 1
test_class.modifier: private
test_class.original_string: private int minBytes = 1;
test_class.type: int
test_class.var_name: minBytes
test_class.declarator: maxWaitMs = 0
test_class.modifier: private
test_class.original_string: private int maxWaitMs = 0;
test_class.type: int
test_class.var_name: maxWaitMs
test_class.declarator: fetchSize = 1000
test_class.modifier: private
test_class.original_string: private int fetchSize = 1000;
test_class.type: int
test_class.var_name: fetchSize
test_class.declarator: retryBackoffMs = 100
test_class.modifier: private
test_class.original_string: private long retryBackoffMs = 100;
test_class.type: long
test_class.var_name: retryBackoffMs
test_class.declarator: time = new MockTime(1)
test_class.modifier: private
test_class.original_string: private MockTime time = new MockTime(1);
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: 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(topicName, 1)
test_class.modifier: private
test_class.original_string: private Cluster cluster = TestUtils.singletonCluster(topicName, 1);
test_class.type: Cluster
test_class.var_name: cluster
test_class.declarator: node = cluster.nodes().get(0)
test_class.modifier: private
test_class.original_string: private Node node = cluster.nodes().get(0);
test_class.type: Node
test_class.var_name: node
test_class.declarator: subscriptions = new SubscriptionState(OffsetResetStrategy.EARLIEST)
test_class.modifier: private
test_class.original_string: private SubscriptionState subscriptions = new SubscriptionState(OffsetResetStrategy.EARLIEST);
test_class.type: SubscriptionState
test_class.var_name: subscriptions
test_class.declarator: subscriptionsNoAutoReset = new SubscriptionState(OffsetResetStrategy.NONE)
test_class.modifier: private
test_class.original_string: private SubscriptionState subscriptionsNoAutoReset = new SubscriptionState(OffsetResetStrategy.NONE);
test_class.type: SubscriptionState
test_class.var_name: subscriptionsNoAutoReset
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: EPSILON = 0.0001
test_class.modifier: private static final
test_class.original_string: private static final double EPSILON = 0.0001;
test_class.type: double
test_class.var_name: EPSILON
test_class.declarator: consumerClient = new ConsumerNetworkClient(client, metadata, time, 100, 1000)
test_class.modifier: private
test_class.original_string: private ConsumerNetworkClient consumerClient = new ConsumerNetworkClient(client, metadata, time, 100, 1000);
test_class.type: ConsumerNetworkClient
test_class.var_name: consumerClient
test_class.declarator: records = MemoryRecords.emptyRecords(ByteBuffer.allocate(1024), CompressionType.NONE)
test_class.modifier: private
test_class.original_string: private MemoryRecords records = MemoryRecords.emptyRecords(ByteBuffer.allocate(1024), CompressionType.NONE);
test_class.type: MemoryRecords
test_class.var_name: records
test_class.declarator: nextRecords = MemoryRecords.emptyRecords(ByteBuffer.allocate(1024), CompressionType.NONE)
test_class.modifier: private
test_class.original_string: private MemoryRecords nextRecords = MemoryRecords.emptyRecords(ByteBuffer.allocate(1024), CompressionType.NONE);
test_class.type: MemoryRecords
test_class.var_name: nextRecords
test_class.declarator: fetcher = createFetcher(subscriptions, metrics)
test_class.modifier: private
test_class.original_string: private Fetcher<byte[], byte[]> fetcher = createFetcher(subscriptions, metrics);
test_class.type: Fetcher<byte[], byte[]>
test_class.var_name: fetcher
test_class.declarator: fetcherMetrics = new Metrics(time)
test_class.modifier: private
test_class.original_string: private Metrics fetcherMetrics = new Metrics(time);
test_class.type: Metrics
test_class.var_name: fetcherMetrics
test_class.declarator: fetcherNoAutoReset = createFetcher(subscriptionsNoAutoReset, fetcherMetrics)
test_class.modifier: private
test_class.original_string: private Fetcher<byte[], byte[]> fetcherNoAutoReset = createFetcher(subscriptionsNoAutoReset, fetcherMetrics);
test_class.type: Fetcher<byte[], byte[]>
test_class.var_name: fetcherNoAutoReset
test_class.file: clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
test_class.identifier: FetcherTest
test_case.body: @Test
      public void testGetTopicMetadataLeaderNotAvailable() {
          client.prepareResponse(newMetadataResponse(topicName, Errors.LEADER_NOT_AVAILABLE).toStruct());
          client.prepareResponse(newMetadataResponse(topicName, Errors.NONE).toStruct());
  
          Map<String, List<PartitionInfo>> topicMetadata = fetcher.getTopicMetadata(new MetadataRequest(Collections.singletonList(topicName)), 5000L);
          assertTrue(topicMetadata.containsKey(topicName));
      }
test_case.class_method_signature: FetcherTest.testGetTopicMetadataLeaderNotAvailable()
test_case.full_signature: @Test public void testGetTopicMetadataLeaderNotAvailable()
test_case.identifier: testGetTopicMetadataLeaderNotAvailable
test_case.invocations: prepareResponse toStruct newMetadataResponse prepareResponse toStruct newMetadataResponse getTopicMetadata singletonList assertTrue containsKey
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testGetTopicMetadataLeaderNotAvailable()
focal_class.declarator: log = LoggerFactory.getLogger(Fetcher.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger log = LoggerFactory.getLogger(Fetcher.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 ConsumerNetworkClient client;
focal_class.type: ConsumerNetworkClient
focal_class.var_name: client
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: minBytes
focal_class.modifier: private final
focal_class.original_string: private final int minBytes;
focal_class.type: int
focal_class.var_name: minBytes
focal_class.declarator: maxWaitMs
focal_class.modifier: private final
focal_class.original_string: private final int maxWaitMs;
focal_class.type: int
focal_class.var_name: maxWaitMs
focal_class.declarator: fetchSize
focal_class.modifier: private final
focal_class.original_string: private final int fetchSize;
focal_class.type: int
focal_class.var_name: fetchSize
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: maxPollRecords
focal_class.modifier: private final
focal_class.original_string: private final int maxPollRecords;
focal_class.type: int
focal_class.var_name: maxPollRecords
focal_class.declarator: checkCrcs
focal_class.modifier: private final
focal_class.original_string: private final boolean checkCrcs;
focal_class.type: boolean
focal_class.var_name: checkCrcs
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: sensors
focal_class.modifier: private final
focal_class.original_string: private final FetchManagerMetrics sensors;
focal_class.type: FetchManagerMetrics
focal_class.var_name: sensors
focal_class.declarator: subscriptions
focal_class.modifier: private final
focal_class.original_string: private final SubscriptionState subscriptions;
focal_class.type: SubscriptionState
focal_class.var_name: subscriptions
focal_class.declarator: completedFetches
focal_class.modifier: private final
focal_class.original_string: private final List<CompletedFetch> completedFetches;
focal_class.type: List<CompletedFetch>
focal_class.var_name: completedFetches
focal_class.declarator: keyDeserializer
focal_class.modifier: private final
focal_class.original_string: private final Deserializer<K> keyDeserializer;
focal_class.type: Deserializer<K>
focal_class.var_name: keyDeserializer
focal_class.declarator: valueDeserializer
focal_class.modifier: private final
focal_class.original_string: private final Deserializer<V> valueDeserializer;
focal_class.type: Deserializer<V>
focal_class.var_name: valueDeserializer
focal_class.declarator: nextInLineRecords = null
focal_class.modifier: private
focal_class.original_string: private PartitionRecords<K, V> nextInLineRecords = null;
focal_class.type: PartitionRecords<K, V>
focal_class.var_name: nextInLineRecords
focal_class.file: clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
focal_class.identifier: Fetcher
focal_class.class_method_signature: Fetcher.Fetcher(ConsumerNetworkClient client,
          int minBytes,
          int maxWaitMs,
          int fetchSize,
          int maxPollRecords,
          boolean checkCrcs,
          Deserializer<K> keyDeserializer,
          Deserializer<V> valueDeserializer,
          Metadata metadata,
          SubscriptionState subscriptions,
          Metrics metrics,
          String metricGrpPrefix,
          Time time,
          long retryBackoffMs)
focal_class.full_signature: public  Fetcher(ConsumerNetworkClient client,
          int minBytes,
          int maxWaitMs,
          int fetchSize,
          int maxPollRecords,
          boolean checkCrcs,
          Deserializer<K> keyDeserializer,
          Deserializer<V> valueDeserializer,
          Metadata metadata,
          SubscriptionState subscriptions,
          Metrics metrics,
          String metricGrpPrefix,
          Time time,
          long retryBackoffMs)
focal_class.identifier: Fetcher
focal_class.modifiers: public
focal_class.parameters: (ConsumerNetworkClient client,
          int minBytes,
          int maxWaitMs,
          int fetchSize,
          int maxPollRecords,
          boolean checkCrcs,
          Deserializer<K> keyDeserializer,
          Deserializer<V> valueDeserializer,
          Metadata metadata,
          SubscriptionState subscriptions,
          Metrics metrics,
          String metricGrpPrefix,
          Time time,
          long retryBackoffMs)
focal_class.signature:  Fetcher(ConsumerNetworkClient client,
          int minBytes,
          int maxWaitMs,
          int fetchSize,
          int maxPollRecords,
          boolean checkCrcs,
          Deserializer<K> keyDeserializer,
          Deserializer<V> valueDeserializer,
          Metadata metadata,
          SubscriptionState subscriptions,
          Metrics metrics,
          String metricGrpPrefix,
          Time time,
          long retryBackoffMs)
focal_class.class_method_signature: Fetcher.sendFetches()
focal_class.full_signature: public void sendFetches()
focal_class.identifier: sendFetches
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void sendFetches()
focal_class.class_method_signature: Fetcher.parseFetchedData(CompletedFetch completedFetch)
focal_class.full_signature: private PartitionRecords<K, V> parseFetchedData(CompletedFetch completedFetch)
focal_class.identifier: parseFetchedData
focal_class.modifiers: private
focal_class.parameters: (CompletedFetch completedFetch)
focal_class.return: PartitionRecords<K, V>
focal_class.signature: PartitionRecords<K, V> parseFetchedData(CompletedFetch completedFetch)
focal_class.class_method_signature: Fetcher.fetchedRecords()
focal_class.full_signature: public Map<TopicPartition, List<ConsumerRecord<K, V>>> fetchedRecords()
focal_class.identifier: fetchedRecords
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: Map<TopicPartition, List<ConsumerRecord<K, V>>>
focal_class.signature: Map<TopicPartition, List<ConsumerRecord<K, V>>> fetchedRecords()
focal_class.class_method_signature: Fetcher.append(Map<TopicPartition, List<ConsumerRecord<K, V>>> drained, PartitionRecords<K, V> partitionRecords, int maxRecords)
focal_class.full_signature: private int append(Map<TopicPartition, List<ConsumerRecord<K, V>>> drained, PartitionRecords<K, V> partitionRecords, int maxRecords)
focal_class.identifier: append
focal_class.modifiers: private
focal_class.parameters: (Map<TopicPartition, List<ConsumerRecord<K, V>>> drained, PartitionRecords<K, V> partitionRecords, int maxRecords)
focal_class.return: int
focal_class.signature: int append(Map<TopicPartition, List<ConsumerRecord<K, V>>> drained, PartitionRecords<K, V> partitionRecords, int maxRecords)
focal_class.class_method_signature: Fetcher.updateFetchPositions(Set<TopicPartition> partitions)
focal_class.full_signature: public void updateFetchPositions(Set<TopicPartition> partitions)
focal_class.identifier: updateFetchPositions
focal_class.modifiers: public
focal_class.parameters: (Set<TopicPartition> partitions)
focal_class.return: void
focal_class.signature: void updateFetchPositions(Set<TopicPartition> partitions)
focal_class.class_method_signature: Fetcher.getAllTopicMetadata(long timeout)
focal_class.full_signature: public Map<String, List<PartitionInfo>> getAllTopicMetadata(long timeout)
focal_class.identifier: getAllTopicMetadata
focal_class.modifiers: public
focal_class.parameters: (long timeout)
focal_class.return: Map<String, List<PartitionInfo>>
focal_class.signature: Map<String, List<PartitionInfo>> getAllTopicMetadata(long timeout)
focal_class.class_method_signature: Fetcher.getTopicMetadata(MetadataRequest request, long timeout)
focal_class.full_signature: public Map<String, List<PartitionInfo>> getTopicMetadata(MetadataRequest request, long timeout)
focal_class.identifier: getTopicMetadata
focal_class.modifiers: public
focal_class.parameters: (MetadataRequest request, long timeout)
focal_class.return: Map<String, List<PartitionInfo>>
focal_class.signature: Map<String, List<PartitionInfo>> getTopicMetadata(MetadataRequest request, long timeout)
focal_class.class_method_signature: Fetcher.sendMetadataRequest(MetadataRequest request)
focal_class.full_signature: private RequestFuture<ClientResponse> sendMetadataRequest(MetadataRequest request)
focal_class.identifier: sendMetadataRequest
focal_class.modifiers: private
focal_class.parameters: (MetadataRequest request)
focal_class.return: RequestFuture<ClientResponse>
focal_class.signature: RequestFuture<ClientResponse> sendMetadataRequest(MetadataRequest request)
focal_class.class_method_signature: Fetcher.resetOffset(TopicPartition partition)
focal_class.full_signature: private void resetOffset(TopicPartition partition)
focal_class.identifier: resetOffset
focal_class.modifiers: private
focal_class.parameters: (TopicPartition partition)
focal_class.return: void
focal_class.signature: void resetOffset(TopicPartition partition)
focal_class.class_method_signature: Fetcher.listOffset(TopicPartition partition, long timestamp)
focal_class.full_signature: private long listOffset(TopicPartition partition, long timestamp)
focal_class.identifier: listOffset
focal_class.modifiers: private
focal_class.parameters: (TopicPartition partition, long timestamp)
focal_class.return: long
focal_class.signature: long listOffset(TopicPartition partition, long timestamp)
focal_class.class_method_signature: Fetcher.sendListOffsetRequest(final TopicPartition topicPartition, long timestamp)
focal_class.full_signature: private RequestFuture<Long> sendListOffsetRequest(final TopicPartition topicPartition, long timestamp)
focal_class.identifier: sendListOffsetRequest
focal_class.modifiers: private
focal_class.parameters: (final TopicPartition topicPartition, long timestamp)
focal_class.return: RequestFuture<Long>
focal_class.signature: RequestFuture<Long> sendListOffsetRequest(final TopicPartition topicPartition, long timestamp)
focal_class.class_method_signature: Fetcher.handleListOffsetResponse(TopicPartition topicPartition,
          ClientResponse clientResponse,
          RequestFuture<Long> future)
focal_class.full_signature: private void handleListOffsetResponse(TopicPartition topicPartition,
          ClientResponse clientResponse,
          RequestFuture<Long> future)
focal_class.identifier: handleListOffsetResponse
focal_class.modifiers: private
focal_class.parameters: (TopicPartition topicPartition,
          ClientResponse clientResponse,
          RequestFuture<Long> future)
focal_class.return: void
focal_class.signature: void handleListOffsetResponse(TopicPartition topicPartition,
          ClientResponse clientResponse,
          RequestFuture<Long> future)
focal_class.class_method_signature: Fetcher.fetchablePartitions()
focal_class.full_signature: private Set<TopicPartition> fetchablePartitions()
focal_class.identifier: fetchablePartitions
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: Set<TopicPartition>
focal_class.signature: Set<TopicPartition> fetchablePartitions()
focal_class.class_method_signature: Fetcher.createFetchRequests()
focal_class.full_signature: private Map<Node, FetchRequest> createFetchRequests()
focal_class.identifier: createFetchRequests
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: Map<Node, FetchRequest>
focal_class.signature: Map<Node, FetchRequest> createFetchRequests()
focal_class.class_method_signature: Fetcher.parseRecord(TopicPartition partition, LogEntry logEntry)
focal_class.full_signature: private ConsumerRecord<K, V> parseRecord(TopicPartition partition, LogEntry logEntry)
focal_class.identifier: parseRecord
focal_class.modifiers: private
focal_class.parameters: (TopicPartition partition, LogEntry logEntry)
focal_class.return: ConsumerRecord<K, V>
focal_class.signature: ConsumerRecord<K, V> parseRecord(TopicPartition partition, LogEntry logEntry)
focal_method.body: public Map<String, List<PartitionInfo>> getTopicMetadata(MetadataRequest request, long timeout) {
          // Save the round trip if no topics are requested.
          if (!request.isAllTopics() && request.topics()
                                               .isEmpty()) {
              return Collections.emptyMap();
          }
  
          long start = time.milliseconds();
          long remaining = timeout;
  
          do {
              RequestFuture<ClientResponse> future = sendMetadataRequest(request);
              client.poll(future, remaining);
  
              if (future.failed() && !future.isRetriable()) {
                  throw future.exception();
              }
  
              if (future.succeeded()) {
                  MetadataResponse response = new MetadataResponse(future.value()
                                                                         .responseBody());
                  Cluster cluster = response.cluster();
  
                  Set<String> unauthorizedTopics = cluster.unauthorizedTopics();
                  if (!unauthorizedTopics.isEmpty()) {
                      throw new TopicAuthorizationException(unauthorizedTopics);
                  }
  
                  boolean shouldRetry = false;
                  Map<String, Errors> errors = response.errors();
                  if (!errors.isEmpty()) {
                      // if there were errors, we need to check whether they were fatal or whether
                      // we should just retry
  
                      log.debug("Topic metadata fetch included errors: {}", errors);
  
                      for (Map.Entry<String, Errors> errorEntry : errors.entrySet()) {
                          String topic = errorEntry.getKey();
                          Errors error = errorEntry.getValue();
  
                          if (error == Errors.INVALID_TOPIC_EXCEPTION) {
                              throw new InvalidTopicException("Topic '" + topic + "' is invalid");
                          } else if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION)
                          // if a requested topic is unknown, we just continue and let it be absent
                          // in the returned map
                          {
                              continue;
                          } else if (error.exception() instanceof RetriableException) {
                              shouldRetry = true;
                          } else {
                              throw new KafkaException("Unexpected error fetching metadata for topic " + topic,
                                  error.exception());
                          }
                      }
                  }
  
                  if (!shouldRetry) {
                      HashMap<String, List<PartitionInfo>> topicsPartitionInfos = new HashMap<>();
                      for (String topic : cluster.topics())
                          topicsPartitionInfos.put(topic, cluster.availablePartitionsForTopic(topic));
                      return topicsPartitionInfos;
                  }
              }
  
              long elapsed = time.milliseconds() - start;
              remaining = timeout - elapsed;
  
              if (remaining > 0) {
                  long backoff = Math.min(remaining, retryBackoffMs);
                  time.sleep(backoff);
                  remaining -= backoff;
              }
          } while (remaining > 0);
  
          throw new TimeoutException("Timeout expired while fetching topic metadata");
      }
focal_method.class_method_signature: Fetcher.getTopicMetadata(MetadataRequest request, long timeout)
focal_method.full_signature: public Map<String, List<PartitionInfo>> getTopicMetadata(MetadataRequest request, long timeout)
focal_method.identifier: getTopicMetadata
focal_method.invocations: isAllTopics isEmpty topics emptyMap milliseconds sendMetadataRequest poll failed isRetriable exception succeeded responseBody value cluster unauthorizedTopics isEmpty errors isEmpty debug entrySet getKey getValue exception exception topics put availablePartitionsForTopic milliseconds min sleep
focal_method.modifiers: public
focal_method.parameters: (MetadataRequest request, long timeout)
focal_method.return: Map<String, List<PartitionInfo>>
focal_method.signature: Map<String, List<PartitionInfo>> getTopicMetadata(MetadataRequest request, long timeout)