@Test public void testListOffsetsSendsIsolationLevel() { for (final IsolationLevel isolationLevel : IsolationLevel.values()) { Fetcher<byte[], byte[]> fetcher = createFetcher(subscriptions, new Metrics(), new ByteArrayDeserializer(), new ByteArrayDeserializer(), Integer.MAX_VALUE, isolationLevel); subscriptions.assignFromUser(singleton(tp1)); subscriptions.needOffsetReset(tp1, OffsetResetStrategy.LATEST); client.prepareResponse(new MockClient.RequestMatcher() { @Override public boolean matches(AbstractRequest body) { ListOffsetRequest request = (ListOffsetRequest) body; return request.isolationLevel() == isolationLevel; } }, listOffsetResponse(Errors.NONE, 1L, 5L)); fetcher.updateFetchPositions(singleton(tp1)); assertFalse(subscriptions.isOffsetResetNeeded(tp1)); assertTrue(subscriptions.isFetchable(tp1)); assertEquals(5, subscriptions.position(tp1).longValue()); } }
public void updateFetchPositions(Set<TopicPartition> partitions) { final Set<TopicPartition> needsOffsetReset = new HashSet<>(); for (TopicPartition tp : partitions) { if (!subscriptions.isAssigned(tp) || subscriptions.hasValidPosition(tp)) continue; if (subscriptions.isOffsetResetNeeded(tp)) { needsOffsetReset.add(tp); } else if (subscriptions.committed(tp) == null) { subscriptions.needOffsetReset(tp); needsOffsetReset.add(tp); } else { long committed = subscriptions.committed(tp).offset(); log.debug("Resetting offset for partition {} to the committed offset {}", tp, committed); subscriptions.seek(tp, committed); } } if (!needsOffsetReset.isEmpty()) { resetOffsets(needsOffsetReset); } }
Fetcher implements SubscriptionState.Listener, Closeable { public void updateFetchPositions(Set<TopicPartition> partitions) { final Set<TopicPartition> needsOffsetReset = new HashSet<>(); for (TopicPartition tp : partitions) { if (!subscriptions.isAssigned(tp) || subscriptions.hasValidPosition(tp)) continue; if (subscriptions.isOffsetResetNeeded(tp)) { needsOffsetReset.add(tp); } else if (subscriptions.committed(tp) == null) { subscriptions.needOffsetReset(tp); needsOffsetReset.add(tp); } else { long committed = subscriptions.committed(tp).offset(); log.debug("Resetting offset for partition {} to the committed offset {}", tp, committed); subscriptions.seek(tp, committed); } } if (!needsOffsetReset.isEmpty()) { resetOffsets(needsOffsetReset); } } }
Fetcher implements SubscriptionState.Listener, Closeable { public void updateFetchPositions(Set<TopicPartition> partitions) { final Set<TopicPartition> needsOffsetReset = new HashSet<>(); for (TopicPartition tp : partitions) { if (!subscriptions.isAssigned(tp) || subscriptions.hasValidPosition(tp)) continue; if (subscriptions.isOffsetResetNeeded(tp)) { needsOffsetReset.add(tp); } else if (subscriptions.committed(tp) == null) { subscriptions.needOffsetReset(tp); needsOffsetReset.add(tp); } else { long committed = subscriptions.committed(tp).offset(); log.debug("Resetting offset for partition {} to the committed offset {}", tp, committed); subscriptions.seek(tp, committed); } } if (!needsOffsetReset.isEmpty()) { resetOffsets(needsOffsetReset); } } Fetcher(ConsumerNetworkClient client,
                   int minBytes,
                   int maxBytes,
                   int maxWaitMs,
                   int fetchSize,
                   int maxPollRecords,
                   boolean checkCrcs,
                   Deserializer<K> keyDeserializer,
                   Deserializer<V> valueDeserializer,
                   Metadata metadata,
                   SubscriptionState subscriptions,
                   Metrics metrics,
                   FetcherMetricsRegistry metricsRegistry,
                   Time time,
                   long retryBackoffMs,
                   IsolationLevel isolationLevel); }
Fetcher implements SubscriptionState.Listener, Closeable { public void updateFetchPositions(Set<TopicPartition> partitions) { final Set<TopicPartition> needsOffsetReset = new HashSet<>(); for (TopicPartition tp : partitions) { if (!subscriptions.isAssigned(tp) || subscriptions.hasValidPosition(tp)) continue; if (subscriptions.isOffsetResetNeeded(tp)) { needsOffsetReset.add(tp); } else if (subscriptions.committed(tp) == null) { subscriptions.needOffsetReset(tp); needsOffsetReset.add(tp); } else { long committed = subscriptions.committed(tp).offset(); log.debug("Resetting offset for partition {} to the committed offset {}", tp, committed); subscriptions.seek(tp, committed); } } if (!needsOffsetReset.isEmpty()) { resetOffsets(needsOffsetReset); } } Fetcher(ConsumerNetworkClient client,
                   int minBytes,
                   int maxBytes,
                   int maxWaitMs,
                   int fetchSize,
                   int maxPollRecords,
                   boolean checkCrcs,
                   Deserializer<K> keyDeserializer,
                   Deserializer<V> valueDeserializer,
                   Metadata metadata,
                   SubscriptionState subscriptions,
                   Metrics metrics,
                   FetcherMetricsRegistry metricsRegistry,
                   Time time,
                   long retryBackoffMs,
                   IsolationLevel isolationLevel); boolean hasCompletedFetches(); int sendFetches(); void resetOffsetsIfNeeded(Set<TopicPartition> partitions); void updateFetchPositions(Set<TopicPartition> partitions); Map<String, List<PartitionInfo>> getAllTopicMetadata(long timeout); Map<String, List<PartitionInfo>> getTopicMetadata(MetadataRequest.Builder request, long timeout); Map<TopicPartition, OffsetAndTimestamp> getOffsetsByTimes(Map<TopicPartition, Long> timestampsToSearch,
                                                                     long timeout); Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> partitions, long timeout); Map<TopicPartition, Long> endOffsets(Collection<TopicPartition> partitions, long timeout); Map<TopicPartition, List<ConsumerRecord<K, V>>> fetchedRecords(); @Override void onAssignment(Set<TopicPartition> assignment); static Sensor throttleTimeSensor(Metrics metrics, FetcherMetricsRegistry metricsRegistry); @Override void close(); }
Fetcher implements SubscriptionState.Listener, Closeable { public void updateFetchPositions(Set<TopicPartition> partitions) { final Set<TopicPartition> needsOffsetReset = new HashSet<>(); for (TopicPartition tp : partitions) { if (!subscriptions.isAssigned(tp) || subscriptions.hasValidPosition(tp)) continue; if (subscriptions.isOffsetResetNeeded(tp)) { needsOffsetReset.add(tp); } else if (subscriptions.committed(tp) == null) { subscriptions.needOffsetReset(tp); needsOffsetReset.add(tp); } else { long committed = subscriptions.committed(tp).offset(); log.debug("Resetting offset for partition {} to the committed offset {}", tp, committed); subscriptions.seek(tp, committed); } } if (!needsOffsetReset.isEmpty()) { resetOffsets(needsOffsetReset); } } Fetcher(ConsumerNetworkClient client,
                   int minBytes,
                   int maxBytes,
                   int maxWaitMs,
                   int fetchSize,
                   int maxPollRecords,
                   boolean checkCrcs,
                   Deserializer<K> keyDeserializer,
                   Deserializer<V> valueDeserializer,
                   Metadata metadata,
                   SubscriptionState subscriptions,
                   Metrics metrics,
                   FetcherMetricsRegistry metricsRegistry,
                   Time time,
                   long retryBackoffMs,
                   IsolationLevel isolationLevel); boolean hasCompletedFetches(); int sendFetches(); void resetOffsetsIfNeeded(Set<TopicPartition> partitions); void updateFetchPositions(Set<TopicPartition> partitions); Map<String, List<PartitionInfo>> getAllTopicMetadata(long timeout); Map<String, List<PartitionInfo>> getTopicMetadata(MetadataRequest.Builder request, long timeout); Map<TopicPartition, OffsetAndTimestamp> getOffsetsByTimes(Map<TopicPartition, Long> timestampsToSearch,
                                                                     long timeout); Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> partitions, long timeout); Map<TopicPartition, Long> endOffsets(Collection<TopicPartition> partitions, long timeout); Map<TopicPartition, List<ConsumerRecord<K, V>>> fetchedRecords(); @Override void onAssignment(Set<TopicPartition> assignment); static Sensor throttleTimeSensor(Metrics metrics, FetcherMetricsRegistry metricsRegistry); @Override void close();  }