@Override
    public void receive(ComplexEvent complexEvent) {

        if (partitionExecutors.size() == 0) {
            ComplexEventChunk<ComplexEvent> outputEventChunk = new ComplexEventChunk<ComplexEvent>();
            ComplexEvent aComplexEvent = complexEvent;
            while (aComplexEvent != null) {
                StreamEvent newEvent = streamEventFactory.newInstance();
                streamEventConverter.convertComplexEvent(aComplexEvent, newEvent);
                outputEventChunk.add(newEvent);
                aComplexEvent = aComplexEvent.getNext();
            }
            send(outputEventChunk.getFirst());
        } else {
            if (complexEvent.getNext() == null) {
                for (PartitionExecutor partitionExecutor : partitionExecutors) {
                    StreamEvent newEvent = streamEventFactory.newInstance();
                    streamEventConverter.convertComplexEvent(complexEvent, newEvent);
                    String key = partitionExecutor.execute(newEvent);
                    send(key, newEvent);
                }
            } else {
                ComplexEventChunk<ComplexEvent> complexEventChunk = new ComplexEventChunk<ComplexEvent>();
                complexEventChunk.add(complexEvent);
                ComplexEventChunk<ComplexEvent> outputEventChunk = new ComplexEventChunk<ComplexEvent>();
                String currentKey = null;
                while (complexEventChunk.hasNext()) {
                    ComplexEvent aEvent = complexEventChunk.next();
                    complexEventChunk.remove();
                    StreamEvent newEvent = streamEventFactory.newInstance();
                    streamEventConverter.convertComplexEvent(aEvent, newEvent);
                    boolean currentEventMatchedPrevPartitionExecutor = false;
                    for (PartitionExecutor partitionExecutor : partitionExecutors) {
                        String key = partitionExecutor.execute(newEvent);
                        if (key != null) {
                            if (currentKey == null) {
                                currentKey = key;
                            } else if (!currentKey.equals(key)) {
                                if (!currentEventMatchedPrevPartitionExecutor) {
                                    ComplexEvent firstEvent = outputEventChunk.getFirst();
                                    send(currentKey, firstEvent);
                                    currentKey = key;
                                    outputEventChunk.clear();
                                } else {
                                    ComplexEvent firstEvent = outputEventChunk.getFirst();
                                    send(currentKey, firstEvent);
                                    currentKey = key;
                                    outputEventChunk.clear();
                                    StreamEvent cloneEvent = streamEventFactory.newInstance();
                                    streamEventConverter.convertComplexEvent(aEvent, cloneEvent);
                                    outputEventChunk.add(cloneEvent);
                                }
                            }
                            if (!currentEventMatchedPrevPartitionExecutor) {
                                outputEventChunk.add(newEvent);
                            }
                            currentEventMatchedPrevPartitionExecutor = true;
                        }
                    }
                }
                send(currentKey, outputEventChunk.getFirst());
                outputEventChunk.clear();
            }
        }

    }
Specific {@link StreamJunction.Receiver} implementation to pump events into partitions. This will send the event
to the matching partition.
modules/siddhi-core/src/main/java/io/siddhi/core/partition/PartitionStreamReceiver.java