xiangfu0 commented on code in PR #19739:
URL: https://github.com/apache/pinot/pull/19739#discussion_r4290535378
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1915,7 +1938,8 @@ && new
LLCSegmentName(segmentEntry.getKey()).getPartitionGroupId() == partitionI
break;
}
}
- if (previousConsumingSegment == null) {
+ // A segment created for an OFFLINE, ONLINE or new partition has no
previous CONSUMING segment
Review Comment:
Nice catch on keeping the data-loss signal honest here. Just to confirm my
reading: for a segment created by an earlier attempt that replaces a CONSUMING
one, this still fires if that predecessor flipped away from CONSUMING between
attempts, and stays quiet for OFFLINE/ONLINE/new-partition segments that
legitimately have no predecessor. The condition is a bit dense, but the comment
above it carries the intent well.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1448,6 +1448,10 @@ public void ensureAllPartitionsConsuming(TableConfig
tableConfig, List<StreamCon
Preconditions.checkState(!_isStopping, "Segment manager is stopping");
String realtimeTableName = tableConfig.getTableName();
+ // The updater below runs again when the IdealState write fails (e.g. on a
version conflict), while the segment ZK
+ // metadata created by the failed attempt remains. Track the created
segments so that the next attempt still adds
+ // them to the IdealState.
+ Map<String, Boolean> newSegmentsCreated = new ConcurrentHashMap<>();
Review Comment:
Good call using a ConcurrentHashMap here rather than a plain map. The
updater can be run by a different leader thread across the group-commit
batching and retries, so sharing the map across attempts needs the thread-safe
type. The @VisibleForTesting overload passing a HashMap is fine since the tests
drive it single-threaded.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1448,6 +1448,10 @@ public void ensureAllPartitionsConsuming(TableConfig
tableConfig, List<StreamCon
Preconditions.checkState(!_isStopping, "Segment manager is stopping");
String realtimeTableName = tableConfig.getTableName();
+ // The updater below runs again when the IdealState write fails (e.g. on a
version conflict), while the segment ZK
+ // metadata created by the failed attempt remains. Track the created
segments so that the next attempt still adds
+ // them to the IdealState.
+ Map<String, Boolean> newSegmentsCreated = new ConcurrentHashMap<>();
Review Comment:
Good call using a ConcurrentHashMap here rather than a plain map. The
updater can be run by a different leader thread across the group-commit
batching and retries, so sharing the map across attempts needs the thread-safe
type. The @VisibleForTesting overload passing a HashMap is fine since the tests
drive it single-threaded.
--
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]
