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 testUpdateFetchPositionResetToEarliestOffset() {
          subscriptions.assignFromUser(Arrays.asList(tp));
          subscriptions.needOffsetReset(tp, OffsetResetStrategy.EARLIEST);
  
          client.prepareResponse(listOffsetRequestMatcher(ListOffsetRequest.EARLIEST_TIMESTAMP),
                                 listOffsetResponse(Errors.NONE, Arrays.asList(5L)));
          fetcher.updateFetchPositions(Collections.singleton(tp));
          assertFalse(subscriptions.isOffsetResetNeeded(tp));
          assertTrue(subscriptions.isFetchable(tp));
          assertEquals(5, subscriptions.position(tp).longValue());
      }
test_case.class_method_signature: FetcherTest.testUpdateFetchPositionResetToEarliestOffset()
test_case.full_signature: @Test public void testUpdateFetchPositionResetToEarliestOffset()
test_case.identifier: testUpdateFetchPositionResetToEarliestOffset
test_case.invocations: assignFromUser asList needOffsetReset prepareResponse listOffsetRequestMatcher listOffsetResponse asList updateFetchPositions singleton assertFalse isOffsetResetNeeded assertTrue isFetchable assertEquals longValue position
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testUpdateFetchPositionResetToEarliestOffset()
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 void updateFetchPositions(Set<TopicPartition> partitions) {
          // reset the fetch position to the committed position
          for (TopicPartition tp : partitions) {
              if (!subscriptions.isAssigned(tp) || subscriptions.isFetchable(tp)) {// 没分配或者没暂停之类的才能update
                  continue;
              }
  
              // TODO: If there are several offsets to reset, we could submit offset requests in parallel
              if (subscriptions.isOffsetResetNeeded(tp)) {// resetStrategy != null; 重设策略不为空
                  resetOffset(tp);
              } else if (subscriptions.committed(tp) == null) {
                  // there's no committed position, so we need to reset with the default strategy
                  subscriptions.needOffsetReset(tp);
                  resetOffset(tp);
              } else {
                  // 如果没有重置策略则简单将commit设置为offset
                  long committed = subscriptions.committed(tp)
                                                .offset();
                  log.debug("Resetting offset for partition {} to the committed offset {}", tp, committed);
                  subscriptions.seek(tp, committed);
              }
          }
      }
focal_method.class_method_signature: Fetcher.updateFetchPositions(Set<TopicPartition> partitions)
focal_method.full_signature: public void updateFetchPositions(Set<TopicPartition> partitions)
focal_method.identifier: updateFetchPositions
focal_method.invocations: isAssigned isFetchable isOffsetResetNeeded resetOffset committed needOffsetReset resetOffset offset committed debug seek
focal_method.modifiers: public
focal_method.parameters: (Set<TopicPartition> partitions)
focal_method.return: void
focal_method.signature: void updateFetchPositions(Set<TopicPartition> partitions)