Build the Partition Graph

buildGraph.java

// Build the Partition Graph
Graph<ClosedShape, Consumer.Control> completionStageGraph = GraphDSL.create(sourceStream, (builder, sourceShape) -> {

    UniformFanOutShape<EventMessage, EventMessage> eventTypeFanOut = builder.add(eventTypePartition);
    Outlet<EventMessage> unknownEventOutlet = eventTypeFanOut.out(0);
    Outlet<EventMessage> firstPartitionOutlet = eventTypeFanOut.out(1);
    Outlet<EventMessage> secondPartitionOutlet = eventTypeFanOut.out(2);

    builder.from(sourceShape)
            .toFanOut(eventTypeFanOut)
                .from(unknownEventOutlet)
                    .to(builder.add(commitOnlySink))
                .from(firstPartitionOutlet)
                    .via(builder.add(firstPartitionFlow))
                    .via(builder.add(produceMessageFlow))
                    .to(builder.add((producerCommitSink)))
                .from(secondPartitionOutlet)
                    .via(builder.add(secondPartitionFlow))
                    .to(builder.add(commitOnlySink));

    return ClosedShape.getInstance();
});