Pausing spark kafka streaming (direct) or exclude/include some partitions on the fly per batch

[sagarcasual .]

Hello, this is for
Pausing spark kafka streaming (direct) or exclude/include some partitions
on the fly per batch
=========================================================
I have following code that creates a direct stream using Kafka connector
for Spark.

final JavaInputDStream<KafkaMessage> msgRecords =
KafkaUtils.createDirectStream(
            jssc, String.class, String.class, StringDecoder.class,
StringDecoder.class,
            KafkaMessage.class, kafkaParams, topicsPartitions,
            message -> {
                return KafkaMessage.builder()
                        .
                        .build();
            }
    );

However I want to handle a situation, where I can decide that this
streaming needs to pause for a while on conditional basis, is there any way
to achieve this? Say my Kafka is undergoing some maintenance, so between
10AM to 12PM stop processing, and then again pick up at 12PM from the last
offset, how do I do it?

Also, assume all of a sudden we want to take one-or-more of the partitions
for a pull and add it back after some pulls, how do I achieve that?

-Regards
Sagar