akka/alpakka-kafka

Hi, I'm using alpakka for consuming data, process them and produce it back to kafka. I want a reliable s/w which process every piece of data. I thought committable consumer source of alpakka would provide me that, where I can commit once I process the data. As my flow is dynamic graph created, I need to split the data and merge them back in the same sequence. I am currently using partitions and merge sequence concepts of graph dsl, which solves my problem. But the materialized value, thats provided by commitable source could not be sent or captured as part of my flow creation. Can some one help me with that, please find the below code, I use java as my programming language:
`
Source<ConsumerMessage.CommittableMessage<String, String>, Consumer.Control> kafkaMessages = Consumer.committableSource( consumerSettings, Subscriptions.topics( topic ) );
Flow<Pair, Pair, NotUsed> someflow1 = constructFlow1();
Flow<Pair, Pair, NotUsed> someflow2 = constructFlow2();
Flow<Pair, Pair, NotUsed> someflow3 = constructFlow3();
Flow<Pair, Pair, NotUsed> combinedFlow = someflow1.async()
.via( someflow2.async() )
.via( someflow3.async() );
Sink<ConsumerMessage.Committable, CompletionStage<Done>> sink = Committer.sink(committerSettings);
final RunnableGraph<Consumer.Control> result =
RunnableGraph.<Consumer.Control>fromGraph(
GraphDSL.create(
builder -> {
UniformFanOutShape<Pair, Pair> partitions =
builder.add(
Partition.create( Pair.class, concurrency, element -> getPartitionNumber() ) );
UniformFanInShape<Pair, Pair> output =
builder.add( MergeSequence.create( concurrency, Pair::second ) );

                                builder
                                        .from( builder.add( kafkaMessages.zipWithIndex() ) )
                                        .viaFanOut( partitions );
                                for( int i = 0; i < concurrency; i++ ) {
                                    builder
                                            .from( partitions.out( i ) )
                                            .via( builder.add( combinedF.async() ) )
                                            .viaFanIn( output );
                                }

                                //Filter the data by removing non-null elements, and extract the producer record from the pair
                                builder
                                        .from(output.out())
                                        .to( builder.add( createProducerRecords ) )
                                        .via( Producer.flexiFlow( producerSettings ) )
                                        .map( m -> m.passThrough() )
                                        .toMat( Committer.sink( committerSettings ), Keep.both() )
                                        .mapMaterializedValue( Consumer::createDrainingControl );
                              ), sink );
                     return ClosedShape.getInstance();

} ) );
result.run( materializer );
`