Fetcher { public void updateFetchPositions(Set<TopicPartition> partitions) { for (TopicPartition tp : partitions) { if (!subscriptions.isAssigned(tp) || subscriptions.isFetchable(tp)) { continue; } if (subscriptions.isOffsetResetNeeded(tp)) { resetOffset(tp); } else if (subscriptions.committed(tp) == null) { subscriptions.needOffsetReset(tp); resetOffset(tp); } else { long committed = subscriptions.committed(tp) .offset(); log.debug("Resetting offset for partition {} to the committed offset {}", tp, committed); subscriptions.seek(tp, committed); } } } 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); void sendFetches(); Map<TopicPartition, List<ConsumerRecord<K, V>>> fetchedRecords(); void updateFetchPositions(Set<TopicPartition> partitions); Map<String, List<PartitionInfo>> getAllTopicMetadata(long timeout); Map<String, List<PartitionInfo>> getTopicMetadata(MetadataRequest request, long timeout);  }
@Test public void testUpdateFetchPositionToCommitted() { subscriptions.assignFromUser(Arrays.asList(tp)); subscriptions.committed(tp, new OffsetAndMetadata(5)); fetcher.updateFetchPositions(Collections.singleton(tp)); assertTrue(subscriptions.isFetchable(tp)); assertEquals(5, subscriptions.position(tp).longValue()); }
@Test public void testUpdateFetchPositionResetToDefaultOffset() { subscriptions.assignFromUser(Arrays.asList(tp)); 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 public void testUpdateFetchPositionResetToLatestOffset() { subscriptions.assignFromUser(Arrays.asList(tp)); subscriptions.needOffsetReset(tp, OffsetResetStrategy.LATEST); client.prepareResponse(listOffsetRequestMatcher(ListOffsetRequest.LATEST_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 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 public void testUpdateFetchPositionDisconnect() { subscriptions.assignFromUser(Arrays.asList(tp)); subscriptions.needOffsetReset(tp, OffsetResetStrategy.LATEST); client.prepareResponse(listOffsetRequestMatcher(ListOffsetRequest.LATEST_TIMESTAMP), listOffsetResponse(Errors.NONE, Arrays.asList(5L)), true); client.prepareResponse(listOffsetRequestMatcher(ListOffsetRequest.LATEST_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()); }