@Override
  public void register(SystemStreamPartition systemStreamPartition, String offset) {
    super.register(systemStreamPartition, offset);
    ssps.add(systemStreamPartition);
    setIsAtHead(systemStreamPartition, true);
  }
Kill all the threads, and shutdown.
samza-test/src/main/java/org/apache/samza/system/mock/MockSystemConsumer.java