### Question:
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 // 订阅指定的topic,并未消费者自动分配分区 void subscribe(Collection<String> topics, ConsumerRebalanceListener listener); @Override void subscribe(Collection<String> topics); @Override void subscribe(Pattern pattern, ConsumerRebalanceListener listener); void unsubscribe(); @Override // 用户手动订阅指定的topic并指定消费的分区，和subscribe方法互斥 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 // 恢复consumer void resume(Collection<TopicPartition> partitions); @Override // 暂停consumer Set<TopicPartition> paused(); @Override Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch); @Override Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> partitions); @Override Map<TopicPartition, Long> endOffsets(Collection<TopicPartition> partitions); @Override void close(); void close(long timeout, TimeUnit timeUnit); @Override void wakeup();  }### Answer:
@Test public void testPause() { KafkaConsumer<byte[], byte[]> consumer = newConsumer(); consumer.assign(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(); }