xiangfu0 commented on code in PR #18744:
URL: https://github.com/apache/pinot/pull/18744#discussion_r3431437312
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManager.java:
##########
@@ -346,12 +377,55 @@ private void computeTablePartitionInfo() {
private static class SegmentInfo {
int _partitionId;
long _creationTimeMs;
- List<String> _onlineServers;
+ List<String> _onlineServers = Collections.emptyList();
+ /// Number of target servers (servers with ONLINE/CONSUMING state in the
ideal state). The target servers are
+ /// tracked as count and order-independent hash instead of the full server
set to reduce the memory footprint.
+ int _numTargetServers;
+ /// Order-independent hash of the target servers, used together with the
count to detect ideal assignment changes.
+ /// A hash collision can only delay the reset of the replication state,
which is benign.
+ int _targetServersHash;
+ /// Time when the target servers were last changed, used to bound how long
the segment can stay pending
+ long _targetServersUpdateTimeMs;
+ /// Whether the segment has been observed fully replicated (online servers
covering the target servers from the
+ /// ideal state) at least once since its target servers last changed. Once
fully replicated, losing a replica
+ /// reduces the fully replicated servers of the partition (queries are
routed to the remaining replicas); before
+ /// that, the segment is excluded from the partition info to not reduce
the fully replicated servers while its
+ /// replicas are still loading.
+ boolean _fullyReplicatedOnce;
- SegmentInfo(int partitionId, long creationTimeMs, List<String>
onlineServers) {
+ SegmentInfo(int partitionId, long creationTimeMs) {
_partitionId = partitionId;
_creationTimeMs = creationTimeMs;
+ }
+
+ void updateServers(List<String> onlineServers, @Nullable Map<String,
String> targetInstanceStateMap,
+ long currentTimeMs) {
_onlineServers = onlineServers;
+ int numTargetServers = 0;
+ int targetServersHash = 0;
+ boolean coversTargetServers = true;
+ if (targetInstanceStateMap != null) {
+ for (Map.Entry<String, String> entry :
targetInstanceStateMap.entrySet()) {
+ if (isServingState(entry.getValue())) {
+ String server = entry.getKey();
+ numTargetServers++;
+ targetServersHash += server.hashCode();
Review Comment:
This target-set fingerprint is not unique enough for assignment changes. For
example, `{Server_localhost_8000, Server_localhost_8003}` and
`{Server_localhost_8001, Server_localhost_8002}` have the same count and summed
`String.hashCode()`, so a rebalance between them will not reset
`_fullyReplicatedOnce`. That lets a moved segment reduce the fully replicated
server set immediately and can reintroduce the partition routing failures this
PR is fixing. Please track the actual target server set, or use a
collision-resistant fingerprint, before treating the assignment as unchanged.
--
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]
