suvodeep-pyne commented on code in PR #20082:
URL: https://github.com/apache/pinot/pull/20082#discussion_r4408941469
##########
pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/KinesisConsumer.java:
##########
@@ -54,15 +57,29 @@ public class KinesisConsumer extends
KinesisConnectionHandler implements Partiti
private static final int RATE_LIMIT_BACKOFF_JITTER_BOUND_MS = 250;
private static final RequestRateLimiter SHARED_REQUEST_RATE_LIMITER = new
SharedKinesisRequestRateLimiter();
private static final double RATE_LIMIT_LOG_RATE_PER_MIN = 5.0;
+ // Exhausted throttling returns an empty batch, which the consume loop
counts as a successful fetch. Like the consume
+ // loop's own tolerance for fetch errors, this many in a row are tolerated;
each further one is reported.
+ private static final int MAX_CONSECUTIVE_THROTTLED_FETCHES = 5;
+ // Fetches in a row that throttling exhausted, by consumer and shard. Kept
outside the consumer instance: every empty
+ // batch counts toward the consume loop's idle timeout, which recreates the
consumer under the same client id.
+ private static final Map<String, Integer> CONSECUTIVE_THROTTLED_FETCHES =
new ConcurrentHashMap<>();
Review Comment:
Fixed in 3ea93d223e: the throttle-count map is now a Guava cache view whose
entries expire after an hour without access, like the shared rate limiters in
the same file. `merge` and `remove` stay atomic. With your repro shape, run
through the real metadata fetcher and end-of-shard check, 100 throttled checks
under unique client ids left 100 entries before. Now they expire once idle, and
the consume loop's counting is unchanged: nothing for fetches 1 to 5, a report
from the 6th, a reset by a returned fetch, and the count kept across recreation.
##########
pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarPartitionLevelConsumer.java:
##########
@@ -93,6 +118,20 @@ public synchronized PulsarMessageBatch
fetchMessages(StreamPartitionMsgOffset st
return new PulsarMessageBatch(messages, offsetOfNextBatch,
_reader.hasReachedEndOfTopic(), batchSizeInBytes);
}
+ /// Throws a transient exception once the reader has stayed disconnected for
[#MAX_DISCONNECTED_MS], so that the
+ /// consume loop recreates it whether or not idle recreation is enabled.
+ private void checkConnected(long nowMs) {
+ if (_reader.isConnected()) {
+ _disconnectedSinceMs = -1;
+ } else if (_disconnectedSinceMs < 0) {
+ _disconnectedSinceMs = nowMs;
+ } else if (nowMs - _disconnectedSinceMs >= MAX_DISCONNECTED_MS) {
+ throw new TransientConsumerException(new TimeoutException(
Review Comment:
Intended. This gives lagging readers the lifecycle that caught-up readers
already have: after 30 s disconnected, Pinot recreates the reader, and if that
creation also fails, consumption stops until repair. We accept that
availability trade-off here, in exchange for no longer hanging forever on a
refusal the client only retries in the background.
Added `droppedLaggingReaderThatReconnectsReadsOn` in 3ea93d223e. It unloads
the topic while the grant is intact. The broker log shows the reader
resubscribing about 100 ms later with 4,001 messages still to read, and the
reader reads all 5,000 messages without an exception. A mutant that throws on
the first disconnected fetch fails this test.
Recovery through longer outages could be a separate change: configurable
timing, or retries on recreation that still report an explicit auth refusal at
once.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
