### Question:
ProcessingPartition implements Closeable { public void committedOffset(long committedOffset) { LOGGER.debug("Offset [{}] has been committed for partition [{}]. Removing all completed offsets below commit value", committedOffset, topicPartition); lastCommittedOffset = committedOffset; committableOffset = null; completedOffsets.removeIf(o -> o < committedOffset); PARTITION_COMMITTED_OFFSETS.update(committedOffset - lastCommittedOffset); } 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 committedOffset() { long previousCommittedOffsets = ProcessingPartition.PARTITION_COMMITTED_OFFSETS.count(); partition.lastCommittedOffset = 0L; partition.committedOffset(123L); assertThat(ProcessingPartition.PARTITION_COMMITTED_OFFSETS.count(), is(previousCommittedOffsets + 1L)); }