@Override
    public void process(IncomingMessageEnvelope incomingMessageEnvelope, MessageCollector messageCollector,
        TaskCoordinator taskCoordinator)
        throws Exception {

      Object message = incomingMessageEnvelope.getMessage();

      String key = new String((byte[]) incomingMessageEnvelope.getKey());
      Integer val = Integer.valueOf((String) message);

      LOG.info("Stream processor " + processorId + ";key=" + key + ";offset=" + incomingMessageEnvelope.getOffset()
          + "; totalRcvd=" + processedMessageCount + ";val=" + val + "; ssp=" + incomingMessageEnvelope
          .getSystemStreamPartition());

      // inject a failure
      if (val >= BAD_MESSAGE_KEY && processorId.equals(processorIdToFail)) {
        LOG.info("process method failing for msg=" + message);
        throw new Exception("Processing in the processor " + processorId + " failed ");
      }

      messageCollector.send(new OutgoingMessageEnvelope(new SystemStream(outputSystem, outputTopic), message));
      processedMessageCount++;

      synchronized (endLatch) {
        if (Integer.valueOf(key) < BAD_MESSAGE_KEY) {
          endLatch.countDown();
        }
      }
    }
Consumes data from the topic until there are no new messages for a while
and asserts that the number of consumed messages is as expected.
samza-test/src/test/java/org/apache/samza/processor/TestZkStreamProcessorBase.java