### Question:
ProcessingPartition implements Closeable { public void maybeUnpause(long currentTime) { if (!paused) { LOGGER.debug("Partition [{}] not paused. Nothing to do", topicPartition); return; } if (currentTime >= pausedTillTime) { if(LOGGER.isInfoEnabled()){ LOGGER.info("Unpausing partition [{}] as the current time [{}] is >= paused time [{}]", new Object[] { topicPartition, new Date(currentTime), new Date(pausedTillTime) }); } consumer.resume(Collections.singleton(topicPartition)); PAUSED_PARTITIONS.dec(); paused = false; resetResults(); } else{ if (LOGGER.isDebugEnabled()) LOGGER.debug("Not unpausing partition [{}] as the current time [{}] is < paused time [{}]", topicPartition, currentTime, pausedTillTime); } } 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 maybeUnpause_notPaused() { partition.maybeUnpause(System.currentTimeMillis()); assertThat(partition.paused, is(false)); verify(consumer, never()).resume(Collections.singleton(topicPartition)); }
@Test public void maybeUnpause_paused() { long previousPausedPartitions = ProcessingPartition.PAUSED_PARTITIONS.count(); long currentTime = System.currentTimeMillis(); partition.paused = true; partition.recentFailedResults = config.getFailSampleSize(); partition.recentProcessingResults = new LinkedList<>(Collections.nCopies(config.getFailSampleSize(), false)); partition.pausedTillTime = currentTime + config.getFailPauseTime(); partition.maybeUnpause(System.currentTimeMillis()); assertThat(partition.paused, is(true)); verify(consumer, never()).resume(Collections.singleton(topicPartition)); partition.maybeUnpause(currentTime + config.getFailPauseTime()); assertThat(partition.paused, is(false)); assertThat(ProcessingPartition.PAUSED_PARTITIONS.count(), is(previousPausedPartitions - 1)); verify(consumer).resume(Collections.singleton(topicPartition)); assertThat(partition.recentFailedResults, is(0)); assertThat(partition.recentProcessingResults, is(new LinkedList<>(Collections.nCopies(config.getFailSampleSize(), true)))); }