### Question:
ProcessingKafkaConsumer implements Closeable { public synchronized void commitOffsets() { if (pauseCommit) { LOGGER.debug("Commits are paused until we poll() again"); return; } LOGGER.debug("committing offsets"); try { commitOffsets(getCommittableOffsets()); } catch(CommitFailedException e) { LOGGER.debug("Failed to commit offsets, pausing commits until next poll", e); pauseCommit = true; throw e; } } ProcessingKafkaConsumer(ProcessingConfig config);  ProcessingKafkaConsumer(ProcessingConfig config, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer);   ProcessingKafkaConsumer(ProcessingConfig config, Consumer<K, V> consumer); Optional<ConsumerRecord<K, V>> nextRecord(final long timeout); boolean ack(ConsumerRecord<K, V> record); boolean ack(TopicPartition topicPartition, long offset); boolean fail(ConsumerRecord<K, V> record); boolean fail(TopicPartition topicPartition, long offset); synchronized void resetOffsets(Map<TopicPartition, Long> offsets); synchronized void commitOffsets(); void subscribe(Collection<String> topics); Consumer<K, V> getConsumer(); @Override void close();  }### Answer:
@Test public void commitOffsets() { long previousCommitCount = ProcessingKafkaConsumer.COMMIT_METER.count(); processingConsumer.nextRecord(POLL_TIME); processingConsumer.nextRecord(POLL_TIME); processingConsumer.nextRecord(POLL_TIME); processingConsumer.nextRecord(POLL_TIME); processingConsumer.nextRecord(POLL_TIME); processingConsumer.nextRecord(POLL_TIME); processingConsumer.nextRecord(POLL_TIME); processingConsumer.nextRecord(POLL_TIME); processingConsumer.ack(topicPartition, record1.offset()); processingConsumer.ack(topicPartition, record3.offset()); processingConsumer.ack(new TopicPartition(record5.topic(), record5.partition()), record5.offset()); processingConsumer.commitOffsets(); assertThat(ProcessingKafkaConsumer.COMMIT_METER.count(), is(previousCommitCount + 1)); Map<TopicPartition, OffsetAndMetadata> committedOffsets = new HashMap<>(); committedOffsets.put(topicPartition, new OffsetAndMetadata(record1.offset() + 1)); committedOffsets.put(new TopicPartition(record5.topic(), record5.partition()), new OffsetAndMetadata(record5.offset() + 1)); verify(consumer).commitSync(committedOffsets); assertThat(processingConsumer.getCommittableOffsets().isEmpty(), is(true)); }