@Override
    public void run() {
      try {
        while (!Thread.interrupted() && ssps.size() > 0) {
          Set<SystemStreamPartition> sspsToFetch = new HashSet<SystemStreamPartition>();

          // Only fetch messages when there are no outstanding messages left.
          for (SystemStreamPartition ssp : ssps) {
            if (getNumMessagesInQueue(ssp) <= 0) {
              sspsToFetch.add(ssp);
            }
          }

          // Simulate a broker fetch request's network latency.
          Thread.sleep(brokerSleepMs);

          // Add messages to the BlockingEnvelopeMap.
          for (SystemStreamPartition ssp : sspsToFetch) {
            for (int i = 0; i < messagesPerBatch; ++i) {
              put(ssp, new IncomingMessageEnvelope(ssp, "0", "key", "value"));
            }
          }
        }
      } catch (InterruptedException e) {
        System.out.println("Got interrupt. Shutting down.");
      }
    }
The worker thread for MockSystemConsumer that simulates reading messages
from a remote streaming system (i.e. Kafka), and writing them to the
BlockingEnvelopeMap.
samza-test/src/main/java/org/apache/samza/system/mock/MockSystemConsumer.java