KafkaConsumer implements Consumer<K, V> { @Override public void pause(Collection<TopicPartition> partitions) { acquire(); try { for (TopicPartition partition : partitions) { log.debug("Pausing partition {}", partition); subscriptions.pause(partition); } } finally { release(); } } KafkaConsumer(Map<String, Object> configs);  KafkaConsumer(Map<String, Object> configs,
        Deserializer<K> keyDeserializer,
        Deserializer<V> valueDeserializer);  KafkaConsumer(Properties properties);  KafkaConsumer(Properties properties,
        Deserializer<K> keyDeserializer,
        Deserializer<V> valueDeserializer); @SuppressWarnings("unchecked") private  KafkaConsumer(ConsumerConfig config,
        Deserializer<K> keyDeserializer,
        Deserializer<V> valueDeserializer);   KafkaConsumer(String clientId,
        ConsumerCoordinator coordinator,
        Deserializer<K> keyDeserializer,
        Deserializer<V> valueDeserializer,
        Fetcher<K, V> fetcher,
        ConsumerInterceptors<K, V> interceptors,
        Time time,
        ConsumerNetworkClient client,
        Metrics metrics,
        SubscriptionState subscriptions,
        Metadata metadata,
        long retryBackoffMs,
        long requestTimeoutMs); Set<TopicPartition> assignment(); Set<String> subscription(); @Override void subscribe(Collection<String> topics, ConsumerRebalanceListener listener); @Override void subscribe(Collection<String> topics); @Override void subscribe(Pattern pattern, ConsumerRebalanceListener listener); void unsubscribe(); @Override void assign(Collection<TopicPartition> partitions); @Override ConsumerRecords<K, V> poll(long timeout); @Override void commitSync(); @Override void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsets); @Override void commitAsync(); @Override void commitAsync(OffsetCommitCallback callback); @Override void commitAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback); @Override void seek(TopicPartition partition, long offset); void seekToBeginning(Collection<TopicPartition> partitions); void seekToEnd(Collection<TopicPartition> partitions); long position(TopicPartition partition); @Override OffsetAndMetadata committed(TopicPartition partition); @Override Map<MetricName, ? extends Metric> metrics(); @Override List<PartitionInfo> partitionsFor(String topic); @Override Map<String, List<PartitionInfo>> listTopics(); @Override void pause(Collection<TopicPartition> partitions); @Override void resume(Collection<TopicPartition> partitions); @Override Set<TopicPartition> paused(); @Override void close(); @Override void wakeup(); static void main(String[] args);  }
@Test public void testPause() { KafkaConsumer<byte[], byte[]> consumer = newConsumer(); consumer.assign(Collections.singletonList(tp0)); assertEquals(singleton(tp0), consumer.assignment()); assertTrue(consumer.paused().isEmpty()); consumer.pause(singleton(tp0)); assertEquals(singleton(tp0), consumer.paused()); consumer.resume(singleton(tp0)); assertTrue(consumer.paused().isEmpty()); consumer.unsubscribe(); assertTrue(consumer.paused().isEmpty()); consumer.close(); }