query/src/main/java/org/infinispan/query/affinity/FixedShardsDistribution.java
/**
    * Associates segments to each shard.
    *
    * @param shardsNumPerServer numbers of shards allocated for each server
    * @param segmentsPerServer  the primary owned segments of each server
    * @param nodes              the members of the cluster
    */
Associates segments to each shard.
@param shardsNumPerServer numbers of shards allocated for each server
@param segmentsPerServer  the primary owned segments of each server
@param nodes              the members of the cluster
Associates segments to each shard . @param shardsNumPerServer numbers of shards allocated for each server @param segmentsPerServer the primary owned segments of each server @param nodes the members of the cluster
private void populateSegments(int[] shardsNumPerServer, List<Set<Integer>> segmentsPerServer, List<Address> nodes) {
      int shardId = 0;
      int n = 0;
      Set<Integer> remainingSegments = new HashSet<>();
      for (Address node : nodes) {
         Collection<Integer> primarySegments = segmentsPerServer.get(n);
         int shardQuantity = shardsNumPerServer[n];
         if (shardQuantity == 0) {
            remainingSegments.addAll(segmentsPerServer.get(n++));
            continue;
         }
         shardsPerAddressMap.computeIfAbsent(node, a -> new HashSet<>(shardQuantity));
         List<Set<Integer>> segments = this.split(primarySegments, shardsNumPerServer[n++]);
         for (Collection<Integer> shardSegments : segments) {
            String id = String.valueOf(shardId++);
            shardSegments.forEach(seg -> shardPerSegmentMap.put(seg, id));
            shardsPerAddressMap.get(node).add(id);
            addressPerShardMap.put(id, node);
         }
      }
      if (!remainingSegments.isEmpty()) {
         Iterator<String> shardIterator = Stream.iterate(0, i -> (i + 1) % numShards).map(String::valueOf).iterator();
         for (Integer segment : remainingSegments) {
            shardPerSegmentMap.put(segment, shardIterator.next());
         }
      }
   }
private void populateSegments ( int [ ] shardsNumPerServer , List < Set < Integer > > segmentsPerServer , List < Address > nodes ) { int shardId = 0 ; int n = 0 ; Set < Integer > remainingSegments = new HashSet < > ( ) ; for ( Address node : nodes ) { Collection < Integer > primarySegments = segmentsPerServer . get ( n ) ; int shardQuantity = shardsNumPerServer [ n ] ; if ( shardQuantity == 0 ) { remainingSegments . addAll ( segmentsPerServer . get ( n ++ ) ) ; continue ; } shardsPerAddressMap . computeIfAbsent ( node , a -> new HashSet < > ( shardQuantity ) ) ; List < Set < Integer > > segments = this . split ( primarySegments , shardsNumPerServer [ n ++ ] ) ; for ( Collection < Integer > shardSegments : segments ) { String id = String . valueOf ( shardId ++ ) ; shardSegments . forEach ( seg -> shardPerSegmentMap . put ( seg , id ) ) ; shardsPerAddressMap . get ( node ) . add ( id ) ; addressPerShardMap . put ( id , node ) ; } } if ( ! remainingSegments . isEmpty ( ) ) { Iterator < String > shardIterator = Stream . iterate ( 0 , i -> ( i + 1 ) % numShards ) . map ( String :: valueOf ) . iterator ( ) ; for ( Integer segment : remainingSegments ) { shardPerSegmentMap . put ( segment , shardIterator . next ( ) ) ; } } }
Associates segments to each shard.
Associates segments to each shard .