selective-pause.java

// assume consumer is subscribed to demo.topic.name && another.topic.name

String pausedTopic = "demo.topic.name";

//
// PAUSE
//

// filter out the assigned partitions for the topic being paused
Set<TopicPartition> partitionsToPause = consumer.assignment()
		.stream()
		.filter(tp -> pausedTopic.equals(tp.topic()))
		.collect(toSet());

Set<TopicPartition> pausedPartitions = consumer.pause(partitionsToPause);

//
// RESUME
//

// filter out the paused partitions for the topic being resumed
Set<TopicPartition> partitionsToResume = consumer.paused()
		.stream()
		.filter(tp -> pausedTopic.equals(tp.topic()))
		.collect(toSet());

consumer.resume(partitionsToResume);