Jackie-Jiang commented on code in PR #19370:
URL: https://github.com/apache/pinot/pull/19370#discussion_r4449003592
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -2266,17 +2307,26 @@ private void makeStreamConsumer(String reason) {
}
}
+ private void closeQuietly(PartitionGroupConsumer consumer) {
+ try {
+ consumer.close();
+ } catch (Exception e) {
+ _segmentLogger.warn("Could not close stream consumer", e);
+ }
+ }
+
/// Checkpoints existing consumer before creating a new consumer instance
/// Assumes there is a valid instance of [PartitionGroupConsumer]
private void recreateStreamConsumer(String reason) {
_segmentLogger.info("Recreating stream consumer for topic partition {},
reason: {}", _clientId, reason);
- _currentOffset = _partitionGroupConsumer.checkpoint(_currentOffset);
+ _currentOffset = _partitionGroupConsumer.get().checkpoint(_currentOffset);
closePartitionGroupConsumer();
try {
- _partitionGroupConsumer =
+ PartitionGroupConsumer consumer =
_streamConsumerFactory.createPartitionGroupConsumer(_clientId,
_partitionGroupConsumptionStatus,
CONSUMER_RECREATE_RETRY_POLICY);
- _partitionGroupConsumer.start(_currentOffset);
+ consumer.start(_currentOffset);
+ _partitionGroupConsumer.set(consumer);
Review Comment:
[P2] Close the replacement consumer when start() fails
The replacement is now kept only in a local variable until start() succeeds.
If a stream plugin overrides start() and throws while initializing its
resources, the catch block rethrows and loses the replacement;
_partitionGroupConsumer still refers to the old, already-closed consumer, so
subsequent offload cannot close the new one. Previously recreation assigned the
replacement before calling start(), preserving it for cleanup. Please
explicitly close the newly created consumer on startup failure, and apply the
same cleanup to makeStreamConsumer(). Stock Kafka, Pulsar, and Kinesis use the
default no-op start(); this affects plugins with a failing startup
implementation.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -820,6 +824,22 @@ public void run() {
_segmentLogger.info("Starting consumption on segment: {}, maxRowCount:
{}, maxEndTime: {}.", _llcSegmentName,
_segmentMaxRowCount, new DateTime(_consumeEndTime,
DateTimeZone.UTC));
+ // Create the stream consumer here (consumer thread) with
CONSUMER_RECREATE_RETRY_POLICY. Doing this in
+ // the Helix constructor would either block the state-transition
thread for the long retry window or keep
+ // the short Kafka 5x2s path and mark the replica OFFLINE. See
+ // https://github.com/apache/pinot/issues/11314 and
https://github.com/apache/pinot/pull/17062.
+ if (_shouldStop) {
+ return;
+ }
+ makeStreamConsumer("Starting");
+ if (_shouldStop) {
+ // Leave a successfully created consumer in place so CONSUMING ->
ONLINE can catch up.
+ return;
Review Comment:
[P1] Complete upsert initialization before allowing catchup
If the ONLINE transition calls stop() after makeStreamConsumer() publishes
the consumer but before this check, the consumer thread returns without running
the snapshot and removeExpiredPrimaryKeys() block below. After joining it,
goOnlineFromConsuming() sees the acquired semaphore and retained consumer and
calls catchupToFinalOffset(), which enters consumeLoop() directly. For full
upsert with metadataTTL=20, consider a previous segment containing A@10 and
B@100, followed by A@110 in catchup: normal initialization expires A's map
entry while preserving the historical A@10 row, but this path retains the entry
and invalidates A@10 when A@110 is ingested. The new segment's contents can
still match the winning replica, so its CRC check does not detect the changed
valid-doc bitmap in the previous segment. Please complete the metadata
initialization before permitting catchup, or fall back to download when
initialization is incomplete. The stop-during-init test should cover this with
TTL
upsert state and actual catchup.
--
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]
