### Question:
ProcessingPartition implements Closeable { public boolean ack(long offset) { Long messageReadTime = pendingOffsets.remove(offset); if (messageReadTime == null) { LOGGER.debug("Ack for record on topic partition [{}] with offset [{}] is invalid as that offset is not " + "pending. Not committing", topicPartition, offset); return false; } PROCESSING_LATENCY.update(System.currentTimeMillis() - messageReadTime); LOGGER.debug("Acking record with offset [{}] for partition [{}]", offset, topicPartition); completedOffsets.add(offset); maybeUpdateCommitableOffset(offset); addResult(true); return true; } 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 ack_recordNotPending() { long previousProcessingLatencyCount = ProcessingPartition.PROCESSING_LATENCY.count(); assertThat(partition.ack(1L), is(false)); assertThat(ProcessingPartition.PROCESSING_LATENCY.count(), is(previousProcessingLatencyCount)); }