void send(ControlMessage message, SystemStream systemStream, MessageCollector collector) {
    int partitionCount = getPartitionCount(systemStream);
    // We pick a partition based on topic hashcode to aggregate the control messages from upstream tasks
    // After aggregation the task will broadcast the results to other partitions
    int aggregatePartition = systemStream.getStream().hashCode() % partitionCount;

    LOG.debug(String.format("Send %s message from task %s to %s partition %s for aggregation",
        MessageType.of(message).name(), message.getTaskName(), systemStream, aggregatePartition));

    OutgoingMessageEnvelope envelopeOut = new OutgoingMessageEnvelope(systemStream, aggregatePartition, null, message);
    collector.send(envelopeOut);
  }
This is a helper class to send control messages to an intermediate stream
samza-core/src/main/java/org/apache/samza/operators/impl/ControlMessageSender.java