@Test public void testReadCommittedLagMetric() { Metrics metrics = new Metrics(); fetcher = createFetcher(subscriptions, metrics, new ByteArrayDeserializer(), new ByteArrayDeserializer(), Integer.MAX_VALUE, IsolationLevel.READ_COMMITTED); subscriptions.assignFromUser(singleton(tp1)); subscriptions.seek(tp1, 0); MetricName maxLagMetric = metrics.metricInstance(metricsRegistry.recordsLagMax); MetricName partitionLagMetric = metrics.metricName(tp1 + ".records-lag", metricGroup); Map<MetricName, KafkaMetric> allMetrics = metrics.metrics(); KafkaMetric recordsFetchLagMax = allMetrics.get(maxLagMetric); assertEquals(Double.NEGATIVE_INFINITY, recordsFetchLagMax.value(), EPSILON); fetchRecords(tp1, MemoryRecords.EMPTY, Errors.NONE, 100L, 50L, 0); assertEquals(50, recordsFetchLagMax.value(), EPSILON); KafkaMetric partitionLag = allMetrics.get(partitionLagMetric); assertEquals(50, partitionLag.value(), EPSILON); MemoryRecordsBuilder builder = MemoryRecords.builder(ByteBuffer.allocate(1024), CompressionType.NONE, TimestampType.CREATE_TIME, 0L); for (int v = 0; v < 3; v++) builder.appendWithOffset(v, RecordBatch.NO_TIMESTAMP, "key".getBytes(), ("value-" + v).getBytes()); fetchRecords(tp1, builder.build(), Errors.NONE, 200L, 150L, 0); assertEquals(147, recordsFetchLagMax.value(), EPSILON); assertEquals(147, partitionLag.value(), EPSILON); subscriptions.unsubscribe(); assertFalse(allMetrics.containsKey(partitionLagMetric)); }
private List<ConsumerRecord<K, V>> fetchRecords(PartitionRecords partitionRecords, int maxRecords) { if (!subscriptions.isAssigned(partitionRecords.partition)) { log.debug("Not returning fetched records for partition {} since it is no longer assigned", partitionRecords.partition); } else { long position = subscriptions.position(partitionRecords.partition); if (!subscriptions.isFetchable(partitionRecords.partition)) { log.debug("Not returning fetched records for assigned partition {} since it is no longer fetchable", partitionRecords.partition); } else if (partitionRecords.nextFetchOffset == position) { List<ConsumerRecord<K, V>> partRecords = partitionRecords.fetchRecords(maxRecords); long nextOffset = partitionRecords.nextFetchOffset; log.trace("Returning fetched records at offset {} for assigned partition {} and update " + "position to {}", position, partitionRecords.partition, nextOffset); subscriptions.position(partitionRecords.partition, nextOffset); Long partitionLag = subscriptions.partitionLag(partitionRecords.partition, isolationLevel); if (partitionLag != null) this.sensors.recordPartitionLag(partitionRecords.partition, partitionLag); return partRecords; } else { log.debug("Ignoring fetched records for {} at offset {} since the current position is {}", partitionRecords.partition, partitionRecords.nextFetchOffset, position); } } partitionRecords.drain(); return emptyList(); }
Fetcher implements SubscriptionState.Listener, Closeable { private List<ConsumerRecord<K, V>> fetchRecords(PartitionRecords partitionRecords, int maxRecords) { if (!subscriptions.isAssigned(partitionRecords.partition)) { log.debug("Not returning fetched records for partition {} since it is no longer assigned", partitionRecords.partition); } else { long position = subscriptions.position(partitionRecords.partition); if (!subscriptions.isFetchable(partitionRecords.partition)) { log.debug("Not returning fetched records for assigned partition {} since it is no longer fetchable", partitionRecords.partition); } else if (partitionRecords.nextFetchOffset == position) { List<ConsumerRecord<K, V>> partRecords = partitionRecords.fetchRecords(maxRecords); long nextOffset = partitionRecords.nextFetchOffset; log.trace("Returning fetched records at offset {} for assigned partition {} and update " + "position to {}", position, partitionRecords.partition, nextOffset); subscriptions.position(partitionRecords.partition, nextOffset); Long partitionLag = subscriptions.partitionLag(partitionRecords.partition, isolationLevel); if (partitionLag != null) this.sensors.recordPartitionLag(partitionRecords.partition, partitionLag); return partRecords; } else { log.debug("Ignoring fetched records for {} at offset {} since the current position is {}", partitionRecords.partition, partitionRecords.nextFetchOffset, position); } } partitionRecords.drain(); return emptyList(); } }
Fetcher implements SubscriptionState.Listener, Closeable { private List<ConsumerRecord<K, V>> fetchRecords(PartitionRecords partitionRecords, int maxRecords) { if (!subscriptions.isAssigned(partitionRecords.partition)) { log.debug("Not returning fetched records for partition {} since it is no longer assigned", partitionRecords.partition); } else { long position = subscriptions.position(partitionRecords.partition); if (!subscriptions.isFetchable(partitionRecords.partition)) { log.debug("Not returning fetched records for assigned partition {} since it is no longer fetchable", partitionRecords.partition); } else if (partitionRecords.nextFetchOffset == position) { List<ConsumerRecord<K, V>> partRecords = partitionRecords.fetchRecords(maxRecords); long nextOffset = partitionRecords.nextFetchOffset; log.trace("Returning fetched records at offset {} for assigned partition {} and update " + "position to {}", position, partitionRecords.partition, nextOffset); subscriptions.position(partitionRecords.partition, nextOffset); Long partitionLag = subscriptions.partitionLag(partitionRecords.partition, isolationLevel); if (partitionLag != null) this.sensors.recordPartitionLag(partitionRecords.partition, partitionLag); return partRecords; } else { log.debug("Ignoring fetched records for {} at offset {} since the current position is {}", partitionRecords.partition, partitionRecords.nextFetchOffset, position); } } partitionRecords.drain(); return emptyList(); } 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 { private List<ConsumerRecord<K, V>> fetchRecords(PartitionRecords partitionRecords, int maxRecords) { if (!subscriptions.isAssigned(partitionRecords.partition)) { log.debug("Not returning fetched records for partition {} since it is no longer assigned", partitionRecords.partition); } else { long position = subscriptions.position(partitionRecords.partition); if (!subscriptions.isFetchable(partitionRecords.partition)) { log.debug("Not returning fetched records for assigned partition {} since it is no longer fetchable", partitionRecords.partition); } else if (partitionRecords.nextFetchOffset == position) { List<ConsumerRecord<K, V>> partRecords = partitionRecords.fetchRecords(maxRecords); long nextOffset = partitionRecords.nextFetchOffset; log.trace("Returning fetched records at offset {} for assigned partition {} and update " + "position to {}", position, partitionRecords.partition, nextOffset); subscriptions.position(partitionRecords.partition, nextOffset); Long partitionLag = subscriptions.partitionLag(partitionRecords.partition, isolationLevel); if (partitionLag != null) this.sensors.recordPartitionLag(partitionRecords.partition, partitionLag); return partRecords; } else { log.debug("Ignoring fetched records for {} at offset {} since the current position is {}", partitionRecords.partition, partitionRecords.nextFetchOffset, position); } } partitionRecords.drain(); return emptyList(); } 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 { private List<ConsumerRecord<K, V>> fetchRecords(PartitionRecords partitionRecords, int maxRecords) { if (!subscriptions.isAssigned(partitionRecords.partition)) { log.debug("Not returning fetched records for partition {} since it is no longer assigned", partitionRecords.partition); } else { long position = subscriptions.position(partitionRecords.partition); if (!subscriptions.isFetchable(partitionRecords.partition)) { log.debug("Not returning fetched records for assigned partition {} since it is no longer fetchable", partitionRecords.partition); } else if (partitionRecords.nextFetchOffset == position) { List<ConsumerRecord<K, V>> partRecords = partitionRecords.fetchRecords(maxRecords); long nextOffset = partitionRecords.nextFetchOffset; log.trace("Returning fetched records at offset {} for assigned partition {} and update " + "position to {}", position, partitionRecords.partition, nextOffset); subscriptions.position(partitionRecords.partition, nextOffset); Long partitionLag = subscriptions.partitionLag(partitionRecords.partition, isolationLevel); if (partitionLag != null) this.sensors.recordPartitionLag(partitionRecords.partition, partitionLag); return partRecords; } else { log.debug("Ignoring fetched records for {} at offset {} since the current position is {}", partitionRecords.partition, partitionRecords.nextFetchOffset, position); } } partitionRecords.drain(); return emptyList(); } 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();  }