xiangfu0 commented on code in PR #20082:
URL: https://github.com/apache/pinot/pull/20082#discussion_r4407732109
##########
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:
Design point for an owner. Once a lagging reader has been disconnected for
30 s, this exception sends it to `recreateStreamConsumer`. The Pulsar factory
ignores the retry policy, so the recreate is a single `create()`, limited by
the 30 s operation timeout. If the broker is still down at that point (an
outage of more than about 60 s), `consumeLoop` throws and the segment stops
consuming until it is repaired. Before this change, a lagging reader resumed by
itself when the broker came back.
Caught-up readers already behave this way, so this may be acceptable, but
please confirm it is intended. Please also add a test for a reader that the
broker drops and that reconnects inside the window, for example an unload with
the grant intact, so the path without an auth failure is covered.
##########
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:
Minor: entries leak in more cases than the PR's "known limit" describes. On
the controller, `KinesisStreamMetadataProvider.consumedEndOfShard` creates this
consumer with the provider's client id, and `PartitionGroupMetadataFetcher`
makes that id unique for each call (`getUniqueClientId`). So an end-of-shard
check whose fetches are all throttled leaves a key that is never seen again. It
also stops at 5 attempts, so it never reports. A throwaway repro of 100 such
calls left 100 entries.
Consider expiring idle entries, for example with a cache using
`expireAfterAccess`, or not counting for consumers that the consume loop didn't
create.
--
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]
