8000 refactor(inkless:consume): drop four never-recorded fetch metrics by jeqo · Pull Request #787 · aiven/inkless · GitHub
[go: up one dir, main page]
Skip to content

refactor(inkless:consume): drop four never-recorded fetch metrics #787

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
giuseppelillo merged 2 commits into from
Sep 9, 2026
Merged

refactor(inkless:consume): drop four never-recorded fetch metrics #787

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
97a65db
refactor(inkless:consume): drop four never-recorded fetch metrics
jeqo Sep 7, 2026
04a88da
refactor(inkless:consume): tidy metrics javadoc and gauge lambda
jeqo Sep 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 0 additions & 4 deletions docs/inkless/metrics.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -131,11 +131,7 @@ Attribute name Description
CacheEntrySize Size of individual cache entries in bytes
CacheFetchErrorRate Rate of errors when fetching from the cache per second
CacheHitBytesPerSec Bytes served from object cache without hitting remote storage per second
CacheHitCount Rate of cache hits per second
CacheMissCount Rate of cache misses per second
CacheQueryTime Time spent querying the object cache in milliseconds
CacheSize Current number of entries in the object cache
CacheStoreTime Time spent storing entries in the object cache in milliseconds
DisklessBytesOutPerSec Total bytes served to consumers via the diskless fetch path per second
FetchBatchesPerPartitionCount Number of batches fetched per partition
FetchCompletionTime Time spent completing the fetch response assembly in milliseconds
Expand Down
46 changes: 3 additions & 43 deletions storage/inkless/src/main/java/io/aiven/inkless/consume/InklessFetchMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,14 +46,6 @@ public class InklessFetchMetrics {
private static final String FIND_BATCHES_TIME_DOC = "Time spent finding batch coordinates in the control plane in milliseconds";
private static final String FETCH_PLAN_TIME = "FetchPlanTime";
private static final String FETCH_PLAN_TIME_DOC = "Time spent creating the fetch plan in milliseconds";
private static final String CACHE_QUERY_TIME = "CacheQueryTime";
private static final String CACHE_QUERY_TIME_DOC = "Time spent querying the object cache in milliseconds";
private static final String CACHE_STORE_TIME = "CacheStoreTime";
private static final String CACHE_STORE_TIME_DOC = "Time spent storing entries in the object cache in milliseconds";
private static final String CACHE_HIT_COUNT = "CacheHitCount";
private static final String CACHE_HIT_COUNT_DOC = "Rate of cache hits per second";
private static final String CACHE_MISS_COUNT = "CacheMissCount";
private static final String CACHE_MISS_COUNT_DOC = "Rate of cache misses per second";
private static final String CACHE_ENTRY_SIZE = "CacheEntrySize";
private static final String CACHE_ENTRY_SIZE_DOC = "Size of individual cache entries in bytes";
private static final String CACHE_SIZE = "CacheSize";
Expand Down Expand Up @@ -133,10 +125,6 @@ public static List<MetricNameTemplate> all() {
new MetricNameTemplate(FETCH_TOTAL_TIME, GROUP, FETCH_TOTAL_TIME_DOC),
new MetricNameTemplate(FIND_BATCHES_TIME, GROUP, FIND_BATCHES_TIME_DOC),
new MetricNameTemplate(FETCH_PLAN_TIME, GROUP, FETCH_PLAN_TIME_DOC),
new MetricNameTemplate(CACHE_QUERY_TIME, GROUP, CACHE_QUERY_TIME_DOC),
new MetricNameTemplate(CACHE_STORE_TIME, GROUP, CACHE_STORE_TIME_DOC),
new MetricNameTemplate(CACHE_HIT_COUNT, GROUP, CACHE_HIT_COUNT_DOC),
new MetricNameTemplate(CACHE_MISS_COUNT, GROUP, CACHE_MISS_COUNT_DOC),
new MetricNameTemplate(CACHE_ENTRY_SIZE, GROUP, CACHE_ENTRY_SIZE_DOC),
new MetricNameTemplate(CACHE_SIZE, GROUP, CACHE_SIZE_DOC),
new MetricNameTemplate(FETCH_FIRST_BYTE_TIME, GROUP, FETCH_FIRST_BYTE_TIME_DOC),
Expand Down Expand Up @@ -176,12 +164,8 @@ public static List<MetricNameTemplate> all() {
private final Histogram fetchTimeHistogram;
private final Histogram findBatchesTimeHistogram;
private final Histogram fetchPlanTimeHistogram;
private final Histogram cacheQueryTimeHistogram;
private final Histogram cacheStoreTimeHistogram;
private final Histogram cacheEntrySize;
private final Gauge<Long> cacheSize;
private final Meter cacheHits;
private final Meter cacheMisses;
private final Histogram fetchFirstByteTimeHistogram;
private final Histogram fetchFileTimeHistogram;
private final Histogram fetchCompletionTimeHistogram;
Expand Down Expand Up @@ -221,10 +205,6 @@ public InklessFetchMetrics(final Time time, final ObjectCache cache, final Kafka
fetchTimeHistogram = metricsGroup.newHistogram(FETCH_TOTAL_TIME, true, Map.of());
findBatchesTimeHistogram = metricsGroup.newHistogram(FIND_BATCHES_TIME, true, Map.of());
fetchPlanTimeHistogram = metricsGroup.newHistogram(FETCH_PLAN_TIME, true, Map.of());
cacheQueryTimeHistogram = metricsGroup.newHistogram(CACHE_QUERY_TIME, true, Map.of());
cacheStoreTimeHistogram = metricsGroup.newHistogram(CACHE_STORE_TIME, true, Map.of());
cacheHits = metricsGroup.newMeter(CACHE_HIT_COUNT, "hits", TimeUnit.SECONDS, Map.of());
cacheMisses = metricsGroup.newMeter(CACHE_MISS_COUNT, "misses", TimeUnit.SECONDS, Map.of());
fetchFirstByteTimeHistogram = metricsGroup.newHistogram(FETCH_FIRST_BYTE_TIME, true, Map.of());
fetchFileTimeHistogram = metricsGroup.newHistogram(FETCH_FILE_TIME, true, Map.of());
fetchCompletionTimeHistogram = metricsGroup.newHistogram(FETCH_COMPLETION_TIME, true, Map.of());
Expand All @@ -242,7 +222,7 @@ public InklessFetchMetrics(final Time time, final ObjectCache cache, final Kafka
partitionCorruptRecordRate = metricsGroup.newMeter(PARTITION_CORRUPT_RECORD_RATE, "errors", TimeUnit.SECONDS, Map.of());
partitionControlPlaneErrorRate = metricsGroup.newMeter(PARTITION_CONTROL_PLANE_ERROR_RATE, "errors", TimeUnit.SECONDS, Map.of());
cacheEntrySize = metricsGroup.newHistogram(CACHE_ENTRY_SIZE, true, Map.of());
cacheSize = metricsGroup.newGauge(CACHE_SIZE, () -> cache.size());
cacheSize = metricsGroup.newGauge(CACHE_SIZE, cache::size);
recentDataRequestRate = metricsGroup.newMeter(RECENT_DATA_REQUEST_RATE, "requests", TimeUnit.SECONDS, Map.of());
laggingConsumerRequestRate = metricsGroup.newMeter(LAGGING_CONSUMER_REQUEST_RATE, "requests", TimeUnit.SECONDS, Map.of());
laggingConsumerRejectedRate = metricsGroup.newMeter(LAGGING_CONSUMER_REQUEST_REJECTED_RATE, "rejections", TimeUnit.SECONDS, Map.of());
Expand Down Expand Up @@ -270,22 +250,6 @@ public void fetchPlanFinished(final long durationMs) {
fetchPlanTimeHistogram.update(durationMs);
}

public void cacheQueryFinished(final long durationMs) {
cacheQueryTimeHistogram.update(durationMs);
}

public void cacheStoreFinished(final long durationMs) {
cacheStoreTimeHistogram.update(durationMs);
}

public void cacheHit(final boolean hit) {
if (hit) {
cacheHits.mark();
} else {
cacheMisses.mark();
}
}

public void cacheEntrySize(final int size) {
cacheEntrySize.update(size);
}
Expand Down Expand Up @@ -323,10 +287,6 @@ public void close() {
metricsGroup.removeMetric(FETCH_FIRST_BYTE_TIME);
metricsGroup.removeMetric(FETCH_FILE_TIME);
metricsGroup.removeMetric(FETCH_PLAN_TIME);
metricsGroup.removeMetric(CACHE_QUERY_TIME);
metricsGroup.removeMetric(CACHE_STORE_TIME);
metricsGroup.removeMetric(CACHE_HIT_COUNT);
metricsGroup.removeMetric(CACHE_MISS_COUNT);
metricsGroup.removeMetric(CACHE_SIZE);
metricsGroup.removeMetric(CACHE_ENTRY_SIZE);
metricsGroup.removeMetric(FIND_BATCHES_TIME);
Expand Down Expand Up @@ -440,11 +400,11 @@ public void recordLaggingConsumerRequest() {
* This typically corresponds to:
* - RejectedExecutionException: Queue full (AbortPolicy triggered)
*
* In this case, backpressure is applied: the consumer receives an error response
* <p>In this case, backpressure is applied: the consumer receives an error response
* and backs off via fetch purgatory.
* Metric: LaggingConsumerRejectedRate
*
* High rejection rate indicates:
* <p>High rejection rate indicates:
* - Sustained lagging consumer load exceeding capacity
* - May need to increase fetch.lagging.consumer.thread.pool.size
* - Or increase fetch.lagging.consumer.request.rate.limit
Expand Down
Loading
0