resilientGraph.java

Graph<ClosedShape, Consumer.Control> completionStageGraph = GraphDSL.create(source, (builder, sourceShape) -> {
        // Add Broadcast
        // Add Flows
        // Add Sinks

        builder
                .from(...)
                .viaFanOut(...)
                .via(...)
                .to(...);

        builder
                .from(...)
                .via(...)
                .to(...);

        return ClosedShape.getInstance();
});

// Run the stream and supervise
RunnableGraph.fromGraph(completionStageGraph)
        .run(materializer)
        .isShutdown()
        .whenComplete((done, throwable) -> {
            log.info("The underlying `KafkaConsumer` has been closed.");
            getSelf().tell(PoisonPill.getInstance(), getSelf());
        });