### Question:
ProcessingPartition implements Closeable { @Override public void close() throws IOException { if (consumer.paused().contains(topicPartition)) PAUSED_PARTITIONS.dec(); } ProcessingPartition(TopicPartition topicPartition, ProcessingConfig config, Consumer<K, V> consumer); boolean ack(long offset); boolean fail(long offset); void maybeUnpause(long currentTime); OffsetAndMetadata getCommittableOffset(); long getCommittableOffsetsSize(); void committedOffset(long committedOffset); void load(List<ConsumerRecord<K, V>> records); ConsumerRecord<K, V> nextRecord(); boolean hasNextRecord(); TopicPartition getTopicPartition(); @Override void close();  }### Answer:
@Test public void pause_thenClosePartition() throws IOException { when(consumer.paused()).thenReturn(Collections.singleton(topicPartition)); long previousPausedPartitions = ProcessingPartition.PAUSED_PARTITIONS.count(); pause_thresholdMet(); assertThat(ProcessingPartition.PAUSED_PARTITIONS.count(), is(previousPausedPartitions + 1)); partition.close(); assertThat(ProcessingPartition.PAUSED_PARTITIONS.count(), is(previousPausedPartitions)); }