hey friends I wonder if anyone has issues like this, we have a table and we want to ingest data from the beginning of the topic, however in the cluster while we are ingesting this data the servers get really really busy and the p99 response metrics for other tables get impacted greatly, has anyone gotten across this, do you all know what the bottleneck is and why servers get so impacted in terms of response times? it’s really weird that a change adding one table will impact the cluster so negatively.

it’s likely CPU. Pinot tries to ingest as many events as possible per second

right but we are at 32 cores and usually they are chilling

there is a

topic.consumption.rate.limit

from https://docs.pinot.apache.org/basics/data-import/pinot-stream-ingestion you can use. we’ve been meaning to experiment with this, but I can’t say for certain how well it works yet

omg

the description for that config also mentions there’s tons of GC with high ingestion, so maybe check those metrics as well? I know we had to disallow “smallest” offset ingestion for our tables to avoid this same latency impact

we just did smallest cause we need the data from the beginning of the topic indeed and wow yea GC pauses I need to check that too

what is that metric? i thought GC metrics were emitted as milliseconds

it’s % of CPU time on GC based on what i see lol

jvm_gc_collection_seconds_sum

sounds like seconds spent doing GC no? ~.3 sounds like 300ms which correlates with the p99 you’re seeing

thank you this is very insightful

as @Johan Adami mentioned limiting it on the table level using

topic.consumption.rate.limit

is the right way for this use case. it works quite well. • this is also a dynamic config - e.g. no need to reload or reset table for it to take effect. although it will not take effect until the next segment start consuming • it doesn't limit rate on a server level. e.g. it doesn't enforce limit across all consumers within the same server. • it only limits on a per partition level for lower-level consumer

do you all know how to establish the rate?

topic.consumption.rate.limit

like i know it’s a double but is it like message per second or something like that it should process? @Rong R @Johan Adami

double topicRateLimit = streamConfig.getTopicConsumptionRateLimit().get();
    double partitionRateLimit = topicRateLimit / partitionCount;
    <http://LOGGER.info|LOGGER.info>("A consumption rate limiter is set up for topic {} in table {} with rate limit: {} "
            + "(topic rate limit: {}, partition count: {})", streamConfig.getTopicName(), tableName, partitionRateLimit,
        topicRateLimit, partitionCount);
    MetricEmitter metricEmitter = new MetricEmitter(serverMetrics, metricKeyName);
    return new RateLimiterImpl(partitionRateLimit, metricEmitter);

It seems it naively assumes equal traffic per partition. so if you set 1000 as the limit on a topic with 10 partitions, each consumer will get 100 as the rate limit

that rate limit is messages/s (?)

correct

thank you Johan you the MVP

just some links so you or others can double check. here is where the rate limiter is created. here is where it’s created in the stream processing code. here is where the rate limiting is being done

it works like a charm