@Override public void start(){
  for (int i=0; i < threadCount; ++i) {
    Set<SystemStreamPartition> threadSsps=new HashSet<SystemStreamPartition>();
    for (    SystemStreamPartition ssp : ssps) {
      if (Math.abs(ssp.hashCode()) % threadCount == i) {
        threadSsps.add(ssp);
      }
    }
    Thread thread=new Thread(new MockSystemConsumerRunnable(threadSsps),"MockSystemConsumer-" + i);
    thread.setDaemon(true);
    threads.add(thread);
    thread.start();
  }
}

@ Override public void start ( ) { for ( int i = 0 ; i < threadCount ; ++ i ) { Set < SystemStreamPartition > threadSsps = new HashSet < SystemStreamPartition > ( ) ; for ( SystemStreamPartition ssp : ssps ) { if ( Math . abs ( ssp . hashCode ( ) ) % threadCount == i ) { threadSsps . add ( ssp ) ; } } Thread thread = new Thread ( new MockSystemConsumerRunnable ( threadSsps ) , "MockSystemConsumer-" + i ) ; thread . setDaemon ( true ) ; threads . add ( thread ) ; thread . start ( ) ; } }
assign systemstreampartitions to all of the threads , and start them up to begin simulating consuming messages .
