public SystemProducer getProducer(String systemName, Config config, MetricsRegistry registry) {
    return new SystemProducer() {
      private final Random seed = new Random(System.currentTimeMillis());

      @Override
      public void start() {

      }

      @Override
      public void stop() {

      }

      @Override
      public void register(String source) {
      }

      @Override
      public void send(String source, OutgoingMessageEnvelope envelope) {
        SystemStream systemStream = envelope.getSystemStream();
        List<SystemStreamPartition> sspForSystem = MSG_QUEUES.keySet().stream()
            .filter(ssp -> ssp.getSystemStream().equals(systemStream))
            .collect(ArrayList::new, (l, ssp) -> l.add(ssp), (l1, l2) -> l1.addAll(l2));
        if (sspForSystem.isEmpty()) {
          MSG_QUEUES.putIfAbsent(new SystemStreamPartition(systemStream, new Partition(0)), new ArrayList<>());
          sspForSystem.add(new SystemStreamPartition(systemStream, new Partition(0)));
        }
        int partitionCount = sspForSystem.size();
        int partitionId = envelope.getPartitionKey() == null ?
            envelope.getKey() == null ? this.seed.nextInt(partitionCount) : envelope.getKey().hashCode() % partitionCount :
            envelope.getPartitionKey().hashCode() % partitionCount;
        SystemStreamPartition ssp = new SystemStreamPartition(envelope.getSystemStream(), new Partition(partitionId));
        List<IncomingMessageEnvelope> msgQueue = MSG_QUEUES.get(ssp);
        msgQueue.add(new IncomingMessageEnvelope(ssp, null, envelope.getKey(), envelope.getMessage()));
      }

      @Override
      public void flush(String source) {

      }
    };
  }
A mock system backed by a set of in-memory queues. Used for testing w/o actual external messaging systems.
samza-core/src/test/java/org/apache/samza/system/MockSystemFactory.java