private static List<Integer> repartitionStreamlet(int incomingInteger, int numPartitions) {
    List<Integer> partitions;

    if (incomingInteger >= 0 && incomingInteger < 25) {
      partitions = Arrays.asList(0, 1);
    } else if (incomingInteger > 26 && incomingInteger < 50) {
      partitions = Arrays.asList(2, 3);
    } else if (incomingInteger > 50 && incomingInteger < 75) {
      partitions = Arrays.asList(4, 5);
    } else if (incomingInteger > 76 && incomingInteger <= 100) {
      partitions = Arrays.asList(6, 7);
    } else {
      partitions = Arrays.asList(ThreadLocalRandom.current().nextInt(0, 8));
    }

    String logMessage = String.format("Sending value %d to partitions: %s",
        incomingInteger,
        StreamletUtils.intListAsString(partitions));

    LOG.info(logMessage);

    return partitions;
  }
private static List < Integer > repartitionStreamlet ( int incomingInteger , int numPartitions ) { List < Integer > partitions ; if ( incomingInteger >= 0 && incomingInteger < 25 ) { partitions = Arrays . asList ( 0 , 1 ) ; } else if ( incomingInteger > 26 && incomingInteger < 50 ) { partitions = Arrays . asList ( 2 , 3 ) ; } else if ( incomingInteger > 50 && incomingInteger < 75 ) { partitions = Arrays . asList ( 4 , 5 ) ; } else if ( incomingInteger > 76 && incomingInteger <= 100 ) { partitions = Arrays . asList ( 6 , 7 ) ; } else { partitions = Arrays . asList ( ThreadLocalRandom . current ( ) . nextInt ( 0 , 8 ) ) ; } String logMessage = String . format ( "Sending value %d to partitions: %s" , incomingInteger , StreamletUtils . intListAsString ( partitions ) ) ; LOG . info ( logMessage ) ; return partitions ; }
The repartition function that determines to which partition each incoming
streamlet element is routed (across 8 possible partitions). Integers between 1
and 24 are routed to partitions 0 and 1, integers between 25 and 40 to partitions
2 and 3, and so on.